24bd34f886
Add ChannelQueue in src/runtime/channel-queue.ts. Router enqueues each message so same-channel requests run serially. Different channels remain parallel. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
100 lines
2.9 KiB
TypeScript
100 lines
2.9 KiB
TypeScript
import { describe, it, expect, vi } from "vitest";
|
|
import { ChannelQueue } from "../../src/runtime/channel-queue";
|
|
|
|
/**
|
|
* Returns a task that resolves after `ms` milliseconds and pushes its label
|
|
* into the shared `order` array.
|
|
*/
|
|
function makeTask(order: string[], label: string, ms = 0): () => Promise<void> {
|
|
return () =>
|
|
new Promise<void>((resolve) => {
|
|
setTimeout(() => {
|
|
order.push(label);
|
|
resolve();
|
|
}, ms);
|
|
});
|
|
}
|
|
|
|
describe("ChannelQueue", () => {
|
|
it("(a) executes tasks in FIFO order within the same channel", async () => {
|
|
const queue = new ChannelQueue();
|
|
const order: string[] = [];
|
|
|
|
const p1 = queue.enqueue("ch1", makeTask(order, "first"));
|
|
const p2 = queue.enqueue("ch1", makeTask(order, "second"));
|
|
const p3 = queue.enqueue("ch1", makeTask(order, "third"));
|
|
|
|
await Promise.all([p1, p2, p3]);
|
|
|
|
expect(order).toEqual(["first", "second", "third"]);
|
|
});
|
|
|
|
it("(b) a failing task does not block the next task", async () => {
|
|
const queue = new ChannelQueue();
|
|
const order: string[] = [];
|
|
|
|
const failing: () => Promise<void> = () =>
|
|
Promise.reject(new Error("intentional failure"));
|
|
|
|
// Suppress the console.error output from ChannelQueue internals
|
|
const spy = vi.spyOn(console, "error").mockImplementation(() => {});
|
|
|
|
const p1 = queue.enqueue("ch1", failing);
|
|
const p2 = queue.enqueue("ch1", makeTask(order, "after-error"));
|
|
|
|
await Promise.all([p1, p2]);
|
|
|
|
spy.mockRestore();
|
|
|
|
expect(order).toEqual(["after-error"]);
|
|
});
|
|
|
|
it("(c) map entry is removed after all tasks complete", async () => {
|
|
const queue = new ChannelQueue();
|
|
|
|
const p1 = queue.enqueue("ch1", () => Promise.resolve());
|
|
const p2 = queue.enqueue("ch1", () => Promise.resolve());
|
|
|
|
await Promise.all([p1, p2]);
|
|
|
|
// Allow the finally-cleanup microtask to run
|
|
await Promise.resolve();
|
|
|
|
expect((queue as unknown as { queues: Map<string, unknown> }).queues.size).toBe(0);
|
|
});
|
|
|
|
it("(d) different channels run in parallel, not serialized", async () => {
|
|
const queue = new ChannelQueue();
|
|
const completionOrder: string[] = [];
|
|
|
|
// ch-slow gets a 50 ms task; ch-fast gets a 10 ms task.
|
|
// If channels were serialized (ch-slow first), ch-fast would finish after
|
|
// ch-slow. With true parallelism ch-fast finishes first.
|
|
const pSlow = queue.enqueue(
|
|
"ch-slow",
|
|
() =>
|
|
new Promise<void>((resolve) => {
|
|
setTimeout(() => {
|
|
completionOrder.push("slow");
|
|
resolve();
|
|
}, 50);
|
|
})
|
|
);
|
|
|
|
const pFast = queue.enqueue(
|
|
"ch-fast",
|
|
() =>
|
|
new Promise<void>((resolve) => {
|
|
setTimeout(() => {
|
|
completionOrder.push("fast");
|
|
resolve();
|
|
}, 10);
|
|
})
|
|
);
|
|
|
|
await Promise.all([pSlow, pFast]);
|
|
|
|
// fast must finish before slow
|
|
expect(completionOrder).toEqual(["fast", "slow"]);
|
|
});
|
|
});
|