fix: drain resumable upload teardown
BT-UP-06: close() previously aborted the lifetime and closed the checkpoint store immediately, so a caller could not wait for an active operation's terminal settlement and a late provider result could still race the store. The runtime now tracks every admitted operation until it settles. close() stays the compatibility facade that closes admission and starts the drain, while dispose() returns that same single-flight promise: it aborts the operation registry, awaits actual settlement, and only then closes the checkpoint store and cancellation channel. lifecycle() exposes OPEN, CLOSING and CLOSED, and a draining runtime refuses new admission. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5
parent
000a2581af
commit
78f1bb273e
@@ -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<void>;
|
||||
lifecycle(): "OPEN" | "CLOSING" | "CLOSED";
|
||||
}>;
|
||||
|
||||
export type ResumableUploadRuntimeDependencies<Capability> = Readonly<{
|
||||
@@ -162,7 +174,11 @@ export function createResumableUploadRuntime<Capability>(
|
||||
const dependencies = snapshotDependencies(inputDependencies);
|
||||
const lifetime = new AbortController();
|
||||
const localUploads = new Map<string, Set<AbortController>>();
|
||||
/** BT-UP-06. Terminal settlement of every admitted operation. */
|
||||
const activeOperations = new Set<Promise<unknown>>();
|
||||
let closed = false;
|
||||
let lifecycle: "OPEN" | "CLOSING" | "CLOSED" = "OPEN";
|
||||
let drain: Promise<void> | 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<Capability>(
|
||||
}
|
||||
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<Capability>(
|
||||
},
|
||||
|
||||
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<void> {
|
||||
return startDrain();
|
||||
},
|
||||
|
||||
lifecycle: () => lifecycle,
|
||||
});
|
||||
|
||||
function startDrain(): Promise<void> {
|
||||
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;
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user