import { describe, expect, it } from "vitest"; import { ABORTED, isAborted, mapWithConcurrency } from "../pool.ts"; /** A manually-resolvable promise, for deterministic ordering control. */ function deferred(): { promise: Promise; resolve: (value: T) => void; } { let resolve!: (value: T) => void; const promise = new Promise((r) => { resolve = r; }); return { promise, resolve }; } /** Let queued microtasks (and a macrotask) drain. */ function flush(): Promise { return new Promise((r) => setTimeout(r, 0)); } describe("mapWithConcurrency", () => { it("(a) returns results in INPUT order even when later items resolve first", async () => { const items = [0, 1, 2]; const ds = items.map(() => deferred()); const seenIndexes: number[] = []; const p = mapWithConcurrency(items, 3, async (item, index) => { seenIndexes.push(index); return ds[item].promise; }); // Resolve in reverse order: item 2 first, then 1, then 0. ds[2].resolve("r2"); ds[1].resolve("r1"); ds[0].resolve("r0"); const results = await p; expect(results).toEqual(["r0", "r1", "r2"]); // fn receives the positional index. expect(seenIndexes.sort()).toEqual([0, 1, 2]); }); it("(b) never exceeds `limit` in flight (10 items, limit 3)", async () => { const items = Array.from({ length: 10 }, (_, i) => i); let inFlight = 0; let maxInFlight = 0; const results = await mapWithConcurrency(items, 3, async (item) => { inFlight++; maxInFlight = Math.max(maxInFlight, inFlight); await new Promise((r) => setTimeout(r, 1)); inFlight--; return item * 2; }); expect(maxInFlight).toBeLessThanOrEqual(3); // Prove it actually parallelised up to the cap (not serial). expect(maxInFlight).toBe(3); expect(results).toEqual(items.map((x) => x * 2)); }); it("(c) limit === items.length runs all items concurrently", async () => { const items = [0, 1, 2, 3, 4]; let started = 0; let release!: () => void; const allStarted = new Promise((r) => { release = r; }); // Each item blocks until ALL have started. This only resolves if the // pool truly runs every item concurrently; a lower cap would deadlock // and the test would hit the vitest timeout. const results = await mapWithConcurrency( items, items.length, async (item) => { started++; if (started === items.length) release(); await allStarted; return item * 10; }, ); expect(started).toBe(items.length); expect(results).toEqual([0, 10, 20, 30, 40]); }); it("(c') empty input resolves to an empty array", async () => { const calls: number[] = []; const results = await mapWithConcurrency([], 3, async (item: number) => { calls.push(item); return item; }); expect(results).toEqual([]); expect(calls).toEqual([]); }); it("(d) an already-aborted signal launches NOTHING and resolves to all abort markers (no reject)", async () => { const items = [0, 1, 2, 3]; const controller = new AbortController(); controller.abort(); let callCount = 0; const results = await mapWithConcurrency( items, 2, async (item) => { callCount++; return item; }, controller.signal, ); // Resolves (did not reject) with one marker slot per input item. expect(callCount).toBe(0); expect(results).toHaveLength(items.length); expect(results.every((r) => isAborted(r))).toBe(true); expect(results).toEqual([ABORTED, ABORTED, ABORTED, ABORTED]); }); it("(e) aborting mid-run keeps completed results in place and marks the un-run tail (no reject)", async () => { const items = [0, 1, 2, 3, 4, 5]; const controller = new AbortController(); const ds = items.map(() => deferred()); const launched: number[] = []; const p = mapWithConcurrency( items, 2, async (item) => { launched.push(item); return ds[item].promise; }, controller.signal, ); // With limit 2, exactly items 0 and 1 are launched synchronously. await flush(); expect(launched.slice().sort()).toEqual([0, 1]); // Abort BEFORE the in-flight items complete, then let them finish. controller.abort(); ds[0].resolve("r0"); ds[1].resolve("r1"); const results = await p; // Already-launched items keep their REAL results. expect(results[0]).toBe("r0"); expect(results[1]).toBe("r1"); // The un-run tail carries abort markers, positionally. expect(isAborted(results[2])).toBe(true); expect(isAborted(results[3])).toBe(true); expect(isAborted(results[4])).toBe(true); expect(isAborted(results[5])).toBe(true); expect(results).toHaveLength(items.length); // Discriminating assertion: the tail was genuinely never launched // (a buggy impl that launches all then overwrites would fail here). expect(launched.slice().sort()).toEqual([0, 1]); }); it("isAborted only matches the abort marker, not real fn results", () => { expect(isAborted(ABORTED)).toBe(true); expect(isAborted(null)).toBe(false); expect(isAborted(undefined)).toBe(false); expect(isAborted({ ok: false, value: null, reason: "x" })).toBe(false); expect(isAborted({ aborted: false })).toBe(false); }); });