From 5a76f952918559b271282abdb695999eecf918c6 Mon Sep 17 00:00:00 2001 From: DongHyeonka Date: Fri, 14 Aug 2026 17:22:45 +0900 Subject: [PATCH] fix: bound resumable teardown, image concurrency and delivery leases MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit TR-RR-06. dispose() now bounds its drain with a cleanupDeadlineMs from policy and returns the result, so a non-cooperative mutation lock or provider can no longer make teardown unbounded and an unproved drain is reported as still CLOSING instead of closed over. The checkpoint store stays open in that case, because something can still write to it. An abort is admitted physical work like an upload, so it joins the tracked set rather than being stepped over. TR-RR-07. The verification slot belongs to the raw verifier, not the wrapper. Releasing it when the caller's wait expired let an abandoned verification keep running while a new one was admitted, so repeated aborts produced more concurrent physical work than the configured cap allows. The slot is now released only once the raw tasks settle. 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. The closeable subtype is lost in the FileByteSource projection, so a holder keeps it from the moment the lease exists and the outermost finally closes it exactly once — on success, validation failure, writer failure and abort alike. check:adapter-inventory now also fails if the shared abortable-operation primitive has no production importers. It was safe to add only once the presigned subsystems actually migrated onto it; a gate that fails CI for a documented, unfixed defect reports the wrong thing. Co-Authored-By: Claude Opus 5 (1M context) --- scripts/check-adapter-inventory.ts | 21 ++++ .../download-delivery-adapter.ts | 43 +++++++ .../image-cdn/image-cdn-runtime.ts | 29 +++-- .../resumable-upload-runtime.ts | 59 +++++++--- .../resumable-upload/runtime-policy.ts | 10 ++ tests/unit/image-cdn-runtime.test.ts | 30 ++++- tests/unit/presigned-transfer.test.ts | 108 ++++++++++++++++++ tests/unit/resumable-upload-runtime.test.ts | 63 ++++++++++ 8 files changed, 336 insertions(+), 27 deletions(-) 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); + }); +});