102 lines
2.8 KiB
TypeScript
102 lines
2.8 KiB
TypeScript
import assert from "node:assert/strict";
|
|
import { describe, it } from "node:test";
|
|
import {
|
|
ALERT_SEARCH_USER_CONCURRENCY,
|
|
forEachWithConcurrency,
|
|
} from "../../src/util/concurrency.js";
|
|
|
|
/** Resolves after `ms`, long enough for the pool to fill before any task ends. */
|
|
const pause = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));
|
|
|
|
describe("forEachWithConcurrency", () => {
|
|
it("visits every item exactly once", async () => {
|
|
const items = Array.from({ length: 50 }, (_, i) => i);
|
|
const seen: number[] = [];
|
|
await forEachWithConcurrency(items, 7, async (item) => {
|
|
seen.push(item);
|
|
});
|
|
assert.deepEqual([...seen].sort((a, b) => a - b), items);
|
|
});
|
|
|
|
it("never exceeds the limit", async () => {
|
|
let inFlight = 0;
|
|
let peak = 0;
|
|
await forEachWithConcurrency(
|
|
Array.from({ length: 30 }, (_, i) => i),
|
|
4,
|
|
async () => {
|
|
inFlight += 1;
|
|
peak = Math.max(peak, inFlight);
|
|
await pause(3);
|
|
inFlight -= 1;
|
|
}
|
|
);
|
|
assert.equal(peak, 4);
|
|
});
|
|
|
|
it("reaches the limit rather than trickling one at a time", async () => {
|
|
let peak = 0;
|
|
let inFlight = 0;
|
|
await forEachWithConcurrency([1, 2, 3, 4, 5, 6], 6, async () => {
|
|
inFlight += 1;
|
|
peak = Math.max(peak, inFlight);
|
|
await pause(3);
|
|
inFlight -= 1;
|
|
});
|
|
assert.equal(peak, 6);
|
|
});
|
|
|
|
it("runs serially at a limit of one, in order", async () => {
|
|
const order: number[] = [];
|
|
await forEachWithConcurrency([1, 2, 3], 1, async (item) => {
|
|
await pause(1);
|
|
order.push(item);
|
|
});
|
|
assert.deepEqual(order, [1, 2, 3]);
|
|
});
|
|
|
|
it("never starts more workers than there are items", async () => {
|
|
let peak = 0;
|
|
let inFlight = 0;
|
|
await forEachWithConcurrency([1, 2], 100, async () => {
|
|
inFlight += 1;
|
|
peak = Math.max(peak, inFlight);
|
|
await pause(3);
|
|
inFlight -= 1;
|
|
});
|
|
assert.equal(peak, 2);
|
|
});
|
|
|
|
it("handles an empty list and a nonsense limit", async () => {
|
|
let calls = 0;
|
|
await forEachWithConcurrency([], 8, async () => {
|
|
calls += 1;
|
|
});
|
|
assert.equal(calls, 0);
|
|
|
|
const seen: number[] = [];
|
|
for (const limit of [0, -5, 0.5]) {
|
|
seen.length = 0;
|
|
await forEachWithConcurrency([1, 2, 3], limit, async (item) => {
|
|
seen.push(item);
|
|
});
|
|
assert.deepEqual(seen, [1, 2, 3], `limit ${limit}`);
|
|
}
|
|
});
|
|
|
|
it("propagates a throw, since callers own their own errors", async () => {
|
|
await assert.rejects(
|
|
() =>
|
|
forEachWithConcurrency([1, 2, 3], 2, async (item) => {
|
|
if (item === 2) throw new Error("boom");
|
|
}),
|
|
/boom/
|
|
);
|
|
});
|
|
|
|
it("ships a conservative default", () => {
|
|
assert.ok(ALERT_SEARCH_USER_CONCURRENCY >= 1);
|
|
assert.ok(ALERT_SEARCH_USER_CONCURRENCY <= 32);
|
|
});
|
|
});
|