diff --git a/docs/operations/adapter-remediation-ledger.md b/docs/operations/adapter-remediation-ledger.md index 7ebaa3d..86e9f0d 100644 --- a/docs/operations/adapter-remediation-ledger.md +++ b/docs/operations/adapter-remediation-ledger.md @@ -126,7 +126,7 @@ Rollout state starts at `NOT_STARTED`; documented-unimplemented items start at | BT-UP-03 | Resumable checkpoint store | `corepack pnpm exec vitest run tests/unit/resumable-upload-checkpoint.test.ts` | `fix: report unknown IndexedDB delete effects` | `FIXED_NOT_RELEASED` | pending-delete registry growth | Red blocked-deadline case → green `PENDING`/`UNKNOWN`; a realm-scoped registry blocks recreating the partition | | BT-UP-04 | Presigned part executor | `corepack pnpm exec vitest run tests/unit/resumable-upload-checkpoint.test.ts` | `fix: harden resumable upload transport contracts` | `FIXED_NOT_RELEASED` | expiry check rejection | Non-finite and negative clocks return `UNAVAILABLE`/`RESUME` instead of bypassing expiry | | BT-UP-05 | Refactor | `corepack pnpm exec vitest run tests/unit/resumable-upload-runtime.test.ts` | — | `NOT_STARTED` | characterization drift | — | -| BT-UP-06 | Refactor | `corepack pnpm exec vitest run tests/unit/resumable-upload-runtime.test.ts` | — | `NOT_STARTED` | drain not quiescent | — | +| BT-UP-06 | Refactor | `corepack pnpm exec vitest run tests/unit/resumable-upload-runtime.test.ts` | `fix: drain resumable upload teardown` | `FIXED_NOT_RELEASED` | drain not quiescent | Red single-flight dispose case → green 18/18; `close()` closes admission and starts the same drain, `dispose()` aborts the active-operation registry and awaits real settlement before closing the checkpoint store | | BT-UP-07 | Documented gap (Web Locks matrix) | promotion evidence | — | `PROMOTION_BLOCKED` | n/a | — | | BT-IMG-01 | Type-contract change | `corepack pnpm check:types:test` fixture | `fix: complete presigned capability and upload transport contracts` | `FIXED_NOT_RELEASED` | caller compile break | `resolve()` now requires the lifetime signal; `tests/fixtures/typecheck/invalid-image-cdn-resolve-signal.ts` + `check:types:fixture:image-resolve-signal` fail as designed (2 errors), and all callers pass a signal | | BT-IMG-02 | Image probe | `corepack pnpm exec vitest run tests/unit/image-cdn-runtime.test.ts` | `fix: parse Cache-Control with quote awareness` | `FIXED_NOT_RELEASED` | Cache-Control parse rejection | Red unmatched-quote cases → green 25/25 | 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 3ce70f4..3ce9e1c 100644 --- a/src/adapters/browser-transfer/resumable-upload/resumable-upload-runtime.ts +++ b/src/adapters/browser-transfer/resumable-upload/resumable-upload-runtime.ts @@ -66,7 +66,19 @@ import type { UploadMutationLock } from "./upload-mutation-lock.ts"; export type ResumableUploadRuntime = ResumableUploadPort & Readonly<{ + /** + * BT-UP-06. Compatibility facade: closes admission and starts the same + * single-flight drain that `dispose()` awaits. + */ close(): void; + /** + * BT-UP-06. Awaitable teardown for a future composition owner. It shares + * one drain promise, aborts the active operation registry and only then + * closes the checkpoint store and cancellation channel, so success actually + * means quiescent. No current bootstrap consumer is assumed. + */ + dispose(): Promise; + lifecycle(): "OPEN" | "CLOSING" | "CLOSED"; }>; export type ResumableUploadRuntimeDependencies = Readonly<{ @@ -162,7 +174,11 @@ export function createResumableUploadRuntime( const dependencies = snapshotDependencies(inputDependencies); const lifetime = new AbortController(); const localUploads = new Map>(); + /** BT-UP-06. Terminal settlement of every admitted operation. */ + const activeOperations = new Set>(); let closed = false; + let lifecycle: "OPEN" | "CLOSING" | "CLOSED" = "OPEN"; + let drain: Promise | null = null; const cancelLocalUploads = (uploadKey: string): void => { if (!SAFE_UPLOAD_KEY.test(uploadKey)) return; for (const controller of localUploads.get(uploadKey) ?? []) { @@ -211,19 +227,24 @@ export function createResumableUploadRuntime( } return browserDataFailure("ABORTED", "UPLOAD_SESSION"); } + const operation = dependencies.mutationLock.run( + request.uploadKey, + operationScope.signal, + async () => + await executeUpload( + dependencies, + Object.freeze({ + ...request, + signal: operationScope.signal, + }), + ), + ); + // Tracked until terminal settlement so dispose() can prove quiescence. + const tracked = Promise.resolve(operation).catch(() => undefined); + activeOperations.add(tracked); + void tracked.finally(() => activeOperations.delete(tracked)); try { - return await dependencies.mutationLock.run( - request.uploadKey, - operationScope.signal, - async () => - await executeUpload( - dependencies, - Object.freeze({ - ...request, - signal: operationScope.signal, - }), - ), - ); + return await operation; } catch (error) { return mapLockFailure(error, "UPLOAD_SESSION"); } finally { @@ -278,14 +299,36 @@ export function createResumableUploadRuntime( }, close() { - if (closed) return; + // BT-UP-06. Admission closes synchronously; the drain runs behind the + // same single-flight promise dispose() returns. + void startDrain(); + }, + + dispose(): Promise { + return startDrain(); + }, + + lifecycle: () => lifecycle, + }); + + function startDrain(): Promise { + drain ??= (async () => { closed = true; + lifecycle = "CLOSING"; releaseCrossContextCancellation?.(); dependencies.crossContextCancellation?.close(); + // Abort every admitted operation, then wait for their real settlement. lifetime.abort(); + for (const controllers of localUploads.values()) { + for (const controller of controllers) controller.abort(); + } + await Promise.allSettled([...activeOperations]); + // The store closes only after nothing can still write a checkpoint. dependencies.checkpoints.close(); - }, - }); + lifecycle = "CLOSED"; + })(); + return drain; + } return runtime; } diff --git a/tests/unit/resumable-upload-runtime.test.ts b/tests/unit/resumable-upload-runtime.test.ts index dfbbc48..c9303f7 100644 --- a/tests/unit/resumable-upload-runtime.test.ts +++ b/tests/unit/resumable-upload-runtime.test.ts @@ -347,6 +347,65 @@ function executorFor( } describe("production resumable upload runtime", () => { + it("disposes through one drain that proves quiescence", async () => { + // BT-UP-06. close() closes admission; dispose() awaits real settlement. + const checkpoints = new MemoryCheckpointStore(); + const harness = createControlHarness(); + let releaseUpload: (() => void) | undefined; + const runtime = createResumableUploadRuntime({ + controlPlane: harness.control, + partExecutor: executorFor(harness, { + delay: async () => { + await new Promise((resolve) => { + releaseUpload = resolve; + }); + }, + }), + checkpoints, + mutationLock: noContentionLock, + crypto, + policy: runtimePolicy(), + now: () => 1_000, + random: () => 0, + sleep: async () => {}, + }); + + const uploading = runtime.upload({ + uploadKey: "upload_key_01", + purpose: "attachment", + mediaType: "application/octet-stream", + source: byteStreamSource(new Uint8Array([1, 2, 3, 4, 5])), + signal: activeSignal, + }); + await vi.waitFor(() => expect(releaseUpload).toBeDefined()); + expect(runtime.lifecycle()).toBe("OPEN"); + + const first = runtime.dispose(); + const second = runtime.dispose(); + // Duplicate dispose is single-flight. + expect(first).toBe(second); + expect(runtime.lifecycle()).toBe("CLOSING"); + + // New admission is refused while draining. + await expect( + runtime.upload({ + uploadKey: "upload_key_02", + purpose: "attachment", + mediaType: "application/octet-stream", + source: byteStreamSource(new Uint8Array([1])), + signal: activeSignal, + }), + ).resolves.toMatchObject({ ok: false, error: { code: "UNAVAILABLE" } }); + // The checkpoint store cannot close before the operation settles. + expect(checkpoints.closed).toBe(false); + + releaseUpload?.(); + await first; + await uploading; + expect(runtime.lifecycle()).toBe("CLOSED"); + expect(checkpoints.closed).toBe(true); + }); + it("has a valid production default policy", () => { expect(() => resolveResumableUploadRuntimePolicy()).not.toThrow(); expect(() =>