fix: settle a shared abort operation by observation, not by drain count
The primitive decided a raced outcome by draining a hard-coded four microtasks and then asking whether the task had landed. That made the answer depend on scheduling rather than on what was observed: a caller abort could fix the terminal owner synchronously and a rejection later in the same call stack still won the public result, so `race()` disagreed with `terminal()` and the failure taxonomy a caller received depended on microtask ordering. Task settlement and the terminal event now share one settle-once state machine. Whichever callback actually runs first owns the outcome; a value that loses is compensated exactly once and a rejection that loses is absorbed, so neither can surface late. The three consumers that kept their own copies of these mechanics move onto it. The Image probe and the Resumable fetch transport attached their caller listener before installing the timer, so a scheduler that threw rejected the public `probe()`/`execute()` promise natively and left the listener on the caller's signal; both now close atomically inside their own Result vocabulary and start no fetch. `snapshotAbortTimers` binds the scheduler callables once at construction, so replacing a method after composition can no longer change how work already in flight is bounded. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
DongHyeonka
co-authored by
Claude Opus 5
Claude Opus 5
parent
8d6d84bfcc
commit
cc91fc6ae0
@@ -198,6 +198,165 @@ describe("shared abortable operation mechanics", () => {
|
||||
).resolves.toEqual({ kind: "TERMINAL", terminal: "CLOSED" });
|
||||
});
|
||||
|
||||
/**
|
||||
* X-AUDIT-01. The outcome must be decided by which callback was actually
|
||||
* observed first, not by how many microtasks the primitive happens to drain
|
||||
* before it gives up waiting. A terminal owner fixed synchronously owns the
|
||||
* result even when the task rejects later in the same call stack.
|
||||
*/
|
||||
describe("settle-once ownership across callback orderings", () => {
|
||||
const terminalOwners = [
|
||||
{
|
||||
label: "caller abort",
|
||||
owner: "CALLER_ABORT" as const,
|
||||
trigger: (caller: AbortController, operation: { close(): void }, timers: Array<() => void>) => {
|
||||
void operation;
|
||||
void timers;
|
||||
caller.abort();
|
||||
},
|
||||
},
|
||||
{
|
||||
label: "deadline",
|
||||
owner: "DEADLINE" as const,
|
||||
trigger: (
|
||||
caller: AbortController,
|
||||
operation: { close(): void },
|
||||
timers: Array<() => void>,
|
||||
) => {
|
||||
void caller;
|
||||
void operation;
|
||||
timers.forEach((callback) => callback());
|
||||
},
|
||||
},
|
||||
{
|
||||
label: "close",
|
||||
owner: "CLOSED" as const,
|
||||
trigger: (
|
||||
caller: AbortController,
|
||||
operation: { close(): void },
|
||||
timers: Array<() => void>,
|
||||
) => {
|
||||
void caller;
|
||||
void timers;
|
||||
operation.close();
|
||||
},
|
||||
},
|
||||
];
|
||||
|
||||
const createHarness = () => {
|
||||
const caller = new AbortController();
|
||||
const timers: Array<() => void> = [];
|
||||
const operation = createAbortableOperation({
|
||||
signal: caller.signal,
|
||||
timeoutMs: 1_000,
|
||||
setTimer: (callback) => {
|
||||
timers.push(callback);
|
||||
return timers.length;
|
||||
},
|
||||
clearTimer: () => {},
|
||||
});
|
||||
return { caller, timers, operation };
|
||||
};
|
||||
|
||||
for (const { label, owner, trigger } of terminalOwners) {
|
||||
it(`keeps ${label} as the owner when the task rejects in the same call stack`, async () => {
|
||||
const { caller, timers, operation } = createHarness();
|
||||
let reject: ((reason: unknown) => void) | undefined;
|
||||
const task = new Promise<never>((_resolve, rejectTask) => {
|
||||
reject = rejectTask;
|
||||
});
|
||||
const raced = operation.race(task);
|
||||
|
||||
trigger(caller, operation, timers);
|
||||
expect(operation.terminal()).toBe(owner);
|
||||
reject?.(new Error("rejected after the terminal owner was fixed"));
|
||||
|
||||
await expect(raced).resolves.toEqual({ kind: "TERMINAL", terminal: owner });
|
||||
expect(operation.terminal()).toBe(owner);
|
||||
});
|
||||
|
||||
it(`keeps ${label} as the owner regardless of how many microtasks drain`, async () => {
|
||||
for (const drains of [0, 1, 2, 3, 5, 9]) {
|
||||
const { caller, timers, operation } = createHarness();
|
||||
let reject: ((reason: unknown) => void) | undefined;
|
||||
const task = new Promise<never>((_resolve, rejectTask) => {
|
||||
reject = rejectTask;
|
||||
});
|
||||
const raced = operation.race(task);
|
||||
|
||||
trigger(caller, operation, timers);
|
||||
for (let turn = 0; turn < drains; turn += 1) {
|
||||
await Promise.resolve();
|
||||
}
|
||||
reject?.(new Error("late rejection"));
|
||||
|
||||
await expect(raced).resolves.toEqual({
|
||||
kind: "TERMINAL",
|
||||
terminal: owner,
|
||||
});
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
it("keeps a rejection that was observed before any terminal owner", async () => {
|
||||
const { caller, operation } = createHarness();
|
||||
const reason = new Error("task failed first");
|
||||
const raced = operation.race(Promise.reject(reason));
|
||||
|
||||
// Let the rejection callback actually run before the abort is requested.
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
caller.abort();
|
||||
|
||||
await expect(raced).resolves.toEqual({ kind: "REJECTED", reason });
|
||||
});
|
||||
|
||||
it("keeps a value that was observed before any terminal owner", async () => {
|
||||
const { caller, operation } = createHarness();
|
||||
const compensated: string[] = [];
|
||||
const raced = operation.race(Promise.resolve("early"), (value) =>
|
||||
compensated.push(value),
|
||||
);
|
||||
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
caller.abort();
|
||||
|
||||
await expect(raced).resolves.toEqual({ kind: "VALUE", value: "early" });
|
||||
expect(compensated).toEqual([]);
|
||||
});
|
||||
|
||||
it("never compensates or re-raises a rejection that lands after the owner", async () => {
|
||||
const unhandled: unknown[] = [];
|
||||
const onUnhandled = (reason: unknown) => unhandled.push(reason);
|
||||
process.on("unhandledRejection", onUnhandled);
|
||||
try {
|
||||
const { caller, operation } = createHarness();
|
||||
const compensated: unknown[] = [];
|
||||
let reject: ((reason: unknown) => void) | undefined;
|
||||
const task = new Promise<never>((_resolve, rejectTask) => {
|
||||
reject = rejectTask;
|
||||
});
|
||||
const raced = operation.race(task, (value) => compensated.push(value));
|
||||
|
||||
caller.abort();
|
||||
reject?.(new Error("late rejection"));
|
||||
await expect(raced).resolves.toEqual({
|
||||
kind: "TERMINAL",
|
||||
terminal: "CALLER_ABORT",
|
||||
});
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
await new Promise((resolve) => setTimeout(resolve, 0));
|
||||
|
||||
expect(compensated).toEqual([]);
|
||||
expect(unhandled).toEqual([]);
|
||||
} finally {
|
||||
process.off("unhandledRejection", onUnhandled);
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
it("compensates a late native handle without changing the outcome", async () => {
|
||||
const cancel = vi.fn(async () => {});
|
||||
compensateLateHandle(Promise.resolve({ body: { cancel } }));
|
||||
|
||||
@@ -1936,6 +1936,115 @@ describe("browser image probe", () => {
|
||||
expect(close).toHaveBeenCalledOnce();
|
||||
});
|
||||
});
|
||||
|
||||
/**
|
||||
* X-AUDIT-02. `probe()` promises a `BrowserDataResult`. A scheduler that
|
||||
* cannot install the probe deadline must close the probe inside that contract
|
||||
* rather than rejecting it, and must not leave the caller's listener behind.
|
||||
*/
|
||||
describe("scheduler boundary", () => {
|
||||
const trackedSignal = () => {
|
||||
const controller = new AbortController();
|
||||
const added: string[] = [];
|
||||
const removed: string[] = [];
|
||||
const add = controller.signal.addEventListener.bind(controller.signal);
|
||||
const remove = controller.signal.removeEventListener.bind(
|
||||
controller.signal,
|
||||
);
|
||||
Object.defineProperty(controller.signal, "addEventListener", {
|
||||
configurable: true,
|
||||
value: (type: string, ...rest: readonly unknown[]) => {
|
||||
added.push(type);
|
||||
return (add as (...args: readonly unknown[]) => unknown)(
|
||||
type,
|
||||
...rest,
|
||||
);
|
||||
},
|
||||
});
|
||||
Object.defineProperty(controller.signal, "removeEventListener", {
|
||||
configurable: true,
|
||||
value: (type: string, ...rest: readonly unknown[]) => {
|
||||
removed.push(type);
|
||||
return (remove as (...args: readonly unknown[]) => unknown)(
|
||||
type,
|
||||
...rest,
|
||||
);
|
||||
},
|
||||
});
|
||||
return { controller, added, removed };
|
||||
};
|
||||
|
||||
it("closes the probe when the scheduler cannot install the deadline", async () => {
|
||||
const fetcher = vi.fn(async () => new Response(null, { status: 200 }));
|
||||
const { controller, added, removed } = trackedSignal();
|
||||
const probe = createBrowserImageProbe({
|
||||
fetcher: fetcher as unknown as typeof fetch,
|
||||
createBitmap: vi.fn(),
|
||||
timeoutMs: 1_000,
|
||||
scheduler: {
|
||||
setTimeout: () => {
|
||||
throw new TypeError("image scheduler install exploded");
|
||||
},
|
||||
clearTimeout: vi.fn(),
|
||||
},
|
||||
});
|
||||
|
||||
await expect(
|
||||
probe.probe({ ...request(), signal: controller.signal }),
|
||||
).resolves.toMatchObject({
|
||||
ok: false,
|
||||
error: { code: "UNAVAILABLE", retryable: true, recovery: "RETRY" },
|
||||
});
|
||||
expect(fetcher).not.toHaveBeenCalled();
|
||||
expect(added.filter((type) => type === "abort")).toHaveLength(1);
|
||||
expect(removed.filter((type) => type === "abort")).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("starts no timer and no fetch for an already aborted caller", async () => {
|
||||
const fetcher = vi.fn(async () => new Response(null, { status: 200 }));
|
||||
const setTimeout_ = vi.fn(() => 1);
|
||||
const controller = new AbortController();
|
||||
controller.abort();
|
||||
const probe = createBrowserImageProbe({
|
||||
fetcher: fetcher as unknown as typeof fetch,
|
||||
createBitmap: vi.fn(),
|
||||
timeoutMs: 1_000,
|
||||
scheduler: { setTimeout: setTimeout_, clearTimeout: vi.fn() },
|
||||
});
|
||||
|
||||
await expect(
|
||||
probe.probe({ ...request(), signal: controller.signal }),
|
||||
).resolves.toMatchObject({ ok: false, error: { code: "ABORTED" } });
|
||||
expect(fetcher).not.toHaveBeenCalled();
|
||||
expect(setTimeout_).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("keeps the classified outcome when clearing the deadline throws", async () => {
|
||||
const png = pngBytes(640, 360);
|
||||
const probe = createBrowserImageProbe({
|
||||
fetcher: (async () =>
|
||||
responseAt(imageUrl, png, {
|
||||
status: 200,
|
||||
headers: publicImageHeaders("image/png", png.byteLength),
|
||||
})) as typeof fetch,
|
||||
createBitmap: vi.fn(async () => ({
|
||||
width: 640,
|
||||
height: 360,
|
||||
close: vi.fn(),
|
||||
})),
|
||||
timeoutMs: 1_000,
|
||||
scheduler: {
|
||||
setTimeout: (callback: () => void, milliseconds: number) =>
|
||||
setTimeout(callback, milliseconds),
|
||||
clearTimeout: () => {
|
||||
throw new TypeError("image scheduler clear exploded");
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
await expect(probe.probe(request())).resolves.toMatchObject({ ok: true });
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
describe("P-256 image capability verifier", () => {
|
||||
|
||||
@@ -494,4 +494,158 @@ describe("resumable upload Web Lock", () => {
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
/**
|
||||
* X-AUDIT-02. The public port promises a `UploadProviderResult`. A scheduler
|
||||
* that cannot install the attempt deadline must close the attempt inside that
|
||||
* contract instead of rejecting it, and must not leave the caller listener
|
||||
* attached to the parent signal.
|
||||
*/
|
||||
describe("scheduler boundary", () => {
|
||||
const trackedSignal = () => {
|
||||
const controller = new AbortController();
|
||||
const added: string[] = [];
|
||||
const removed: string[] = [];
|
||||
const add = controller.signal.addEventListener.bind(controller.signal);
|
||||
const remove = controller.signal.removeEventListener.bind(
|
||||
controller.signal,
|
||||
);
|
||||
Object.defineProperty(controller.signal, "addEventListener", {
|
||||
configurable: true,
|
||||
value: (type: string, ...rest: readonly unknown[]) => {
|
||||
added.push(type);
|
||||
return (add as (...args: readonly unknown[]) => unknown)(
|
||||
type,
|
||||
...rest,
|
||||
);
|
||||
},
|
||||
});
|
||||
Object.defineProperty(controller.signal, "removeEventListener", {
|
||||
configurable: true,
|
||||
value: (type: string, ...rest: readonly unknown[]) => {
|
||||
removed.push(type);
|
||||
return (remove as (...args: readonly unknown[]) => unknown)(
|
||||
type,
|
||||
...rest,
|
||||
);
|
||||
},
|
||||
});
|
||||
return { controller, added, removed };
|
||||
};
|
||||
|
||||
it("closes the attempt when the scheduler cannot install the deadline", async () => {
|
||||
const fetcher = vi.fn(async () => new Response(null, { status: 200 }));
|
||||
const { controller, added, removed } = trackedSignal();
|
||||
const transport = createTransport(fetcher as unknown as typeof fetch, {
|
||||
scheduler: {
|
||||
setTimeout: () => {
|
||||
throw new TypeError("upload scheduler install exploded");
|
||||
},
|
||||
clearTimeout: () => {},
|
||||
},
|
||||
});
|
||||
|
||||
const result = await transport.execute({
|
||||
operation: "GET_STATUS",
|
||||
body: { sessionId: "session_01" },
|
||||
signal: controller.signal,
|
||||
});
|
||||
|
||||
expect(result.ok).toBe(false);
|
||||
if (!result.ok) {
|
||||
expect(result.error.code).toBe("UNAVAILABLE");
|
||||
expect(result.error.retryable).toBe(true);
|
||||
}
|
||||
expect(fetcher).not.toHaveBeenCalled();
|
||||
expect(added.filter((type) => type === "abort")).toHaveLength(1);
|
||||
expect(removed.filter((type) => type === "abort")).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("keeps the classified outcome when clearing the deadline throws", async () => {
|
||||
const fetcher = vi.fn(
|
||||
async () =>
|
||||
jsonResponseAt(ENDPOINTS.GET_STATUS, { sessionId: "session_01" }),
|
||||
);
|
||||
const { controller, added, removed } = trackedSignal();
|
||||
const transport = createTransport(fetcher as unknown as typeof fetch, {
|
||||
scheduler: {
|
||||
setTimeout: (callback: () => void, delayMs: number) =>
|
||||
setTimeout(callback, delayMs),
|
||||
clearTimeout: () => {
|
||||
throw new TypeError("upload scheduler clear exploded");
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
const result = await transport.execute({
|
||||
operation: "GET_STATUS",
|
||||
body: { sessionId: "session_01" },
|
||||
signal: controller.signal,
|
||||
});
|
||||
|
||||
expect(result.ok).toBe(true);
|
||||
expect(added.filter((type) => type === "abort")).toHaveLength(1);
|
||||
expect(removed.filter((type) => type === "abort")).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("uses the scheduler methods captured at construction", async () => {
|
||||
const fetcher = vi.fn(
|
||||
async () =>
|
||||
jsonResponseAt(ENDPOINTS.GET_STATUS, { sessionId: "session_01" }),
|
||||
);
|
||||
const scheduler = {
|
||||
setTimeout: (callback: () => void, delayMs: number) =>
|
||||
setTimeout(callback, delayMs),
|
||||
clearTimeout: (handle: unknown) => {
|
||||
clearTimeout(handle as ReturnType<typeof setTimeout>);
|
||||
},
|
||||
};
|
||||
const transport = createTransport(fetcher as unknown as typeof fetch, {
|
||||
scheduler,
|
||||
});
|
||||
scheduler.setTimeout = () => {
|
||||
throw new TypeError("mutated upload setTimeout");
|
||||
};
|
||||
scheduler.clearTimeout = () => {
|
||||
throw new TypeError("mutated upload clearTimeout");
|
||||
};
|
||||
|
||||
await expect(
|
||||
transport.execute({
|
||||
operation: "GET_STATUS",
|
||||
body: { sessionId: "session_01" },
|
||||
signal: new AbortController().signal,
|
||||
}),
|
||||
).resolves.toMatchObject({ ok: true });
|
||||
});
|
||||
|
||||
it("starts no timer and no fetch for an already aborted caller", async () => {
|
||||
const fetcher = vi.fn(async () => new Response(null, { status: 200 }));
|
||||
const setTimeout_ = vi.fn(
|
||||
(callback: () => void, delayMs: number) =>
|
||||
setTimeout(callback, delayMs) as unknown,
|
||||
);
|
||||
const controller = new AbortController();
|
||||
controller.abort();
|
||||
const transport = createTransport(fetcher as unknown as typeof fetch, {
|
||||
scheduler: {
|
||||
setTimeout: setTimeout_,
|
||||
clearTimeout: (handle: unknown) => {
|
||||
clearTimeout(handle as ReturnType<typeof setTimeout>);
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
const result = await transport.execute({
|
||||
operation: "GET_STATUS",
|
||||
body: { sessionId: "session_01" },
|
||||
signal: controller.signal,
|
||||
});
|
||||
|
||||
expect(result.ok).toBe(false);
|
||||
if (!result.ok) expect(result.error.code).toBe("ABORTED");
|
||||
expect(fetcher).not.toHaveBeenCalled();
|
||||
expect(setTimeout_).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user