File size: 4,990 Bytes
3144483 | 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 110 111 112 113 114 115 116 117 118 | import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { createDeferred } from "../../test/helpers/promise.js";
import { createBackgroundWorkOwner, getBackgroundWorkSnapshot } from "./background-work.js";
import {
enqueueCommandInLane,
getCommandLaneSnapshot,
markGatewayDraining,
resetAllLanes,
} from "./command-queue.js";
import { resetCommandQueueStateForTest } from "./command-queue.test-support.js";
import { getGatewayRestartDrainSignal } from "./gateway-work-admission.js";
import { CommandLane } from "./lanes.js";
vi.mock("../logging/diagnostic-runtime.js", () => ({
logLaneEnqueue: vi.fn(),
logLaneDequeue: vi.fn(),
diagnosticLogger: { debug: vi.fn(), warn: vi.fn(), error: vi.fn() },
}));
describe("background work admission", () => {
beforeEach(resetCommandQueueStateForTest);
afterEach(resetCommandQueueStateForTest);
it("shares three slots, preserves owner widths and FIFO, and leaves foreground capacity free", async () => {
const parallel = createBackgroundWorkOwner({ owner: "plugin:parallel", maxConcurrent: 3 });
const serial = createBackgroundWorkOwner({ owner: "core:serial", maxConcurrent: 1 });
const gates = Array.from({ length: 3 }, () => createDeferred());
const parallelRuns = gates.map((gate) => parallel.enqueue(async () => await gate.promise));
const serialGate = createDeferred();
const order: number[] = [];
const first = serial.enqueue(async () => {
order.push(1);
await serialGate.promise;
});
const second = serial.enqueue(async () => {
order.push(2);
});
try {
expect(getBackgroundWorkSnapshot()).toMatchObject({ activeCount: 3, queuedCount: 2 });
await expect(enqueueCommandInLane(CommandLane.Main, async () => "foreground")).resolves.toBe(
"foreground",
);
gates[0]!.resolve();
await parallelRuns[0];
expect(order).toEqual([1]);
expect(getBackgroundWorkSnapshot()).toMatchObject({ activeCount: 3, queuedCount: 1 });
gates[1]!.resolve();
gates[2]!.resolve();
await Promise.all(parallelRuns);
expect(getCommandLaneSnapshot(serial.lane)).toMatchObject({ activeCount: 1, queuedCount: 1 });
serialGate.resolve();
await Promise.all([first, second]);
expect(order).toEqual([1, 2]);
} finally {
gates.forEach((gate) => gate.resolve());
serialGate.resolve();
await Promise.all([...parallelRuns, first, second]);
}
});
it("removes cancelled work immediately without invoking it or reordering its successors", async () => {
const owner = createBackgroundWorkOwner({ owner: "core:cancel", maxConcurrent: 1 });
const gate = createDeferred();
const active = owner.enqueue(async () => await gate.promise);
const controller = new AbortController();
const cancelledTask = vi.fn(async () => undefined);
const order: number[] = [];
const first = owner.enqueue(async () => {
order.push(1);
});
const cancelled = owner.enqueue(cancelledTask, { abortSignal: controller.signal });
const last = owner.enqueue(async () => {
order.push(2);
});
const rejection = expect(cancelled).rejects.toThrow("cancel background work");
controller.abort(new Error("cancel background work"));
await rejection;
expect(getBackgroundWorkSnapshot()).toMatchObject({ activeCount: 1, queuedCount: 2 });
gate.resolve();
await Promise.all([active, first, last]);
expect(cancelledTask).not.toHaveBeenCalled();
expect(order).toEqual([1, 2]);
});
it.each(["drain", "reset"])(
"cancels old work on restart %s and admits fresh work",
async (restart) => {
const owner = createBackgroundWorkOwner({ owner: "core:restart", maxConcurrent: 1 });
const started = createDeferred();
const active = owner.enqueue(async (signal) => {
started.resolve();
await new Promise<void>((resolve) => {
signal.addEventListener("abort", () => resolve(), { once: true });
});
signal.throwIfAborted();
});
await started.promise;
const staleTask = vi.fn(async () => undefined);
const stale = owner.enqueue(staleTask);
const oldSignal = getGatewayRestartDrainSignal();
const activeRejected = expect(active).rejects.toThrow(/draining for restart|runtime reset/u);
const staleRejected = expect(stale).rejects.toThrow(/draining for restart|runtime reset/u);
if (restart === "drain") {
markGatewayDraining();
} else {
resetAllLanes();
}
await Promise.all([activeRejected, staleRejected]);
resetAllLanes();
expect(oldSignal.aborted).toBe(true);
expect(getGatewayRestartDrainSignal().aborted).toBe(false);
await expect(owner.enqueue(async () => "fresh")).resolves.toBe("fresh");
expect(staleTask).not.toHaveBeenCalled();
expect(getBackgroundWorkSnapshot()).toMatchObject({ activeCount: 0, queuedCount: 0 });
},
);
});
|