Friendship-and-chat-api/script/migrate-db-files-to-oss/test/concurrency.test.mjs

import test from 'node:test';
import assert from 'node:assert/strict';
import { setImmediate } from 'node:timers/promises';
import { mapBounded } from '../lib/concurrency.mjs';

function deferred() {
  let resolve, reject;
  const promise = new Promise((res, rej) => { resolve = res; reject = rej; });
  return { promise, resolve, reject };
}

test('bounded workers keep input order when tasks finish out of order', { timeout: 2000 }, async () => {
  const gates = Array.from({ length: 5 }, deferred);
  const started = [];
  let active = 0, maximum = 0;
  const pending = mapBounded(['a', 'b', 'c', 'd', 'e'], 2, async (value, index) => {
    started.push(index);
    maximum = Math.max(maximum, ++active);
    try { await gates[index].promise; return value.toUpperCase(); }
    finally { active--; }
  });
  assert.deepEqual(started, [0, 1]);
  gates[1].resolve();
  await setImmediate();
  assert.deepEqual(started, [0, 1, 2]);
  gates[2].resolve();
  await setImmediate();
  assert.deepEqual(started, [0, 1, 2, 3]);
  gates[0].resolve();
  await setImmediate();
  assert.deepEqual(started, [0, 1, 2, 3, 4]);
  gates[4].resolve();
  gates[3].resolve();
  assert.deepEqual(await pending, ['A', 'B', 'C', 'D', 'E']);
  assert.equal(maximum, 2);
  assert.equal(active, 0);
});

test('failure stops new work and waits for all running workers before rejecting', { timeout: 2000 }, async () => {
  const gates = Array.from({ length: 5 }, deferred);
  const started = [];
  const finished = [];
  const failure = new Error('first verification failed');
  let settled = false;
  const pending = mapBounded([0, 1, 2, 3, 4], 3, async index => {
    started.push(index);
    try { await gates[index].promise; return index; }
    finally { finished.push(index); }
  });
  const outcome = pending.then(
    value => { settled = true; return { value }; },
    error => { settled = true; return { error }; },
  );
  assert.deepEqual(started, [0, 1, 2]);
  gates[1].reject(failure);
  await setImmediate();
  assert.deepEqual(started, [0, 1, 2]);
  assert.equal(settled, false);
  gates[0].resolve();
  await setImmediate();
  assert.deepEqual(started, [0, 1, 2]);
  assert.equal(settled, false);
  gates[2].resolve();
  assert.equal((await outcome).error, failure);
  assert.equal(settled, true);
  assert.deepEqual(finished.sort(), [0, 1, 2]);
});

test('additional in-flight rejections are drained while preserving the first failure', { timeout: 2000 }, async () => {
  const gates = Array.from({ length: 3 }, deferred);
  const first = new Error('first');
  const second = new Error('second');
  let calls = 0, settled = false;
  const outcome = mapBounded([0, 1, 2, 3], 3, async index => {
    calls++;
    await gates[index].promise;
    return index;
  }).then(
    () => { settled = true; assert.fail('must reject'); },
    error => { settled = true; return error; },
  );
  gates[0].reject(first);
  await setImmediate();
  gates[1].reject(second);
  await setImmediate();
  assert.equal(settled, false);
  assert.equal(calls, 3);
  gates[2].resolve();
  assert.equal(await outcome, first);
});

test('a synchronous failure stops scheduling immediately', async () => {
  const failure = new Error('synchronous failure');
  const started = [];
  await assert.rejects(mapBounded([0, 1, 2], 3, value => {
    started.push(value);
    throw failure;
  }), error => error === failure);
  assert.deepEqual(started, [0]);
});

test('falsy rejection values still reject and never get treated as success', async () => {
  for (const failure of [undefined, null, false, 0, '']) {
    let rejected = false;
    await mapBounded([1], 1, () => Promise.reject(failure)).then(
      () => assert.fail('must reject'),
      error => { rejected = true; assert.equal(error, failure); },
    );
    assert.equal(rejected, true);
  }
});

test('limit one executes serially and accepts synchronous return values', async () => {
  const order = [];
  assert.deepEqual(await mapBounded([2, 4, 6], 1, (value, index) => {
    order.push(index);
    return value * 2;
  }), [4, 8, 12]);
  assert.deepEqual(order, [0, 1, 2]);
});

test('limits larger than the item count only start existing items', async () => {
  let calls = 0;
  const result = await mapBounded(['x', 'y'], 1000, async value => { calls++; return value; });
  assert.deepEqual(result, ['x', 'y']);
  assert.equal(calls, 2);
});

test('empty input returns an empty array without invoking the worker', async () => {
  assert.deepEqual(await mapBounded([], 4, () => assert.fail('must not run')), []);
});

test('invalid limits reject even for an empty input', async () => {
  for (const limit of [0, -1, 1.5, Infinity, NaN, '4', null, undefined, Number.MAX_SAFE_INTEGER + 1]) {
    await assert.rejects(mapBounded([], limit, () => {}), /正整数/);
  }
});

test('input values are snapshotted before workers mutate the original collection', async () => {
  const input = ['first', 'second', 'third'];
  const result = await mapBounded(input, 1, async (value, index) => {
    if (index === 0) input.splice(1, 2, 'changed');
    return value;
  });
  assert.deepEqual(result, ['first', 'second', 'third']);
});