File size: 2,892 Bytes
fc93158 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 | import { describe, expect, it, vi } from "vitest";
import { enqueueKeyedTask, KeyedAsyncQueue } from "./keyed-async-queue.js";
function deferred<T>() {
let resolve!: (value: T | PromiseLike<T>) => void;
let reject!: (reason?: unknown) => void;
const promise = new Promise<T>((res, rej) => {
resolve = res;
reject = rej;
});
return { promise, resolve, reject };
}
describe("enqueueKeyedTask", () => {
it("serializes tasks per key and keeps different keys independent", async () => {
const tails = new Map<string, Promise<void>>();
const gate = deferred<void>();
const order: string[] = [];
const first = enqueueKeyedTask({
tails,
key: "a",
task: async () => {
order.push("a1:start");
await gate.promise;
order.push("a1:end");
},
});
const second = enqueueKeyedTask({
tails,
key: "a",
task: async () => {
order.push("a2:start");
order.push("a2:end");
},
});
const third = enqueueKeyedTask({
tails,
key: "b",
task: async () => {
order.push("b1:start");
order.push("b1:end");
},
});
await vi.waitFor(() => {
expect(order).toContain("a1:start");
expect(order).toContain("b1:start");
});
expect(order).not.toContain("a2:start");
gate.resolve();
await Promise.all([first, second, third]);
expect(order).toEqual(["a1:start", "b1:start", "b1:end", "a1:end", "a2:start", "a2:end"]);
expect(tails.size).toBe(0);
});
it("keeps queue alive after task failures", async () => {
const tails = new Map<string, Promise<void>>();
await expect(
enqueueKeyedTask({
tails,
key: "a",
task: async () => {
throw new Error("boom");
},
}),
).rejects.toThrow("boom");
await expect(
enqueueKeyedTask({
tails,
key: "a",
task: async () => "ok",
}),
).resolves.toBe("ok");
});
it("runs enqueue/settle hooks once per task", async () => {
const tails = new Map<string, Promise<void>>();
const onEnqueue = vi.fn();
const onSettle = vi.fn();
await enqueueKeyedTask({
tails,
key: "a",
task: async () => undefined,
hooks: { onEnqueue, onSettle },
});
expect(onEnqueue).toHaveBeenCalledTimes(1);
expect(onSettle).toHaveBeenCalledTimes(1);
});
});
describe("KeyedAsyncQueue", () => {
it("exposes tail map for observability", async () => {
const queue = new KeyedAsyncQueue();
const gate = deferred<void>();
const run = queue.enqueue("actor", async () => {
await gate.promise;
return 1;
});
expect(queue.getTailMapForTesting().has("actor")).toBe(true);
gate.resolve();
await run;
await Promise.resolve();
expect(queue.getTailMapForTesting().has("actor")).toBe(false);
});
});
|