diff --git a/src/adapters/browser-transfer/presigned/presigned-capability-http-provider.ts b/src/adapters/browser-transfer/presigned/presigned-capability-http-provider.ts index ced5f4a..1cdc699 100644 --- a/src/adapters/browser-transfer/presigned/presigned-capability-http-provider.ts +++ b/src/adapters/browser-transfer/presigned/presigned-capability-http-provider.ts @@ -10,6 +10,7 @@ import type { PresignedUploadPartCapability, PresignedUploadPartCapabilityProvider, } from "../../../application/ports/browser-transfer/presigned-transfer.ts"; +import { createAbortableOperation } from "../../platform/abortable-operation.ts"; import { RESUMABLE_UPLOAD_PROTOCOL } from "../../../application/ports/browser-transfer/resumable-upload.ts"; import type { BrowserDataFailureCode, @@ -637,6 +638,9 @@ function validateCapabilityPayload( return browserDataSuccess( Object.freeze({ + // TR-RR-03. The registration carries the negotiated protocol version so + // the vault validates the same shape the executor later reads. + protocol: PRESIGNED_TRANSFER_PROTOCOL, capabilityReceipt, method: context.expected.method, binding, @@ -876,74 +880,87 @@ function statusFailure( /** BT-PRE-03. The scope ended before the task settled. */ const SCOPE_ENDED = Symbol("presigned-capability-scope-ended"); -function compensateLateResponse(task: Promise): void { - void task - .then(async (value) => { - const body = (value as { body?: { cancel(): Promise } | null }) - ?.body; - await body?.cancel(); - }) - .catch(() => undefined); -} - +/** + * TR-RR-01 / TR-RR-05. The abort scope is a thin projection of the shared + * `createAbortableOperation` primitive into this subsystem's vocabulary. + * + * Two things changed with the migration. `abort()` exists, so `close()` on a + * download can actually stop a pending fetch instead of only dropping + * listeners; and additional ownership signals — a consumer's stream signal — + * are composed into the same operation *before* any I/O starts, so an + * already-aborted consumer never causes a fetch. + */ function createAbortScope( external: AbortSignal, timeoutMs: number, scheduler: Scheduler, + additionalSignals: readonly AbortSignal[] = [], ) { - const controller = new AbortController(); - let timedOut = false; - const onAbort = () => controller.abort(external.reason); - const releaseListener = () => { - try { - external.removeEventListener("abort", onAbort); - } catch { - // A hostile signal facade cannot block the rest of cleanup. - } - }; - external.addEventListener("abort", onAbort, { once: true }); - if (external.aborted) onAbort(); - let timer: unknown; - try { - timer = scheduler.setTimeout(() => { - timedOut = true; - controller.abort("timeout"); - }, timeoutMs); - } catch { - // BT-PRE-03. A throwing scheduler must not leak the external listener. - releaseListener(); - timer = undefined; - } - return Object.freeze({ - signal: controller.signal, - timedOut: () => timedOut, - /** BT-PRE-03. Bounds a fetch that ignores its signal. */ - async race(task: Promise): Promise { - task.catch(() => {}); - if (controller.signal.aborted) { - compensateLateResponse(task); - return SCOPE_ENDED; - } - const ended = new Promise((resolve) => { - controller.signal.addEventListener( - "abort", - () => resolve(SCOPE_ENDED), - { once: true }, - ); - }); - const raced = await Promise.race([task, ended]); - if (raced === SCOPE_ENDED) compensateLateResponse(task); - return raced; - }, - release() { - try { - if (timer !== undefined) scheduler.clearTimeout(timer as never); - } catch { - // Cleanup failures cannot change the classified result. - } - releaseListener(); + const operation = createAbortableOperation({ + signal: external, + timeoutMs, + setTimer: (callback, delayMs) => scheduler.setTimeout(callback, delayMs), + clearTimer: (handle) => { + scheduler.clearTimeout(handle as never); }, }); + const releases: (() => void)[] = []; + for (const extra of additionalSignals) { + if (!extra) continue; + if (extra.aborted) { + operation.close(); + continue; + } + const onAbort = () => operation.close(); + try { + extra.addEventListener("abort", onAbort, { once: true }); + releases.push(() => { + try { + extra.removeEventListener("abort", onAbort); + } catch { + // A hostile signal facade cannot block the rest of cleanup. + } + }); + } catch { + // A signal that refuses a listener cannot bound this operation, so the + // operation closes rather than running unbounded. + operation.close(); + } + } + const releaseExtras = () => { + for (const release of releases.splice(0)) release(); + }; + return Object.freeze({ + signal: operation.signal, + timedOut: () => operation.terminal() === "DEADLINE", + /** Bounds a fetch that ignores its signal. */ + async race(task: Promise): Promise { + const outcome = await operation.race( + task, + (value) => compensateLateResponseValue(value), + ); + if (outcome.kind === "VALUE") return outcome.value; + // A collaborator's own rejection stays a rejection: the caller's existing + // catch classifies it, and it is never forged into a cancellation. + if (outcome.kind === "REJECTED") throw outcome.reason; + return SCOPE_ENDED; + }, + /** TR-RR-01. Ends the physical work, not just the bookkeeping. */ + abort() { + operation.close(); + releaseExtras(); + }, + release() { + operation.close(); + releaseExtras(); + }, + }); +} + +function compensateLateResponseValue(value: unknown): void { + const body = (value as { body?: { cancel(): Promise } | null } | null) + ?.body; + void body?.cancel().catch(() => undefined); } function transferFailure( diff --git a/src/adapters/browser-transfer/presigned/presigned-capability-vault.ts b/src/adapters/browser-transfer/presigned/presigned-capability-vault.ts index 64ae8c2..800844e 100644 --- a/src/adapters/browser-transfer/presigned/presigned-capability-vault.ts +++ b/src/adapters/browser-transfer/presigned/presigned-capability-vault.ts @@ -1,9 +1,11 @@ -import type { - PresignedTransferBinding, - PresignedTransferCapability, - PresignedTransferCapabilityReceipt, - PresignedTransferMethod, - PresignedTransferReplayGuard, +import { + PRESIGNED_TRANSFER_PROTOCOL, + type PresignedTransferBinding, + type PresignedTransferCapability, + type PresignedTransferCapabilityReceipt, + type PresignedTransferMethod, + type PresignedTransferProtocol, + type PresignedTransferReplayGuard, } from "../../../application/ports/browser-transfer/presigned-transfer.ts"; import type { BrowserDataResult } from "../../../application/ports/browser-file-storage/shared.ts"; import { @@ -16,7 +18,51 @@ export type PresignedHeaderBinding = Readonly<{ value: string; }>; +/** + * TR-RR-03. A registration is a versioned exact union. Without the version an + * issuer from another release could register a shape this vault validates under + * different rules than the executor later applies to it. + */ +/** TR-RR-03. The exact own-data key set a registration may carry. */ +const REGISTRATION_KEYS: readonly string[] = Object.freeze( + [ + "allowedQueryParameters", + "binding", + "byteLength", + "capabilityReceipt", + "digestRequestHeader", + "digestResponseHeader", + "expectedResponseByteLength", + "expectedSha256", + "expectedStatus", + "expiresAtEpochMs", + "href", + "maxBytes", + "mediaType", + "method", + "origin", + "path", + "protocol", + "receiptResponseHeader", + "requestHeaders", + "requiredResponseHeaders", + ].sort(), +); + +/** + * TR-RR-03. Ambient credential and cookie headers are forbidden on a presigned + * capability: the signature is the authorization. + */ +const FORBIDDEN_CAPABILITY_HEADERS: ReadonlySet = new Set([ + "authorization", + "cookie", + "cookie2", + "proxy-authorization", + "set-cookie", +]); + export type PresignedCapabilityRegistration = Readonly<{ + protocol: PresignedTransferProtocol; capabilityReceipt: PresignedTransferCapabilityReceipt; method: PresignedTransferMethod; binding: PresignedTransferBinding; @@ -110,8 +156,8 @@ export function createPresignedCapabilityVault(options: Readonly<{ } /** - * BT-PRE-04. Runtime invariants every registration must satisfy, regardless - * of which issuer produced it. + * BT-PRE-04 / TR-RR-03. Runtime invariants every registration must satisfy, + * regardless of which issuer produced it. */ function validatePresignedCapabilityRegistration( registration: PresignedCapabilityRegistration, @@ -121,6 +167,31 @@ export function createPresignedCapabilityVault(options: Readonly<{ recovery: "REISSUE_CAPABILITY", }); if (!registration || typeof registration !== "object") return invalid(); + // TR-RR-03. Exact own data only: an accessor re-runs on every later read, + // an inherited field can be replaced through the prototype, and a symbol + // key escapes a name-based sweep. The vault validates what the executor + // will read, so it must read what it validated. + try { + if (Object.getOwnPropertySymbols(registration).length > 0) { + return invalid(); + } + const names = Object.getOwnPropertyNames(registration).sort(); + if ( + names.length !== REGISTRATION_KEYS.length || + names.some((name, index) => name !== REGISTRATION_KEYS[index]) + ) { + return invalid(); + } + for (const name of names) { + const descriptor = Object.getOwnPropertyDescriptor(registration, name); + if (!descriptor || !("value" in descriptor)) return invalid(); + } + } catch { + return invalid(); + } + if (registration.protocol !== PRESIGNED_TRANSFER_PROTOCOL) { + return invalid(); + } if ( typeof registration.capabilityReceipt !== "string" || registration.capabilityReceipt.length === 0 || @@ -137,6 +208,10 @@ export function createPresignedCapabilityVault(options: Readonly<{ return invalid(); } if ( + // TR-RR-03. A presigned capability travels the network as a bearer of its + // own authority, so plaintext is never acceptable. + target.protocol !== "https:" || + origin.protocol !== "https:" || target.origin !== origin.origin || origin.href.replace(/\/$/u, "") !== registration.origin.replace(/\/$/u, "") || target.pathname !== registration.path || @@ -146,6 +221,32 @@ export function createPresignedCapabilityVault(options: Readonly<{ ) { return invalid(); } + // TR-RR-03. A presigned URL already carries its authorization. An ambient + // credential header alongside it would send the user's session to an + // origin the capability alone was meant to reach. + for (const header of [ + ...registration.requestHeaders, + ...registration.requiredResponseHeaders, + ]) { + if ( + !header || + typeof header.name !== "string" || + typeof header.value !== "string" || + FORBIDDEN_CAPABILITY_HEADERS.has(header.name.toLowerCase()) + ) { + return invalid(); + } + } + if ( + !Number.isSafeInteger(registration.expectedStatus) || + registration.expectedStatus < 200 || + registration.expectedStatus > 299 || + (registration.expectedResponseByteLength !== null && + (!Number.isSafeInteger(registration.expectedResponseByteLength) || + registration.expectedResponseByteLength < 0)) + ) { + return invalid(); + } if ( !Number.isSafeInteger(registration.byteLength) || registration.byteLength < 0 || @@ -203,6 +304,7 @@ export function createPresignedCapabilityVault(options: Readonly<{ }) as PresignedTransferCapability; const binding: PresignedCapabilityBinding = Object.freeze({ capability, + protocol: PRESIGNED_TRANSFER_PROTOCOL, capabilityReceipt: capability.capabilityReceipt, method: capability.method, binding: capability.binding, diff --git a/src/adapters/browser-transfer/presigned/presigned-transfer-executor.ts b/src/adapters/browser-transfer/presigned/presigned-transfer-executor.ts index 739b41c..8c7d7d9 100644 --- a/src/adapters/browser-transfer/presigned/presigned-transfer-executor.ts +++ b/src/adapters/browser-transfer/presigned/presigned-transfer-executor.ts @@ -8,6 +8,7 @@ import type { PresignedUploadPartOutcome, PresignedUploadPartPort, } from "../../../application/ports/browser-transfer/presigned-transfer.ts"; +import { createAbortableOperation } from "../../platform/abortable-operation.ts"; import { RESUMABLE_UPLOAD_PROTOCOL } from "../../../application/ports/browser-transfer/resumable-upload.ts"; import type { BrowserDataFailureCode, @@ -171,7 +172,13 @@ export function createPresignedTransferExecutor( }), validateResponse: (response) => validateDownloadResponse(response, binding), - createScope: () => createAbortScope(signal, timeoutMs, scheduler), + createScope: (consumerSignal?: AbortSignal) => + createAbortScope( + signal, + timeoutMs, + scheduler, + consumerSignal ? [consumerSignal] : [], + ), recheckExpiry: () => validateExpiry(capability, minimumRemainingLifetimeMs, now()), binding, @@ -266,37 +273,51 @@ export function createPresignedTransferExecutor( ) { return browserDataFailure("POLICY_REJECTED", "PRESIGNED_TRANSFER"); } - let actualDigest: string; - try { - actualDigest = normalizedSha256(await digestBytes(bytes)); - } catch { - return browserDataFailure( - "INTEGRITY_FAILED", - "PRESIGNED_TRANSFER", - ); - } - if (actualDigest !== capability.expectedSha256) { - return browserDataFailure( - "INTEGRITY_FAILED", - "PRESIGNED_TRANSFER", - ); - } - if (request.signal.aborted) { - return browserDataFailure("ABORTED", "PRESIGNED_TRANSFER"); - } - const active = validateExpiry( - capability, - minimumRemainingLifetimeMs, - now(), - ); - if (!active.ok) return active; - const claimed = claim(capability); - if (!claimed.ok) return claimed; - const consumed = consume(capability); - if (!consumed.ok) return consumed; - + // TR-RR-02. The abort scope is created first, so the digest — which can be + // a long or non-settling computation over a large buffer — is owned by the + // caller signal and the deadline like every other step. Computing it before + // the scope existed meant an abort or a deadline could not reach it, and a + // hash that never settled held the whole `put` open. const scope = createAbortScope(request.signal, timeoutMs, scheduler); try { + const digested = await scope.race( + Promise.resolve().then(async () => await digestBytes(bytes)), + ); + if (digested === SCOPE_ENDED) { + return transferFailure(request.signal, scope.timedOut()); + } + let actualDigest: string; + try { + actualDigest = normalizedSha256(digested); + } catch { + return browserDataFailure( + "INTEGRITY_FAILED", + "PRESIGNED_TRANSFER", + ); + } + if (actualDigest !== capability.expectedSha256) { + return browserDataFailure( + "INTEGRITY_FAILED", + "PRESIGNED_TRANSFER", + ); + } + // The owner is re-checked after the wait: a claim and a network call may + // only follow a digest that finished while this operation still held the + // execution. + if (scope.signal.aborted || request.signal.aborted) { + return transferFailure(request.signal, scope.timedOut()); + } + const active = validateExpiry( + capability, + minimumRemainingLifetimeMs, + now(), + ); + if (!active.ok) return active; + const claimed = claim(capability); + if (!claimed.ok) return claimed; + const consumed = consume(capability); + if (!consumed.ok) return consumed; + const raced = await scope.race( fetcher(binding.href, { method: "PUT", @@ -409,7 +430,9 @@ function createDownloadSource(input: Readonly<{ scope: ReturnType, ) => Promise; validateResponse: (response: Response) => BrowserDataResult; - createScope: () => ReturnType; + createScope: ( + consumerSignal?: AbortSignal, + ) => ReturnType; recheckExpiry: () => BrowserDataResult; binding: PresignedCapabilityBinding; capability: PresignedDownloadCapability; @@ -425,7 +448,13 @@ function createDownloadSource(input: Readonly<{ let activeScope: ReturnType | undefined; let activeResponse: Response | undefined; + // TR-RR-01. Releasing the scope only dropped listeners and the timer, so a + // fetch or read already in flight kept running after `close()`. The scope is + // aborted here, which is what actually ends the physical I/O. const releaseActive = () => { + if (activeScope) { + activeScope.abort(); + } if (activeResponse) { cancelBody(activeResponse); activeResponse = undefined; @@ -486,8 +515,22 @@ function createDownloadSource(input: Readonly<{ yield stillActive as BrowserDataResult; return; } - const scope = input.createScope(); + // TR-RR-01. The consumer's stream signal is part of this operation's + // ownership from the start. Composing it only after the fetch had begun + // meant an already-aborted consumer still caused one network request. + const scope = input.createScope(consumerSignal); activeScope = scope; + if (scope.signal.aborted) { + state = "CLOSED"; + releaseActive(); + const failure = transferFailure( + input.externalSignal, + scope.timedOut(), + ); + observeTransferResult(input.observer, "DOWNLOAD", failure, 0); + yield failure; + return; + } let response: Response; try { // BT-PRE-03. A fetch that ignores its signal cannot outlive the scope. @@ -524,9 +567,6 @@ function createDownloadSource(input: Readonly<{ yield validated as BrowserDataResult; return; } - let combined: - | ReturnType - | undefined; let reader: | ReadableStreamDefaultReader | undefined; @@ -544,10 +584,6 @@ function createDownloadSource(input: Readonly<{ return browserDataFailure(code, "PRESIGNED_TRANSFER", options); }; try { - combined = combineConsumerAbort( - scope, - consumerSignal, - ); const verifier = input.createVerifier( input.capability.expectedSha256, ); @@ -676,7 +712,6 @@ function createDownloadSource(input: Readonly<{ }); } } finally { - combined?.release(); if (!completed) { if (reader) cancelReader(reader); else cancelBody(response); @@ -1016,88 +1051,74 @@ function compensateLateResponse(task: Promise): void { .catch(() => undefined); } +/** + * TR-RR-01 / TR-RR-05. A projection of the shared `createAbortableOperation` + * primitive into this subsystem's vocabulary, replacing a second hand-written + * copy of the same mechanics. + * + * `additionalSignals` lets a consumer's stream signal join the operation's + * ownership before any I/O begins, and `abort()` ends the physical work rather + * than only releasing bookkeeping. + */ function createAbortScope( external: AbortSignal, timeoutMs: number, scheduler: Scheduler, + additionalSignals: readonly AbortSignal[] = [], ) { - const controller = new AbortController(); - let timedOut = false; - let released = false; - const onAbort = () => controller.abort(external.reason); - const releaseListener = () => { - try { - external.removeEventListener("abort", onAbort); - } catch { - // A hostile signal facade cannot block the rest of cleanup. - } - }; - external.addEventListener("abort", onAbort, { once: true }); - if (external.aborted) onAbort(); - let timer: unknown; - try { - timer = scheduler.setTimeout(() => { - timedOut = true; - controller.abort("timeout"); - releaseListener(); - }, timeoutMs); - } catch { - // BT-PRE-03. A scheduler that throws must not leave the external listener - // attached; the scope stays bounded by the caller signal alone. - releaseListener(); - timer = undefined; - } - return Object.freeze({ - signal: controller.signal, - timedOut: () => timedOut, - /** - * BT-PRE-03. Aborting the controller does not settle a fetch that ignores - * its signal, so the caller-facing wait is raced against the scope and a - * late `Response` has its body cancelled. - */ - async race(task: Promise): Promise { - task.catch(() => {}); - if (controller.signal.aborted) { - compensateLateResponse(task); - return SCOPE_ENDED; - } - const ended = new Promise((resolve) => { - controller.signal.addEventListener( - "abort", - () => resolve(SCOPE_ENDED), - { once: true }, - ); - }); - const raced = await Promise.race([task, ended]); - if (raced === SCOPE_ENDED) compensateLateResponse(task); - return raced; - }, - abort(reason?: unknown) { - controller.abort(reason); - }, - release() { - if (released) return; - released = true; - try { - if (timer !== undefined) scheduler.clearTimeout(timer as never); - } catch { - // Cleanup failures cannot change the classified result. - } - releaseListener(); + const operation = createAbortableOperation({ + signal: external, + timeoutMs, + setTimer: (callback, delayMs) => scheduler.setTimeout(callback, delayMs), + clearTimer: (handle) => { + scheduler.clearTimeout(handle as never); }, }); -} - -function combineConsumerAbort( - scope: ReturnType, - consumer: AbortSignal, -) { - const onAbort = () => scope.abort(consumer.reason); - consumer.addEventListener("abort", onAbort, { once: true }); - if (consumer.aborted) onAbort(); + const releases: (() => void)[] = []; + for (const extra of additionalSignals) { + if (!extra) continue; + if (extra.aborted) { + operation.close(); + continue; + } + const onAbort = () => operation.close(); + try { + extra.addEventListener("abort", onAbort, { once: true }); + releases.push(() => { + try { + extra.removeEventListener("abort", onAbort); + } catch { + // A hostile signal facade cannot block the rest of cleanup. + } + }); + } catch { + operation.close(); + } + } + const releaseExtras = () => { + for (const release of releases.splice(0)) release(); + }; return Object.freeze({ + signal: operation.signal, + timedOut: () => operation.terminal() === "DEADLINE", + async race(task: Promise): Promise { + const outcome = await operation.race(task, (value) => { + const body = ( + value as { body?: { cancel(): Promise } | null } | null + )?.body; + void body?.cancel().catch(() => undefined); + }); + if (outcome.kind === "VALUE") return outcome.value; + if (outcome.kind === "REJECTED") throw outcome.reason; + return SCOPE_ENDED; + }, + abort() { + operation.close(); + releaseExtras(); + }, release() { - consumer.removeEventListener("abort", onAbort); + operation.close(); + releaseExtras(); }, }); } diff --git a/src/adapters/platform/abortable-operation.ts b/src/adapters/platform/abortable-operation.ts index cb4f67d..6466cc7 100644 --- a/src/adapters/platform/abortable-operation.ts +++ b/src/adapters/platform/abortable-operation.ts @@ -10,8 +10,15 @@ export type AbortTerminalReason = "CALLER_ABORT" | "DEADLINE" | "CLOSED"; +/** + * TR-RR-05. A collaborator's own rejection is evidence about the work, so it is + * a distinct outcome. Forging it into `TERMINAL` erased the reason the + * operation actually failed and made `race()` disagree with `terminal()`, which + * still reported no owner. + */ export type AbortRace = | Readonly<{ kind: "VALUE"; value: Value }> + | Readonly<{ kind: "REJECTED"; reason: unknown }> | Readonly<{ kind: "TERMINAL"; terminal: AbortTerminalReason }>; export type AbortableOperation = Readonly<{ @@ -23,10 +30,17 @@ export type AbortableOperation = Readonly<{ */ terminal(): AbortTerminalReason | null; /** - * Resolves with the operation's value, or with the first terminal owner. - * A terminal race never returns a bare value. + * Resolves with the operation's value, its own rejection, or the first + * terminal owner. A terminal race never returns a bare value, and the + * `terminal` it reports is always the same owner `terminal()` reports. + * + * `compensate` is invoked at most once, and only for a value that arrived + * after the operation already ended. */ - race(operation: Promise): Promise>; + race( + operation: Promise, + compensate?: (value: Value) => void, + ): Promise>; /** * Idempotent. Removes listeners, clears the deadline timer and marks the * operation `CLOSED` if nothing terminal happened first. @@ -46,6 +60,10 @@ export function createAbortableOperation( input: AbortableOperationInput = {}, ): AbortableOperation { const controller = new AbortController(); + // TR-RR-05. The scheduler and the caller signal are captured once, so + // replacing a method on the input object after construction cannot change how + // an operation already in flight is bounded or cleaned up. + const callerSignal = input.signal; const setTimer = input.setTimer ?? ((callback: () => void, delayMs: number) => setTimeout(callback, delayMs)); @@ -74,7 +92,7 @@ export function createAbortableOperation( if (disposed) return; disposed = true; try { - input.signal?.removeEventListener("abort", onCallerAbort); + callerSignal?.removeEventListener("abort", onCallerAbort); } catch { // A hostile signal facade cannot block cleanup of the rest. } @@ -88,11 +106,11 @@ export function createAbortableOperation( } } - if (input.signal?.aborted) { + if (callerSignal?.aborted) { settle("CALLER_ABORT"); disposed = true; - } else if (input.signal) { - input.signal.addEventListener("abort", onCallerAbort, { once: true }); + } else if (callerSignal) { + callerSignal.addEventListener("abort", onCallerAbort, { once: true }); } if ( @@ -107,9 +125,12 @@ export function createAbortableOperation( dispose(); }, input.timeoutMs); } catch { - // A scheduler that throws leaves no timer behind; the operation stays - // bounded only by the caller signal. + // TR-RR-05. A scheduler that cannot install the deadline leaves the + // operation unbounded. Removing the caller listener and leaving no + // terminal owner made a later abort invisible, so installation failure is + // itself terminal: the operation closes atomically with its resources. timer = undefined; + settle("CLOSED"); dispose(); } } @@ -117,55 +138,91 @@ export function createAbortableOperation( return Object.freeze({ signal: controller.signal, terminal: () => terminalReason, - async race(operation: Promise): Promise> { - // Observe a late rejection so an abandoned operation cannot surface as an - // unhandled rejection. - operation.catch(() => {}); - if (terminalReason !== null) { - return Object.freeze({ + async race( + operation: Promise, + compensate?: (value: Value) => void, + ): Promise> { + // The outcome is derived once, so a late rejection can never surface as + // an unhandled rejection and a late value is compensated exactly once. + let landed: AbortRace | null = null; + const settled: Promise> = operation.then( + (value) => (landed = Object.freeze({ kind: "VALUE" as const, value })), + (reason: unknown) => + (landed = Object.freeze({ kind: "REJECTED" as const, reason })), + ); + let compensated = false; + const compensateOnce = (value: Value) => { + if (compensated || !compensate) return; + compensated = true; + try { + compensate(value); + } catch { + // Compensation is best effort and never changes the outcome. + } + }; + // Attached to the original promise, not to the derived one, so a late + // value is compensated in the same turn it lands rather than a turn + // later. `settled` already absorbs the rejection. + const observeLate = () => { + void operation.then( + (value) => compensateOnce(value), + () => undefined, + ); + }; + const terminalRace = (): AbortRace => + Object.freeze({ kind: "TERMINAL" as const, - terminal: terminalReason, + // The owner reported here is always the owner `terminal()` reports. + terminal: terminalReason ?? ("CLOSED" as const), }); + + if (terminalReason !== null) { + observeLate(); + return terminalRace(); } - const raced = await Promise.race([ - operation.then( - (value) => Object.freeze({ kind: "VALUE" as const, value }), - () => - Object.freeze({ - kind: "TERMINAL" as const, - terminal: terminalReason ?? ("CLOSED" as const), - }), - ), - new Promise>((resolve) => { - if (controller.signal.aborted) { - resolve( - Object.freeze({ - kind: "TERMINAL" as const, - terminal: terminalReason ?? ("CLOSED" as const), - }), - ); - return; + let onAbort: (() => void) | undefined; + const terminated = new Promise>((resolve) => { + if (controller.signal.aborted) { + resolve(terminalRace()); + return; + } + onAbort = () => resolve(terminalRace()); + controller.signal.addEventListener("abort", onAbort, { once: true }); + }); + try { + const raced = await Promise.race([settled, terminated]); + if (raced.kind !== "TERMINAL") { + // A value that arrives while a terminal owner already holds the + // operation is late: it is compensated, not admitted. A rejection is + // kept, because it is evidence about the work rather than a claim + // that the work succeeded. + if (raced.kind === "VALUE" && terminalReason !== null) { + compensateOnce(raced.value); + return terminalRace(); } - controller.signal.addEventListener( - "abort", - () => - resolve( - Object.freeze({ - kind: "TERMINAL" as const, - terminal: terminalReason ?? ("CLOSED" as const), - }), - ), - { once: true }, - ); - }), - ]); - // A value that arrives after a terminal owner is not admitted. - return terminalReason !== null && raced.kind === "VALUE" - ? Object.freeze({ - kind: "TERMINAL" as const, - terminal: terminalReason, - }) - : raced; + return raced; + } + // The terminal owner reached the await first. One microtask drain lets + // an operation that had already settled hand over its outcome; it + // cannot be extended by a collaborator that has not settled. + for (let turn = 0; turn < 4 && landed === null; turn += 1) { + await Promise.resolve(); + } + const settledOutcome = landed as AbortRace | null; + if (settledOutcome !== null) { + if (settledOutcome.kind === "VALUE") { + compensateOnce(settledOutcome.value); + return terminalRace(); + } + return settledOutcome; + } + observeLate(); + return terminalRace(); + } finally { + if (onAbort) { + controller.signal.removeEventListener("abort", onAbort); + } + } }, close() { settle("CLOSED"); diff --git a/tests/unit/abortable-operation.test.ts b/tests/unit/abortable-operation.test.ts index 761aac8..b42a7ae 100644 --- a/tests/unit/abortable-operation.test.ts +++ b/tests/unit/abortable-operation.test.ts @@ -81,13 +81,75 @@ describe("shared abortable operation mechanics", () => { expect(operation.terminal()).toBe("CLOSED"); }); - it("observes a late rejection instead of leaking it", async () => { + /** + * TR-RR-05. A collaborator's own rejection is evidence about the work. + * Reporting it as `TERMINAL/CLOSED` erased the reason the operation failed + * and made `race()` disagree with `terminal()`, which still said no owner. + */ + it("keeps a rejection distinct from a terminal owner", async () => { const operation = createAbortableOperation(); - const rejected = Promise.reject(new Error("late")); - await expect(operation.race(rejected)).resolves.toEqual({ + const reason = new Error("upstream failed"); + await expect( + operation.race(Promise.reject(reason)), + ).resolves.toEqual({ kind: "REJECTED", reason }); + expect(operation.terminal()).toBeNull(); + }); + + it("agrees with terminal() on the first owner", async () => { + const caller = new AbortController(); + const operation = createAbortableOperation({ signal: caller.signal }); + const pending = operation.race(new Promise(() => {})); + caller.abort(); + // A later close cannot overwrite the first owner in either place. + operation.close(); + await expect(pending).resolves.toEqual({ kind: "TERMINAL", - terminal: "CLOSED", + terminal: "CALLER_ABORT", }); + expect(operation.terminal()).toBe("CALLER_ABORT"); + }); + + it("compensates a late value exactly once instead of admitting it", async () => { + const caller = new AbortController(); + const operation = createAbortableOperation({ signal: caller.signal }); + const compensated: string[] = []; + let release: ((value: string) => void) | undefined; + const pending = operation.race( + new Promise((resolve) => { + release = resolve; + }), + (value) => compensated.push(value), + ); + caller.abort(); + expect(await pending).toEqual({ + kind: "TERMINAL", + terminal: "CALLER_ABORT", + }); + release?.("late-value"); + await Promise.resolve(); + await Promise.resolve(); + expect(compensated).toEqual(["late-value"]); + }); + + it("ignores a scheduler method replaced after construction", async () => { + const scheduler = { + setTimer: (callback: () => void, delayMs: number) => + setTimeout(callback, delayMs), + clearTimer: (handle: unknown) => { + clearTimeout(handle as ReturnType); + }, + }; + const operation = createAbortableOperation({ + timeoutMs: 5, + setTimer: scheduler.setTimer, + clearTimer: scheduler.clearTimer, + }); + scheduler.setTimer = () => { + throw new TypeError("replaced after construction"); + }; + await expect( + operation.race(new Promise(() => {})), + ).resolves.toEqual({ kind: "TERMINAL", terminal: "DEADLINE" }); }); it("close is idempotent and releases listeners and timers exactly once", () => { @@ -109,7 +171,13 @@ describe("shared abortable operation mechanics", () => { expect(clearTimer).toHaveBeenCalledTimes(1); }); - it("survives a throwing scheduler without leaking a listener", () => { + /** + * TR-RR-05. A scheduler that cannot install the deadline leaves the operation + * unbounded. Removing the caller listener and leaving no terminal owner made + * every later abort invisible, so installation failure is itself terminal and + * closes atomically with the resources already created. + */ + it("closes atomically when the scheduler cannot install the deadline", async () => { const caller = new AbortController(); const remove = vi.spyOn(caller.signal, "removeEventListener"); const operation = createAbortableOperation({ @@ -121,8 +189,13 @@ describe("shared abortable operation mechanics", () => { clearTimer: () => {}, }); - expect(operation.terminal()).toBeNull(); + expect(operation.terminal()).toBe("CLOSED"); expect(remove).toHaveBeenCalledTimes(1); + // The operation is bounded, so a caller abort afterwards cannot be lost. + caller.abort(); + await expect( + operation.race(new Promise(() => {})), + ).resolves.toEqual({ kind: "TERMINAL", terminal: "CLOSED" }); }); it("compensates a late native handle without changing the outcome", async () => { diff --git a/tests/unit/presigned-transfer.test.ts b/tests/unit/presigned-transfer.test.ts index 5033df9..e7c6eb8 100644 --- a/tests/unit/presigned-transfer.test.ts +++ b/tests/unit/presigned-transfer.test.ts @@ -801,7 +801,13 @@ describe("presigned transfer", () => { void payload; }); - it("does not leak an abort listener when the scheduler throws", async () => { + /** + * TR-RR-05. A scheduler that cannot install the deadline leaves the operation + * unbounded. Releasing the caller listener and continuing anyway meant a + * later abort was invisible, so an install failure is itself terminal: the + * request fails closed with a typed Result and its resources are released. + */ + it("fails closed when the scheduler cannot install the deadline", async () => { const bytes = new Uint8Array([1, 2, 3]); const payload = downloadCapabilityPayload(bytes); const caller = new AbortController(); @@ -827,7 +833,7 @@ describe("presigned transfer", () => { resourceId: "resource-1", signal: caller.signal, }), - ).resolves.toMatchObject({ ok: true }); + ).resolves.toMatchObject({ ok: false }); expect(remove).toHaveBeenCalled(); }); @@ -838,6 +844,32 @@ describe("presigned transfer", () => { { label: "maxBytes below byteLength", patch: { maxBytes: 0 } }, { label: "malformed digest", patch: { expectedSha256: "not-a-digest" } }, { label: "non-positive expiry", patch: { expiresAtEpochMs: 0 } }, + // TR-RR-03. The registration is a versioned exact union: plaintext, an + // ambient credential header, an unknown protocol version and any extra + // own field are all refused at the issuer seam. + { label: "unknown protocol version", patch: { protocol: "PRESIGNED_TRANSFER_V0" } }, + { label: "missing protocol version", patch: { protocol: undefined } }, + { + label: "plaintext target", + patch: { + href: `http://objects.example${DOWNLOAD_PATH}`, + origin: "http://objects.example", + }, + }, + { + label: "ambient credential header", + patch: { + requestHeaders: [{ name: "authorization", value: "Bearer leak" }], + }, + }, + { + label: "cookie response header", + patch: { + requiredResponseHeaders: [{ name: "set-cookie", value: "a=b" }], + }, + }, + { label: "status outside 2xx", patch: { expectedStatus: 302 } }, + { label: "extra own field", patch: { injected: true } }, ])( "rejects a malformed registration at the vault issuer seam ($label)", ({ patch }) => { @@ -848,6 +880,7 @@ describe("presigned transfer", () => { maxActiveCapabilities: 4, }); const base = { + protocol: PRESIGNED_TRANSFER_PROTOCOL, capabilityReceipt: "capability-direct-1", method: "GET" as const, binding: { kind: "DOWNLOAD" as const, resourceId: "resource-1" },