File size: 5,562 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 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 | import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { createDeferred } from "../../test/helpers/promise.js";
import { enqueueCommandInLane, getCommandLaneSnapshot } from "./command-queue.js";
import { resetCommandQueueStateForTest } from "./command-queue.test-support.js";
import type { CommandQueueTaskDeadline } from "./command-queue.types.js";
const lane = "runtime-deadline-test";
const finishers: Array<() => void> = [];
function enqueueOwnedTask(initialDeadline: CommandQueueTaskDeadline, initiallyAborted = false) {
const finish = createDeferred();
finishers.push(finish.resolve);
const abort = new AbortController();
const release = new AbortController();
if (initiallyAborted) {
abort.abort();
}
const unsubscribe = vi.fn();
let progressAtMs = Date.now();
let publish: (deadline: CommandQueueTaskDeadline | undefined) => void = () => {
throw new Error("deadline subscription is not active");
};
const task = enqueueCommandInLane(lane, () => finish.promise, {
taskTimeoutMs: 25,
taskTimeoutProgressAtMs: () => progressAtMs,
taskTimeoutAbortSignal: abort.signal,
taskTimeoutAbortGraceMs: 5,
taskTimeoutReleaseSignal: release.signal,
taskTimeoutSubscribe: (onDeadline) => {
publish = onDeadline;
onDeadline(initialDeadline);
return unsubscribe;
},
});
const outcome = task.then(
() => ({ status: "completed" as const }),
(error: unknown) => ({ status: "failed" as const, error }),
);
return {
abort,
release,
finish: finish.resolve,
outcome,
unsubscribe,
publish: (deadline: CommandQueueTaskDeadline | undefined) => publish(deadline),
progress: () => {
progressAtMs = Date.now();
},
};
}
beforeEach(() => {
resetCommandQueueStateForTest();
vi.useFakeTimers();
vi.setSystemTime(Date.parse("2026-08-20T12:00:00Z"));
});
afterEach(async () => {
for (const finish of finishers.splice(0)) {
finish();
}
await vi.advanceTimersByTimeAsync(0);
resetCommandQueueStateForTest();
vi.useRealTimers();
});
describe("command lane owner deadlines", () => {
it("replaces idle timing with an absolute deadline that progress cannot extend", async () => {
const owner = enqueueOwnedTask({ kind: "bounded", deadlineAtMs: Date.now() + 100 });
const next = vi.fn(async () => "next");
const queued = enqueueCommandInLane(lane, next);
await vi.advanceTimersByTimeAsync(50);
owner.progress();
await vi.advanceTimersByTimeAsync(49);
expect(next).not.toHaveBeenCalled();
expect(getCommandLaneSnapshot(lane)).toMatchObject({ activeCount: 1, queuedCount: 1 });
await vi.advanceTimersByTimeAsync(1);
await expect(owner.outcome).resolves.toMatchObject({
status: "failed",
error: {
name: "CommandLaneTaskTimeoutError",
message: expect.stringContaining("owner deadline"),
},
});
await expect(queued).resolves.toBe("next");
expect(owner.unsubscribe).toHaveBeenCalledOnce();
});
it("lets unlimited execution hand off to bounded terminal settlement", async () => {
const owner = enqueueOwnedTask({ kind: "unlimited" });
await vi.advanceTimersByTimeAsync(49 * 60 * 60 * 1000);
expect(getCommandLaneSnapshot(lane).activeCount).toBe(1);
owner.publish({ kind: "bounded", deadlineAtMs: Date.now() + 120_000 });
await vi.advanceTimersByTimeAsync(119_999);
expect(getCommandLaneSnapshot(lane).activeCount).toBe(1);
owner.finish();
await expect(owner.outcome).resolves.toEqual({ status: "completed" });
owner.publish({ kind: "bounded", deadlineAtMs: Date.now() });
await vi.advanceTimersByTimeAsync(1);
expect(getCommandLaneSnapshot(lane).activeCount).toBe(0);
expect(vi.getTimerCount()).toBe(0);
});
it("restores ordinary idle recovery after the runtime releases deadline ownership", async () => {
const owner = enqueueOwnedTask({ kind: "unlimited" });
await vi.advanceTimersByTimeAsync(1_000);
owner.progress();
owner.publish(undefined);
await vi.advanceTimersByTimeAsync(24);
expect(getCommandLaneSnapshot(lane).activeCount).toBe(1);
await vi.advanceTimersByTimeAsync(1);
await expect(owner.outcome).resolves.toMatchObject({
status: "failed",
error: {
name: "CommandLaneTaskTimeoutError",
message: expect.stringContaining("no progress for 25ms"),
},
});
});
it("does not let a late deadline update replace an accepted abort grace", async () => {
const owner = enqueueOwnedTask({ kind: "unlimited" });
owner.abort.abort();
owner.publish({ kind: "unlimited" });
await vi.advanceTimersByTimeAsync(4);
expect(getCommandLaneSnapshot(lane).activeCount).toBe(1);
await vi.advanceTimersByTimeAsync(1);
await expect(owner.outcome).resolves.toMatchObject({
status: "failed",
error: {
name: "CommandLaneTaskTimeoutError",
message: expect.stringContaining("abort grace 5ms"),
},
});
});
it.each([false, true])(
"honors immediate release when initially aborted=%s",
async (initiallyAborted) => {
const owner = enqueueOwnedTask({ kind: "unlimited" }, initiallyAborted);
owner.release.abort();
await expect(owner.outcome).resolves.toMatchObject({
status: "failed",
error: {
name: "CommandLaneTaskTimeoutError",
message: expect.stringContaining("lane release requested"),
},
});
expect(getCommandLaneSnapshot(lane).activeCount).toBe(0);
},
);
});
|