diff --git a/scripts/check-adapter-inventory.ts b/scripts/check-adapter-inventory.ts index 3ad921f..9ae7c0e 100644 --- a/scripts/check-adapter-inventory.ts +++ b/scripts/check-adapter-inventory.ts @@ -123,6 +123,27 @@ async function main(): Promise { ); } + // TR-RR-05. The shared abort/deadline primitive is only shared if production + // code imports it. A helper with zero importers is a second implementation + // waiting to happen, which is exactly how the four hand-written copies of + // these mechanics diverged in the first place. + const primitiveImporters = spawnSync( + "git", + ["grep", "-l", "platform/abortable-operation.ts", "--", "src"], + { encoding: "utf8" }, + ); + const importers = ( + primitiveImporters.status === 0 ? primitiveImporters.stdout : "" + ) + .split("\n") + .filter(Boolean) + .filter((file) => !file.endsWith("platform/abortable-operation.ts")); + if (importers.length === 0) { + problems.push( + "abortable-operation: the shared primitive has no production importers", + ); + } + if (problems.length > 0) { for (const problem of problems) console.error(problem); process.exitCode = 1; diff --git a/src/adapters/browser-files/download-delivery-adapter.ts b/src/adapters/browser-files/download-delivery-adapter.ts index 6b25314..687e834 100644 --- a/src/adapters/browser-files/download-delivery-adapter.ts +++ b/src/adapters/browser-files/download-delivery-adapter.ts @@ -569,11 +569,13 @@ async function promptAndStream(context: Readonly<{ ); } + const sourceHolder = createSourceHolder(); try { const sourceResult = await resolveByteSource( context.input, context.input.signal, context.options, + sourceHolder, ); if (!sourceResult.ok) { return observeResult(sourceResult, context.options.observer); @@ -702,6 +704,10 @@ async function promptAndStream(context: Readonly<{ mapDownloadException(error), context.options.observer, ); + } finally { + // TR-RR-04. Exactly once, on every path: success, validation failure, + // writer failure and abort. + closeHeldSource(sourceHolder); } } @@ -724,12 +730,15 @@ async function boundedObjectUrlHandoff(context: Readonly<{ context.options.observer, ); } + const sourceHolder = createSourceHolder(); const sourceResult = await resolveByteSource( context.input, context.input.signal, context.options, + sourceHolder, ); if (!sourceResult.ok) { + closeHeldSource(sourceHolder); return observeResult(sourceResult, context.options.observer); } const source = sourceResult.value; @@ -738,6 +747,7 @@ async function boundedObjectUrlHandoff(context: Readonly<{ (source.byteLength > context.input.maxBufferedBytes || source.byteLength > context.input.maxTransferBytes) ) { + closeHeldSource(sourceHolder); return observeResult( browserDataFailure("LIMIT_EXCEEDED", "DOWNLOAD"), context.options.observer, @@ -838,6 +848,9 @@ async function boundedObjectUrlHandoff(context: Readonly<{ context.options.observer, transferred, ); + } finally { + // TR-RR-04. Exactly once, on every path. + closeHeldSource(sourceHolder); } } @@ -997,10 +1010,37 @@ function validateDownloadInput( return browserDataSuccess(true); } +/** + * TR-RR-04. A presigned byte source owns a fetch reader and a capability lease, + * and its port requires `close()`. The delivery consumer never called it, so + * every success, validation failure, writer failure and abort leaked both. The + * closeable subtype is lost in the `FileByteSource` projection, so the holder + * keeps it and the outermost boundary closes it exactly once. + */ +type CloseableSourceHolder = { source: FileByteSource | null; closed: boolean }; + +function createSourceHolder(): CloseableSourceHolder { + return { source: null, closed: false }; +} + +function closeHeldSource(holder: CloseableSourceHolder): void { + if (holder.closed) return; + holder.closed = true; + const source = holder.source; + holder.source = null; + if (!source || !isVerifiedPresignedSource(source)) return; + try { + source.close(); + } catch { + // Closing is best effort and never changes the delivery outcome. + } +} + async function resolveByteSource( input: DeliveryInput, signal: AbortSignal, options: DownloadDeliveryAdapterOptions, + holder?: CloseableSourceHolder, ): Promise> { const source = input.source; if (signal.aborted) { @@ -1030,6 +1070,9 @@ async function resolveByteSource( recovery: result.error.recovery, }); } + // Held from the moment the lease exists, so a validation failure below still + // closes it. + if (holder) holder.source = result.value; return validPresignedByteSource( result.value, source.capability, diff --git a/src/adapters/browser-transfer/image-cdn/image-cdn-runtime.ts b/src/adapters/browser-transfer/image-cdn/image-cdn-runtime.ts index 08806d5..33b297d 100644 --- a/src/adapters/browser-transfer/image-cdn/image-cdn-runtime.ts +++ b/src/adapters/browser-transfer/image-cdn/image-cdn-runtime.ts @@ -323,6 +323,11 @@ export function createImageCdnRuntime( ); } activeCapabilityVerifications += 1; + // TR-RR-07. The slot belongs to the raw verifier, not to this wrapper. + // Releasing it when the wrapper's deadline expired let an abandoned + // verification keep running while a new one was admitted, so repeated + // timeouts produced more physical work than the configured cap allows. + const rawVerificationTasks: Promise[] = []; try { const canonicalPayload = canonicalImageCapabilityPayload(snapshot); @@ -345,8 +350,10 @@ export function createImageCdnRuntime( if (deadline.signal.aborted) { throw capabilityVerificationAbortException(); } + const digestTask = sha256Hex(digest, canonicalPayload); + rawVerificationTasks.push(digestTask); bindingDigest = await awaitImageRuntimeAbort( - sha256Hex(digest, canonicalPayload), + digestTask, deadline.signal, ); if ( @@ -363,14 +370,15 @@ export function createImageCdnRuntime( if (deadline.signal.aborted) { throw capabilityVerificationAbortException(); } + const verifyTask = verifyCapability({ + algorithm: snapshot.signature.algorithm, + keyId: snapshot.signature.keyId, + canonicalPayload: Uint8Array.from(canonicalPayload), + signatureBase64Url: snapshot.signature.valueBase64Url, + }); + rawVerificationTasks.push(verifyTask); verified = await awaitImageRuntimeAbort( - verifyCapability({ - algorithm: snapshot.signature.algorithm, - keyId: snapshot.signature.keyId, - canonicalPayload: Uint8Array.from(canonicalPayload), - signatureBase64Url: - snapshot.signature.valueBase64Url, - }), + verifyTask, deadline.signal, ); } catch { @@ -427,7 +435,10 @@ export function createImageCdnRuntime( ); return browserDataSuccess(reference); } finally { - activeCapabilityVerifications -= 1; + // Released only once the physical work this slot admitted has settled. + void Promise.allSettled(rawVerificationTasks).then(() => { + activeCapabilityVerifications -= 1; + }); } }; diff --git a/src/adapters/browser-transfer/resumable-upload/resumable-upload-runtime.ts b/src/adapters/browser-transfer/resumable-upload/resumable-upload-runtime.ts index 3ce9e1c..2b4adc6 100644 --- a/src/adapters/browser-transfer/resumable-upload/resumable-upload-runtime.ts +++ b/src/adapters/browser-transfer/resumable-upload/resumable-upload-runtime.ts @@ -77,7 +77,12 @@ export type ResumableUploadRuntime = ResumableUploadPort & * closes the checkpoint store and cancellation channel, so success actually * means quiescent. No current bootstrap consumer is assumed. */ - dispose(): Promise; + /** + * TR-RR-06. Closes admission and returns the *bounded* drain result. A + * failure means the runtime is still `CLOSING`: physical work the caller + * must not treat as finished is still in flight. + */ + dispose(): Promise>; lifecycle(): "OPEN" | "CLOSING" | "CLOSED"; }>; @@ -178,7 +183,7 @@ export function createResumableUploadRuntime( const activeOperations = new Set>(); let closed = false; let lifecycle: "OPEN" | "CLOSING" | "CLOSED" = "OPEN"; - let drain: Promise | null = null; + let drain: Promise> | null = null; const cancelLocalUploads = (uploadKey: string): void => { if (!SAFE_UPLOAD_KEY.test(uploadKey)) return; for (const controller of localUploads.get(uploadKey) ?? []) { @@ -280,17 +285,19 @@ export function createResumableUploadRuntime( dependencies.crossContextCancellation?.publish(uploadKey); } const combined = combineAbortSignals(input.signal, lifetime.signal); + const operation = dependencies.mutationLock.run( + uploadKey, + combined.signal, + async () => + await executeAbort(dependencies, uploadKey, combined.signal), + ); + // TR-RR-06. An abort is admitted physical work like an upload, so it is + // tracked from admission and `dispose()` cannot step over it. + const tracked = Promise.resolve(operation).catch(() => undefined); + activeOperations.add(tracked); + void tracked.finally(() => activeOperations.delete(tracked)); try { - return await dependencies.mutationLock.run( - uploadKey, - combined.signal, - async () => - await executeAbort( - dependencies, - uploadKey, - combined.signal, - ), - ); + return await operation; } catch (error) { return mapLockFailure(error, "UPLOAD_ABORT"); } finally { @@ -304,14 +311,14 @@ export function createResumableUploadRuntime( void startDrain(); }, - dispose(): Promise { + dispose(): Promise> { return startDrain(); }, lifecycle: () => lifecycle, }); - function startDrain(): Promise { + function startDrain(): Promise> { drain ??= (async () => { closed = true; lifecycle = "CLOSING"; @@ -322,10 +329,32 @@ export function createResumableUploadRuntime( for (const controllers of localUploads.values()) { for (const controller of controllers) controller.abort(); } - await Promise.allSettled([...activeOperations]); + // TR-RR-06. Bounded. A non-cooperative mutation lock or provider must not + // make teardown unbounded, and an unproved drain is reported as such + // rather than closed over. + let timer: ReturnType | undefined; + const expired = new Promise<"EXPIRED">((resolve) => { + timer = setTimeout( + () => resolve("EXPIRED"), + dependencies.policy.cleanupDeadlineMs, + ); + }); + const drained = await Promise.race([ + Promise.allSettled([...activeOperations]).then(() => "DRAINED" as const), + expired, + ]); + if (timer !== undefined) clearTimeout(timer); + if (drained === "EXPIRED") { + // The store stays open: something can still write a checkpoint. + return browserDataFailure("UNAVAILABLE", "UPLOAD_ABORT", { + retryable: true, + recovery: "RESUME", + }); + } // The store closes only after nothing can still write a checkpoint. dependencies.checkpoints.close(); lifecycle = "CLOSED"; + return browserDataSuccess(undefined); })(); return drain; } diff --git a/src/adapters/browser-transfer/resumable-upload/runtime-policy.ts b/src/adapters/browser-transfer/resumable-upload/runtime-policy.ts index b6db27a..ebd81b5 100644 --- a/src/adapters/browser-transfer/resumable-upload/runtime-policy.ts +++ b/src/adapters/browser-transfer/resumable-upload/runtime-policy.ts @@ -13,6 +13,12 @@ export type ResumableUploadRuntimePolicy = Readonly<{ capabilityRefreshSkewMs: number; maxSessionLifetimeMs: number; providerAttemptTimeoutMs: number; + /** + * TR-RR-06. The bound `dispose()` applies to its drain. A non-cooperative + * mutation lock or provider would otherwise make teardown unbounded, so a + * caller could never learn whether the runtime was quiescent. + */ + cleanupDeadlineMs: number; }>; const MIB = 1024 * 1024; @@ -49,6 +55,7 @@ const DEFAULT_POLICY: ResumableUploadRuntimePolicy = Object.freeze({ capabilityRefreshSkewMs: 5_000, maxSessionLifetimeMs: 24 * 60 * 60_000, providerAttemptTimeoutMs: 30_000, + cleanupDeadlineMs: 10_000, }); export function resolveResumableUploadRuntimePolicy( @@ -95,6 +102,9 @@ export function resolveResumableUploadRuntimePolicy( !positiveSafeInteger(policy.providerAttemptTimeoutMs) || policy.providerAttemptTimeoutMs > ABSOLUTE_LIMITS.maxProviderAttemptTimeoutMs || + !positiveSafeInteger(policy.cleanupDeadlineMs) || + policy.cleanupDeadlineMs > + ABSOLUTE_LIMITS.maxProviderAttemptTimeoutMs * 2 || Math.ceil(policy.maxFileBytes / policy.partSizeBytes) > policy.maxPartCount ) { diff --git a/tests/unit/image-cdn-runtime.test.ts b/tests/unit/image-cdn-runtime.test.ts index 89229d5..1f26b3b 100644 --- a/tests/unit/image-cdn-runtime.test.ts +++ b/tests/unit/image-cdn-runtime.test.ts @@ -1006,7 +1006,13 @@ describe("production image CDN runtime", () => { expect(abortDeadline.clearTimeout).toHaveBeenCalledOnce(); }); - it("bounds concurrent capability verification and releases the slot after abort", async () => { + /** + * TR-RR-07. The concurrency cap exists to bound *physical* verification work. + * Releasing the slot when the wrapper's abort resolved let an abandoned + * verifier keep running while a new one was admitted, so repeated aborts + * produced more concurrent work than the configured cap allows. + */ + it("holds the verification slot until the raw verifier settles", async () => { const preset = imageCdnPresetReference( "verification-concurrency", "bound-image-verification-concurrency", @@ -1018,10 +1024,13 @@ describe("production image CDN runtime", () => { }, }); let verificationAttempt = 0; + let releaseFirst: ((value: boolean) => void) | undefined; const verify = vi.fn(() => { verificationAttempt += 1; return verificationAttempt === 1 - ? new Promise(() => undefined) + ? new Promise((resolve) => { + releaseFirst = resolve; + }) : Promise.resolve(true); }); const runtime = createImageCdnRuntime({ @@ -1054,9 +1063,24 @@ describe("production image CDN runtime", () => { ok: false, error: { code: "ABORTED" }, }); + + // The caller's wait ended, but the raw verifier has not. Admitting a second + // one here would put two physical verifications under a cap of one. await expect( runtime.assets.acceptBackendIssued(issued), - ).resolves.toMatchObject({ ok: true }); + ).resolves.toMatchObject({ + ok: false, + error: { code: "LIMIT_EXCEEDED" }, + }); + expect(verify).toHaveBeenCalledOnce(); + + // Once the raw verifier settles the slot is free again. + releaseFirst?.(true); + await vi.waitFor(async () => { + await expect( + runtime.assets.acceptBackendIssued(issued), + ).resolves.toMatchObject({ ok: true }); + }); expect(verify).toHaveBeenCalledTimes(2); }); diff --git a/tests/unit/presigned-transfer.test.ts b/tests/unit/presigned-transfer.test.ts index e7c6eb8..fd2eac0 100644 --- a/tests/unit/presigned-transfer.test.ts +++ b/tests/unit/presigned-transfer.test.ts @@ -1975,6 +1975,114 @@ describe("presigned transfer", () => { }); expect(written).toEqual([...bytes]); }); + + /** + * TR-RR-04. A presigned byte source owns a fetch reader and a capability + * lease, and its port requires `close()`. The delivery consumer never called + * it, so every outcome — success, validation failure, writer failure and + * abort — leaked both. + */ + it.each([ + { label: "success", mode: "SUCCESS" as const }, + { label: "writer failure", mode: "WRITER_FAILURE" as const }, + { label: "abort", mode: "ABORT" as const }, + ])("closes the presigned source exactly once on $label", async ({ mode }) => { + const bytes = new Uint8Array([1, 2, 3]); + let closes = 0; + const controller = new AbortController(); + const source = { + byteLength: bytes.byteLength, + integrity: "VERIFIED_ON_SUCCESSFUL_EXHAUSTION" as const, + capability: undefined as never, + close() { + closes += 1; + }, + async *stream() { + if (mode === "ABORT") controller.abort(); + yield { ok: true as const, value: bytes }; + }, + }; + const capability = Object.freeze({ + capabilityReceipt: "capability-close-1", + method: "GET" as const, + binding: { kind: "DOWNLOAD" as const, resourceId: "resource-1" }, + mediaType: "application/octet-stream", + byteLength: bytes.byteLength, + maxBytes: bytes.byteLength, + expectedSha256: "a".repeat(64), + expiresAtEpochMs: NOW + 60_000, + }); + source.capability = capability as never; + + const closePolicy = browserFilePolicyReference( + "download", + "presigned-close", + ); + const policies = new BrowserFilePolicyRegistry({ + profiles: [ + { + reference: closePolicy, + download: { + strategy: "PROMPT_AND_STREAM", + mediaType: "application/octet-stream", + safeExtension: ".bin", + maxTransferBytes: 64, + maxBufferedBytes: 8, + integrity: "REQUIRED", + }, + }, + ], + hardLimits: { + maxInspectionBytes: 64, + maxRetainedFileBytes: 64, + maxPreviewBytes: 64, + maxObjectUrlBytes: 64, + maxTransferBytes: 64, + }, + }); + const handle: SaveFileHandle = { + async createWritable() { + return new WritableStream({ + write() { + if (mode === "WRITER_FAILURE") { + throw new TypeError("writer exploded"); + } + }, + }); + }, + }; + const downloads = createDownloadDeliveryAdapter({ + host: { handoff() {} }, + policies, + hardMaxObjectUrlBytes: 64, + hardMaxTransferBytes: 64, + browserManagedCapabilities: { + resolve() { + throw new TypeError("not used"); + }, + }, + openAuthorizedSource: async () => + ({ ok: true, value: source }) as never, + showSaveFilePicker: async () => handle, + userActivation: { isActive: true }, + now: () => NOW, + }); + + const deliveryResult = await downloads.deliver({ + policy: closePolicy, + source: { + kind: "AUTHORIZED_STREAM_RESOURCE", + resourceId: "resource-1", + capability: capability as never, + }, + suggestedFileName: "artifact.bin", + signal: controller.signal, + onProgress() {}, + }); + + void deliveryResult; + expect(closes).toBe(1); + }); }); /** diff --git a/tests/unit/resumable-upload-runtime.test.ts b/tests/unit/resumable-upload-runtime.test.ts index c9303f7..d30dce3 100644 --- a/tests/unit/resumable-upload-runtime.test.ts +++ b/tests/unit/resumable-upload-runtime.test.ts @@ -1280,3 +1280,66 @@ describe("production resumable upload runtime", () => { } }); }); +/** + * TR-RR-06. A non-cooperative mutation lock or provider must not make teardown + * unbounded: `dispose()` bounds its drain and reports honestly when the runtime + * is still CLOSING, and an abort is admitted physical work it cannot step over. + */ +describe("TR-RR-06 bounded resumable teardown", () => { + it("reports an unproved drain instead of waiting forever", async () => { + const checkpoints = new MemoryCheckpointStore(); + const harness = createControlHarness(); + const runtime = createResumableUploadRuntime({ + controlPlane: harness.control, + partExecutor: executorFor(harness, { delay: async () => {} }), + checkpoints, + // A lock that never grants: dispose must still be bounded. + mutationLock: Object.freeze({ + async run(): Promise { + return await new Promise(() => {}); + }, + }), + crypto, + policy: runtimePolicy({ cleanupDeadlineMs: 20 }), + now: () => 1_000, + random: () => 0, + sleep: async () => {}, + }); + + void runtime.upload({ + uploadKey: "upload_key_hung", + purpose: "attachment", + mediaType: "application/octet-stream", + source: byteStreamSource(new Uint8Array([1, 2, 3, 4, 5])), + signal: activeSignal, + }); + await Promise.resolve(); + await Promise.resolve(); + + const disposed = await runtime.dispose(); + expect(disposed.ok).toBe(false); + // Still CLOSING: physical work the caller must not treat as finished. + expect(runtime.lifecycle()).toBe("CLOSING"); + expect(checkpoints.closed).toBe(false); + }); + + it("closes once every admitted operation settles", async () => { + const checkpoints = new MemoryCheckpointStore(); + const harness = createControlHarness(); + const runtime = createResumableUploadRuntime({ + controlPlane: harness.control, + partExecutor: executorFor(harness, { delay: async () => {} }), + checkpoints, + mutationLock: noContentionLock, + crypto, + policy: runtimePolicy({ cleanupDeadlineMs: 200 }), + now: () => 1_000, + random: () => 0, + sleep: async () => {}, + }); + const disposed = await runtime.dispose(); + expect(disposed).toMatchObject({ ok: true }); + expect(runtime.lifecycle()).toBe("CLOSED"); + expect(checkpoints.closed).toBe(true); + }); +});