DongHyeonkaandClaude Opus 5
5a76f95291
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) <noreply@anthropic.com>
1346 lines
41 KiB
TypeScript
1346 lines
41 KiB
TypeScript
import { describe, expect, it, vi } from "vitest";
|
|
|
|
import type {
|
|
PresignedUploadPartCapability,
|
|
PresignedUploadPartPort,
|
|
} from "../../src/application/ports/browser-transfer/presigned-transfer.ts";
|
|
import type {
|
|
ResumableUploadCheckpoint,
|
|
ResumableUploadCheckpointStore,
|
|
ResumableUploadControlPlane,
|
|
ResumableUploadSource,
|
|
UploadPartExecutor,
|
|
UploadPartReceipt,
|
|
UploadProviderResult,
|
|
UploadSession,
|
|
} from "../../src/application/ports/browser-transfer/resumable-upload.ts";
|
|
import { RESUMABLE_UPLOAD_PROTOCOL } from "../../src/application/ports/browser-transfer/resumable-upload.ts";
|
|
import type {
|
|
BrowserDataObservation,
|
|
BrowserDataResult,
|
|
} from "../../src/application/ports/browser-file-storage/shared.ts";
|
|
import {
|
|
browserDataFailure,
|
|
browserDataSuccess,
|
|
} from "../../src/adapters/browser-file-storage/result.ts";
|
|
import { createPresignedUploadPartExecutor } from "../../src/adapters/browser-transfer/resumable-upload/presigned-upload-part-executor.ts";
|
|
import { createResumableUploadRuntime } from "../../src/adapters/browser-transfer/resumable-upload/resumable-upload-runtime.ts";
|
|
import { resolveResumableUploadRuntimePolicy } from "../../src/adapters/browser-transfer/resumable-upload/runtime-policy.ts";
|
|
import type {
|
|
UploadCancellationChannel,
|
|
UploadCancellationListener,
|
|
} from "../../src/adapters/browser-transfer/resumable-upload/upload-cancellation-channel.ts";
|
|
import type { UploadMutationLock } from "../../src/adapters/browser-transfer/resumable-upload/upload-mutation-lock.ts";
|
|
|
|
type TestCapability = Readonly<{ id: string }>;
|
|
|
|
const activeSignal = new AbortController().signal;
|
|
const noContentionLock: UploadMutationLock = Object.freeze({
|
|
async run<Value>(
|
|
_uploadKey: string,
|
|
_signal: AbortSignal,
|
|
task: () => Promise<Value>,
|
|
): Promise<Value> {
|
|
return await task();
|
|
},
|
|
});
|
|
|
|
function createSerialMutationLock(): UploadMutationLock {
|
|
let tail = Promise.resolve();
|
|
return Object.freeze({
|
|
run<Value>(
|
|
_uploadKey: string,
|
|
signal: AbortSignal,
|
|
task: () => Promise<Value>,
|
|
): Promise<Value> {
|
|
const result = tail.then(async () => {
|
|
if (signal.aborted) {
|
|
throw new DOMException(
|
|
"The operation was aborted.",
|
|
"AbortError",
|
|
);
|
|
}
|
|
return await task();
|
|
});
|
|
tail = result.then(
|
|
() => undefined,
|
|
() => undefined,
|
|
);
|
|
return result;
|
|
},
|
|
});
|
|
}
|
|
|
|
function createMemoryCancellationPair(): readonly [
|
|
UploadCancellationChannel,
|
|
UploadCancellationChannel,
|
|
] {
|
|
const listeners = [
|
|
new Set<UploadCancellationListener>(),
|
|
new Set<UploadCancellationListener>(),
|
|
] as const;
|
|
const channels = listeners.map((ownListeners, ownIndex) => {
|
|
let closed = false;
|
|
return Object.freeze({
|
|
publish(uploadKey: string) {
|
|
if (closed) return false;
|
|
for (const [index, peerListeners] of listeners.entries()) {
|
|
if (index === ownIndex) continue;
|
|
for (const listener of [...peerListeners]) {
|
|
listener(uploadKey);
|
|
}
|
|
}
|
|
return true;
|
|
},
|
|
subscribe(listener: UploadCancellationListener) {
|
|
if (closed) throw new TypeError("closed");
|
|
ownListeners.add(listener);
|
|
return () => ownListeners.delete(listener);
|
|
},
|
|
close() {
|
|
closed = true;
|
|
ownListeners.clear();
|
|
},
|
|
});
|
|
});
|
|
return channels as unknown as readonly [
|
|
UploadCancellationChannel,
|
|
UploadCancellationChannel,
|
|
];
|
|
}
|
|
|
|
class MemoryCheckpointStore implements ResumableUploadCheckpointStore {
|
|
readonly rows = new Map<string, ResumableUploadCheckpoint>();
|
|
closed = false;
|
|
|
|
async read(
|
|
uploadKey: string,
|
|
): Promise<BrowserDataResult<ResumableUploadCheckpoint | null>> {
|
|
return browserDataSuccess(
|
|
structuredClone(this.rows.get(uploadKey) ?? null),
|
|
);
|
|
}
|
|
|
|
async compareAndSwap(
|
|
input: Parameters<
|
|
ResumableUploadCheckpointStore["compareAndSwap"]
|
|
>[0],
|
|
): Promise<BrowserDataResult<ResumableUploadCheckpoint>> {
|
|
const current = this.rows.get(input.checkpoint.uploadKey);
|
|
if (
|
|
(input.expectedRevision === null && current) ||
|
|
(input.expectedRevision !== null &&
|
|
current?.revision !== input.expectedRevision)
|
|
) {
|
|
return browserDataFailure("CONFLICT", "UPLOAD_RECONCILE", {
|
|
recovery: "RECONCILE",
|
|
});
|
|
}
|
|
const snapshot = structuredClone(input.checkpoint);
|
|
this.rows.set(snapshot.uploadKey, snapshot);
|
|
return browserDataSuccess(snapshot);
|
|
}
|
|
|
|
async remove(
|
|
input: Parameters<ResumableUploadCheckpointStore["remove"]>[0],
|
|
): Promise<BrowserDataResult<void>> {
|
|
const current = this.rows.get(input.uploadKey);
|
|
if (current?.revision !== input.expectedRevision) {
|
|
return browserDataFailure("CONFLICT", "UPLOAD_RECONCILE", {
|
|
recovery: "RECONCILE",
|
|
});
|
|
}
|
|
this.rows.delete(input.uploadKey);
|
|
return browserDataSuccess(undefined);
|
|
}
|
|
|
|
close(): void {
|
|
this.closed = true;
|
|
}
|
|
}
|
|
|
|
function rangeSource(bytes: Uint8Array): ResumableUploadSource {
|
|
return Object.freeze({
|
|
kind: "RANGE_READER" as const,
|
|
reader: Object.freeze({
|
|
byteLength: bytes.byteLength,
|
|
async readRange(input: Readonly<{
|
|
offset: number;
|
|
length: number;
|
|
signal: AbortSignal;
|
|
}>) {
|
|
if (input.signal.aborted) {
|
|
return browserDataFailure("ABORTED", "FILE_READ");
|
|
}
|
|
return browserDataSuccess(
|
|
bytes.slice(input.offset, input.offset + input.length),
|
|
);
|
|
},
|
|
}),
|
|
});
|
|
}
|
|
|
|
function byteStreamSource(bytes: Uint8Array): ResumableUploadSource {
|
|
return Object.freeze({
|
|
kind: "FILE_BYTE_SOURCE" as const,
|
|
bytes: Object.freeze({
|
|
byteLength: bytes.byteLength,
|
|
async *stream(signal: AbortSignal) {
|
|
if (signal.aborted) {
|
|
yield browserDataFailure("ABORTED", "FILE_READ");
|
|
return;
|
|
}
|
|
yield browserDataSuccess(bytes.slice(0, 3));
|
|
yield browserDataSuccess(bytes.slice(3));
|
|
},
|
|
}),
|
|
});
|
|
}
|
|
|
|
function runtimePolicy(
|
|
overrides: Partial<
|
|
Parameters<typeof resolveResumableUploadRuntimePolicy>[0]
|
|
> = {},
|
|
) {
|
|
return {
|
|
partSizeBytes: 4,
|
|
maxFileBytes: 100,
|
|
maxPartCount: 25,
|
|
maxConcurrency: 3,
|
|
maxInFlightBytes: 48,
|
|
partBufferCopyFactor: 4,
|
|
maxSourceChunkBytes: 8,
|
|
maxRetries: 2,
|
|
retryBaseDelayMs: 1,
|
|
retryMaxDelayMs: 10,
|
|
maxRetryAfterMs: 100,
|
|
capabilityRefreshSkewMs: 5,
|
|
maxSessionLifetimeMs: 10_000,
|
|
providerAttemptTimeoutMs: 100,
|
|
...overrides,
|
|
};
|
|
}
|
|
|
|
type ControlHarness = Readonly<{
|
|
control: ResumableUploadControlPlane<TestCapability>;
|
|
accepted: Map<number, UploadPartReceipt>;
|
|
issued: ReturnType<typeof vi.fn>;
|
|
completedParts: UploadPartReceipt[][];
|
|
getSession(): UploadSession | null;
|
|
}>;
|
|
|
|
function createControlHarness(options: Readonly<{
|
|
now?: number;
|
|
serverMaxConcurrency?: number;
|
|
sessionId?: (createIndex: number) => string;
|
|
statusParts?: (
|
|
session: UploadSession,
|
|
accepted: Map<number, UploadPartReceipt>,
|
|
) => readonly UploadPartReceipt[];
|
|
issueCapability?: (
|
|
input: Parameters<
|
|
ResumableUploadControlPlane<TestCapability>["issuePartCapability"]
|
|
>[0],
|
|
callIndex: number,
|
|
) => UploadProviderResult<Readonly<{
|
|
capability: TestCapability;
|
|
uploadBindingSha256: string;
|
|
expiresAtEpochMs: number;
|
|
}>>;
|
|
}> = {}): ControlHarness {
|
|
const now = options.now ?? 1_000;
|
|
const accepted = new Map<number, UploadPartReceipt>();
|
|
const completedParts: UploadPartReceipt[][] = [];
|
|
let session: UploadSession | null = null;
|
|
let createCount = 0;
|
|
let issueCount = 0;
|
|
const issued = vi.fn();
|
|
const control: ResumableUploadControlPlane<TestCapability> = {
|
|
async createSession(input) {
|
|
createCount += 1;
|
|
session = Object.freeze({
|
|
protocol: RESUMABLE_UPLOAD_PROTOCOL,
|
|
sessionId:
|
|
options.sessionId?.(createCount) ?? "session_01",
|
|
requestBindingSha256: input.requestBindingSha256,
|
|
fingerprint: input.fingerprint,
|
|
partSizeBytes: input.requestedPartSizeBytes,
|
|
partCount: input.fingerprint.partCount,
|
|
maxConcurrency: options.serverMaxConcurrency ?? 2,
|
|
expiresAtEpochMs: now + 5_000,
|
|
});
|
|
return browserDataSuccess(session);
|
|
},
|
|
async getStatus() {
|
|
if (!session) {
|
|
return browserDataFailure("NOT_FOUND", "UPLOAD_RECONCILE");
|
|
}
|
|
const parts =
|
|
options.statusParts?.(session, accepted) ??
|
|
[...accepted.values()].sort(
|
|
(left, right) => left.partNumber - right.partNumber,
|
|
);
|
|
return browserDataSuccess({
|
|
state: "ACTIVE",
|
|
session,
|
|
acceptedParts: parts,
|
|
});
|
|
},
|
|
async issuePartCapability(input) {
|
|
issueCount += 1;
|
|
issued(input);
|
|
return (
|
|
options.issueCapability?.(input, issueCount) ??
|
|
browserDataSuccess({
|
|
capability: Object.freeze({ id: `cap-${issueCount}` }),
|
|
uploadBindingSha256: input.uploadBindingSha256,
|
|
expiresAtEpochMs: now + 4_000,
|
|
})
|
|
);
|
|
},
|
|
async complete(input) {
|
|
completedParts.push([...input.orderedParts]);
|
|
return browserDataSuccess({
|
|
state: "QUARANTINED",
|
|
protocol: RESUMABLE_UPLOAD_PROTOCOL,
|
|
sessionId: input.sessionId,
|
|
requestBindingSha256: input.requestBindingSha256,
|
|
fingerprint: input.fingerprint,
|
|
resourceId: "resource_01",
|
|
});
|
|
},
|
|
async abort() {
|
|
return browserDataSuccess({ state: "ABORTED" });
|
|
},
|
|
};
|
|
return {
|
|
control,
|
|
accepted,
|
|
issued,
|
|
completedParts,
|
|
getSession: () => session,
|
|
};
|
|
}
|
|
|
|
function executorFor(
|
|
harness: ControlHarness,
|
|
options: Readonly<{
|
|
delay?: () => Promise<void>;
|
|
onActive?: (active: number) => void;
|
|
}> = {},
|
|
): UploadPartExecutor<TestCapability> {
|
|
let active = 0;
|
|
return {
|
|
async uploadPart(input) {
|
|
active += 1;
|
|
options.onActive?.(active);
|
|
await options.delay?.();
|
|
active -= 1;
|
|
const receipt = Object.freeze({
|
|
...input.part,
|
|
receiptToken: `etag-part-${input.part.partNumber}`,
|
|
});
|
|
harness.accepted.set(input.part.partNumber, receipt);
|
|
return browserDataSuccess(receipt);
|
|
},
|
|
};
|
|
}
|
|
|
|
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<void>((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(() =>
|
|
resolveResumableUploadRuntimePolicy(
|
|
runtimePolicy({ maxInFlightBytes: 15 }),
|
|
),
|
|
).toThrow(TypeError);
|
|
});
|
|
|
|
it("uploads bounded range parts, obeys the server concurrency ceiling and completes as quarantined", async () => {
|
|
const checkpoints = new MemoryCheckpointStore();
|
|
const harness = createControlHarness({ serverMaxConcurrency: 1 });
|
|
let maxActive = 0;
|
|
const observations: BrowserDataObservation[] = [];
|
|
const runtime = createResumableUploadRuntime({
|
|
controlPlane: harness.control,
|
|
partExecutor: executorFor(harness, {
|
|
delay: async () => await Promise.resolve(),
|
|
onActive(active) {
|
|
maxActive = Math.max(maxActive, active);
|
|
},
|
|
}),
|
|
checkpoints,
|
|
mutationLock: noContentionLock,
|
|
crypto,
|
|
policy: runtimePolicy(),
|
|
now: () => 1_000,
|
|
random: () => 0,
|
|
sleep: async () => {},
|
|
observer: {
|
|
record(observation) {
|
|
observations.push(observation);
|
|
},
|
|
},
|
|
});
|
|
|
|
const result = await runtime.upload({
|
|
uploadKey: "upload_key_01",
|
|
purpose: "attachment",
|
|
mediaType: "application/octet-stream",
|
|
source: rangeSource(
|
|
new Uint8Array([1, 2, 3, 4, 5, 6, 7, 8, 9]),
|
|
),
|
|
signal: activeSignal,
|
|
});
|
|
|
|
expect(result).toMatchObject({
|
|
ok: true,
|
|
value: {
|
|
state: "QUARANTINED",
|
|
resourceId: "resource_01",
|
|
byteLength: 9,
|
|
},
|
|
});
|
|
if (result.ok) {
|
|
expect(Object.keys(result.value).sort()).toEqual([
|
|
"byteLength",
|
|
"replayed",
|
|
"resourceId",
|
|
"state",
|
|
]);
|
|
}
|
|
expect(maxActive).toBe(1);
|
|
expect(harness.completedParts[0]?.map((part) => part.partNumber)).toEqual(
|
|
[1, 2, 3],
|
|
);
|
|
expect(checkpoints.rows.size).toBe(0);
|
|
expect(
|
|
observations.map((observation) => observation.operation),
|
|
).toEqual(
|
|
expect.arrayContaining([
|
|
"UPLOAD_SESSION",
|
|
"UPLOAD_RECONCILE",
|
|
"UPLOAD_PART",
|
|
"UPLOAD_COMPLETE",
|
|
]),
|
|
);
|
|
const serialized = JSON.stringify(observations);
|
|
expect(serialized).not.toContain("upload_key_01");
|
|
expect(serialized).not.toContain("session_01");
|
|
expect(serialized).not.toContain("etag-part");
|
|
});
|
|
|
|
it("supports replayable FileByteSource without retaining the whole file", async () => {
|
|
const checkpoints = new MemoryCheckpointStore();
|
|
const harness = createControlHarness();
|
|
const runtime = createResumableUploadRuntime({
|
|
controlPlane: harness.control,
|
|
partExecutor: executorFor(harness),
|
|
checkpoints,
|
|
mutationLock: noContentionLock,
|
|
crypto,
|
|
policy: runtimePolicy(),
|
|
now: () => 1_000,
|
|
sleep: async () => {},
|
|
});
|
|
const result = await runtime.upload({
|
|
uploadKey: "upload_key_02",
|
|
purpose: "attachment",
|
|
mediaType: "application/octet-stream",
|
|
source: byteStreamSource(new Uint8Array([1, 2, 3, 4, 5])),
|
|
signal: activeSignal,
|
|
});
|
|
expect(result).toMatchObject({
|
|
ok: true,
|
|
value: { state: "QUARANTINED", byteLength: 5 },
|
|
});
|
|
});
|
|
|
|
it("fails closed when server list-parts disagrees with the current source manifest", async () => {
|
|
const checkpoints = new MemoryCheckpointStore();
|
|
const harness = createControlHarness({
|
|
statusParts(session) {
|
|
return [
|
|
{
|
|
partNumber: 1,
|
|
offset: 0,
|
|
byteLength: Math.min(4, session.fingerprint.byteLength),
|
|
checksumSha256: "f".repeat(64),
|
|
receiptToken: "etag-wrong-part",
|
|
},
|
|
];
|
|
},
|
|
});
|
|
const partExecutor = { uploadPart: vi.fn() };
|
|
const runtime = createResumableUploadRuntime({
|
|
controlPlane: harness.control,
|
|
partExecutor: partExecutor as unknown as UploadPartExecutor<TestCapability>,
|
|
checkpoints,
|
|
mutationLock: noContentionLock,
|
|
crypto,
|
|
policy: runtimePolicy(),
|
|
now: () => 1_000,
|
|
sleep: async () => {},
|
|
});
|
|
const result = await runtime.upload({
|
|
uploadKey: "upload_key_03",
|
|
purpose: "attachment",
|
|
mediaType: "application/octet-stream",
|
|
source: rangeSource(new Uint8Array([1, 2, 3, 4])),
|
|
signal: activeSignal,
|
|
});
|
|
expect(result).toMatchObject({
|
|
ok: false,
|
|
error: {
|
|
code: "INTEGRITY_FAILED",
|
|
operation: "UPLOAD_RECONCILE",
|
|
recovery: "RECONCILE",
|
|
},
|
|
});
|
|
expect(partExecutor.uploadPart).not.toHaveBeenCalled();
|
|
});
|
|
|
|
it.each(["NOT_FOUND", "EXPIRED_RESOURCE"] as const)(
|
|
"removes a definitive %s status checkpoint and creates a fresh session",
|
|
async (code) => {
|
|
const checkpoints = new MemoryCheckpointStore();
|
|
const harness = createControlHarness({
|
|
sessionId: (index) => `session_terminal_${index}`,
|
|
});
|
|
const originalCreate = harness.control.createSession.bind(
|
|
harness.control,
|
|
);
|
|
const createSession = vi.fn(originalCreate);
|
|
harness.control.createSession = createSession;
|
|
let statusCalls = 0;
|
|
harness.control.getStatus = async () => {
|
|
statusCalls += 1;
|
|
const session = harness.getSession();
|
|
if (!session) {
|
|
return browserDataFailure(
|
|
"NOT_FOUND",
|
|
"UPLOAD_RECONCILE",
|
|
);
|
|
}
|
|
if (statusCalls === 1) {
|
|
return browserDataFailure(code, "UPLOAD_RECONCILE", {
|
|
recovery:
|
|
code === "NOT_FOUND" ? "RECONCILE" : "RESTART",
|
|
});
|
|
}
|
|
return browserDataSuccess({
|
|
state: "ACTIVE",
|
|
session,
|
|
acceptedParts: [...harness.accepted.values()],
|
|
});
|
|
};
|
|
const runtime = createResumableUploadRuntime({
|
|
controlPlane: harness.control,
|
|
partExecutor: executorFor(harness),
|
|
checkpoints,
|
|
mutationLock: noContentionLock,
|
|
crypto,
|
|
policy: runtimePolicy(),
|
|
now: () => 1_000,
|
|
sleep: async () => {},
|
|
});
|
|
|
|
const result = await runtime.upload({
|
|
uploadKey: `upload_key_terminal_${code.toLowerCase()}`,
|
|
purpose: "attachment",
|
|
mediaType: "application/octet-stream",
|
|
source: rangeSource(new Uint8Array([1, 2, 3, 4])),
|
|
signal: activeSignal,
|
|
});
|
|
|
|
expect(result.ok).toBe(true);
|
|
expect(createSession).toHaveBeenCalledTimes(2);
|
|
expect(
|
|
createSession.mock.calls[0]?.[0].idempotencyKey,
|
|
).not.toBe(
|
|
createSession.mock.calls[1]?.[0].idempotencyKey,
|
|
);
|
|
expect(checkpoints.rows.size).toBe(0);
|
|
},
|
|
);
|
|
|
|
it("removes the exact checkpoint revision when status becomes terminal after part transfer", async () => {
|
|
const checkpoints = new MemoryCheckpointStore();
|
|
const harness = createControlHarness();
|
|
let statusCalls = 0;
|
|
harness.control.getStatus = async () => {
|
|
statusCalls += 1;
|
|
const session = harness.getSession();
|
|
if (!session) {
|
|
return browserDataFailure("NOT_FOUND", "UPLOAD_RECONCILE");
|
|
}
|
|
return statusCalls === 1
|
|
? browserDataSuccess({
|
|
state: "ACTIVE",
|
|
session,
|
|
acceptedParts: [],
|
|
})
|
|
: browserDataFailure(
|
|
"EXPIRED_RESOURCE",
|
|
"UPLOAD_RECONCILE",
|
|
{ recovery: "RESTART" },
|
|
);
|
|
};
|
|
const runtime = createResumableUploadRuntime({
|
|
controlPlane: harness.control,
|
|
partExecutor: executorFor(harness),
|
|
checkpoints,
|
|
mutationLock: noContentionLock,
|
|
crypto,
|
|
policy: runtimePolicy({ maxRetries: 0 }),
|
|
now: () => 1_000,
|
|
sleep: async () => {},
|
|
});
|
|
|
|
const result = await runtime.upload({
|
|
uploadKey: "upload_key_post_transfer_terminal",
|
|
purpose: "attachment",
|
|
mediaType: "application/octet-stream",
|
|
source: rangeSource(new Uint8Array([1, 2, 3, 4])),
|
|
signal: activeSignal,
|
|
});
|
|
|
|
expect(result).toMatchObject({
|
|
ok: false,
|
|
error: {
|
|
code: "EXPIRED_RESOURCE",
|
|
operation: "UPLOAD_SESSION",
|
|
recovery: "RESTART",
|
|
},
|
|
});
|
|
expect(checkpoints.rows.size).toBe(0);
|
|
expect(harness.completedParts).toHaveLength(0);
|
|
});
|
|
|
|
it("reissues an expiring part capability before sending bytes", async () => {
|
|
const checkpoints = new MemoryCheckpointStore();
|
|
const delays: number[] = [];
|
|
const harness = createControlHarness({
|
|
issueCapability(input, callIndex) {
|
|
return browserDataSuccess({
|
|
capability: { id: `cap-${callIndex}` },
|
|
uploadBindingSha256: input.uploadBindingSha256,
|
|
expiresAtEpochMs: callIndex === 1 ? 1_003 : 4_000,
|
|
});
|
|
},
|
|
});
|
|
const runtime = createResumableUploadRuntime({
|
|
controlPlane: harness.control,
|
|
partExecutor: executorFor(harness),
|
|
checkpoints,
|
|
mutationLock: noContentionLock,
|
|
crypto,
|
|
policy: runtimePolicy({ maxRetries: 1 }),
|
|
now: () => 1_000,
|
|
random: () => 0,
|
|
sleep: async (delay) => {
|
|
delays.push(delay);
|
|
},
|
|
});
|
|
const result = await runtime.upload({
|
|
uploadKey: "upload_key_04",
|
|
purpose: "attachment",
|
|
mediaType: "application/octet-stream",
|
|
source: rangeSource(new Uint8Array([1, 2, 3, 4])),
|
|
signal: activeSignal,
|
|
});
|
|
expect(result.ok).toBe(true);
|
|
expect(harness.issued).toHaveBeenCalledTimes(2);
|
|
expect(delays).toHaveLength(1);
|
|
});
|
|
|
|
it("honors a bounded Retry-After delay and reissues the single-use capability", async () => {
|
|
const checkpoints = new MemoryCheckpointStore();
|
|
const harness = createControlHarness();
|
|
const delays: number[] = [];
|
|
let uploadAttempts = 0;
|
|
const runtime = createResumableUploadRuntime({
|
|
controlPlane: harness.control,
|
|
partExecutor: {
|
|
async uploadPart(input) {
|
|
uploadAttempts += 1;
|
|
if (uploadAttempts === 1) {
|
|
return Object.freeze({
|
|
ok: false as const,
|
|
error: Object.freeze({
|
|
code: "UNAVAILABLE" as const,
|
|
operation: "UPLOAD_PART" as const,
|
|
retryable: true,
|
|
recovery: "REISSUE_CAPABILITY" as const,
|
|
retryAfterMs: 7,
|
|
}),
|
|
});
|
|
}
|
|
const receipt = Object.freeze({
|
|
...input.part,
|
|
receiptToken: "etag-retried-part",
|
|
});
|
|
harness.accepted.set(input.part.partNumber, receipt);
|
|
return browserDataSuccess(receipt);
|
|
},
|
|
},
|
|
checkpoints,
|
|
mutationLock: noContentionLock,
|
|
crypto,
|
|
policy: runtimePolicy({ maxRetries: 1 }),
|
|
now: () => 1_000,
|
|
random: () => 0,
|
|
sleep: async (delay) => {
|
|
delays.push(delay);
|
|
},
|
|
});
|
|
const result = await runtime.upload({
|
|
uploadKey: "upload_key_retry_after",
|
|
purpose: "attachment",
|
|
mediaType: "application/octet-stream",
|
|
source: rangeSource(new Uint8Array([1, 2, 3, 4])),
|
|
signal: activeSignal,
|
|
});
|
|
|
|
expect(result.ok).toBe(true);
|
|
expect(uploadAttempts).toBe(2);
|
|
expect(harness.issued).toHaveBeenCalledTimes(2);
|
|
expect(delays).toEqual([7]);
|
|
});
|
|
|
|
it("detects source mutation between the fingerprint and transfer passes", async () => {
|
|
let reads = 0;
|
|
const source: ResumableUploadSource = {
|
|
kind: "RANGE_READER",
|
|
reader: {
|
|
byteLength: 4,
|
|
async readRange() {
|
|
reads += 1;
|
|
return browserDataSuccess(
|
|
reads === 1
|
|
? new Uint8Array([1, 2, 3, 4])
|
|
: new Uint8Array([9, 2, 3, 4]),
|
|
);
|
|
},
|
|
},
|
|
};
|
|
const harness = createControlHarness();
|
|
const executor = { uploadPart: vi.fn() };
|
|
const runtime = createResumableUploadRuntime({
|
|
controlPlane: harness.control,
|
|
partExecutor: executor as unknown as UploadPartExecutor<TestCapability>,
|
|
checkpoints: new MemoryCheckpointStore(),
|
|
mutationLock: noContentionLock,
|
|
crypto,
|
|
policy: runtimePolicy(),
|
|
now: () => 1_000,
|
|
sleep: async () => {},
|
|
});
|
|
const result = await runtime.upload({
|
|
uploadKey: "upload_key_05",
|
|
purpose: "attachment",
|
|
mediaType: "application/octet-stream",
|
|
source,
|
|
signal: activeSignal,
|
|
});
|
|
expect(result).toMatchObject({
|
|
ok: false,
|
|
error: { code: "STALE_RESULT", recovery: "RESELECT" },
|
|
});
|
|
expect(executor.uploadPart).not.toHaveBeenCalled();
|
|
});
|
|
|
|
it("bounds a control-plane provider that ignores AbortSignal", async () => {
|
|
const harness = createControlHarness();
|
|
harness.control.createSession = (() =>
|
|
new Promise(() => {})) as typeof harness.control.createSession;
|
|
const runtime = createResumableUploadRuntime({
|
|
controlPlane: harness.control,
|
|
partExecutor: executorFor(harness),
|
|
checkpoints: new MemoryCheckpointStore(),
|
|
mutationLock: noContentionLock,
|
|
crypto,
|
|
policy: runtimePolicy({
|
|
maxRetries: 0,
|
|
providerAttemptTimeoutMs: 1,
|
|
}),
|
|
now: () => 1_000,
|
|
sleep: async () => {},
|
|
});
|
|
const result = await runtime.upload({
|
|
uploadKey: "upload_key_06",
|
|
purpose: "attachment",
|
|
mediaType: "application/octet-stream",
|
|
source: rangeSource(new Uint8Array([1, 2, 3, 4])),
|
|
signal: activeSignal,
|
|
});
|
|
expect(result).toMatchObject({
|
|
ok: false,
|
|
error: { code: "UNAVAILABLE", operation: "UPLOAD_SESSION" },
|
|
});
|
|
});
|
|
|
|
it("persists ABORT_PENDING until backend abort reconciliation succeeds", async () => {
|
|
const checkpoints = new MemoryCheckpointStore();
|
|
const harness = createControlHarness();
|
|
const runtime = createResumableUploadRuntime({
|
|
controlPlane: harness.control,
|
|
partExecutor: executorFor(harness),
|
|
checkpoints,
|
|
mutationLock: noContentionLock,
|
|
crypto,
|
|
policy: runtimePolicy({ maxRetries: 0 }),
|
|
now: () => 1_000,
|
|
sleep: async () => {},
|
|
});
|
|
const failedUpload = await runtime.upload({
|
|
uploadKey: "upload_key_07",
|
|
purpose: "attachment",
|
|
mediaType: "application/octet-stream",
|
|
source: {
|
|
kind: "RANGE_READER",
|
|
reader: {
|
|
byteLength: 4,
|
|
async readRange() {
|
|
return browserDataFailure("NOT_READABLE", "FILE_READ");
|
|
},
|
|
},
|
|
},
|
|
signal: activeSignal,
|
|
});
|
|
expect(failedUpload.ok).toBe(false);
|
|
// Seed a valid checkpoint because source failure occurs before session create.
|
|
const seededHarness = createControlHarness();
|
|
let abortCalls = 0;
|
|
seededHarness.control.abort = async () => {
|
|
abortCalls += 1;
|
|
return abortCalls === 1
|
|
? browserDataFailure("UNAVAILABLE", "UPLOAD_ABORT", {
|
|
retryable: true,
|
|
recovery: "RESUME",
|
|
})
|
|
: browserDataSuccess({ state: "EXPIRED" });
|
|
};
|
|
const seededRuntime = createResumableUploadRuntime({
|
|
controlPlane: seededHarness.control,
|
|
partExecutor: {
|
|
async uploadPart() {
|
|
return browserDataFailure("UNAVAILABLE", "UPLOAD_PART");
|
|
},
|
|
},
|
|
checkpoints,
|
|
mutationLock: noContentionLock,
|
|
crypto,
|
|
policy: runtimePolicy({ maxRetries: 0 }),
|
|
now: () => 1_000,
|
|
sleep: async () => {},
|
|
});
|
|
const fingerprint = {
|
|
algorithm: "SHA-256-PARTS-V1" as const,
|
|
digestHex: "a".repeat(64),
|
|
byteLength: 4,
|
|
partSizeBytes: 4,
|
|
partCount: 1,
|
|
};
|
|
checkpoints.rows.set("upload_key_08", {
|
|
schemaVersion: 1,
|
|
protocol: RESUMABLE_UPLOAD_PROTOCOL,
|
|
revision: 1,
|
|
state: "ACTIVE",
|
|
uploadKey: "upload_key_08",
|
|
requestBindingSha256: "b".repeat(64),
|
|
fingerprint,
|
|
sessionId: "session_08",
|
|
sessionExpiresAtEpochMs: 5_000,
|
|
sessionMaxConcurrency: 1,
|
|
acceptedParts: [],
|
|
updatedAtEpochMs: 1_000,
|
|
});
|
|
|
|
expect(
|
|
await seededRuntime.abort({
|
|
uploadKey: "upload_key_08",
|
|
signal: activeSignal,
|
|
}),
|
|
).toMatchObject({
|
|
ok: false,
|
|
error: { code: "UNAVAILABLE" },
|
|
});
|
|
expect(checkpoints.rows.get("upload_key_08")?.state).toBe(
|
|
"ABORT_PENDING",
|
|
);
|
|
expect(
|
|
await seededRuntime.abort({
|
|
uploadKey: "upload_key_08",
|
|
signal: activeSignal,
|
|
}),
|
|
).toEqual({ ok: true, value: { state: "ORPHANED" } });
|
|
expect(checkpoints.rows.has("upload_key_08")).toBe(false);
|
|
});
|
|
|
|
it("treats definitive abort 404/410 outcomes as orphan cleanup", async () => {
|
|
const checkpoints = new MemoryCheckpointStore();
|
|
const harness = createControlHarness();
|
|
harness.control.abort = async () =>
|
|
browserDataFailure("NOT_FOUND", "UPLOAD_ABORT", {
|
|
recovery: "RECONCILE",
|
|
});
|
|
checkpoints.rows.set("upload_key_abort_not_found", {
|
|
schemaVersion: 1,
|
|
protocol: RESUMABLE_UPLOAD_PROTOCOL,
|
|
revision: 1,
|
|
state: "ACTIVE",
|
|
uploadKey: "upload_key_abort_not_found",
|
|
requestBindingSha256: "a".repeat(64),
|
|
fingerprint: {
|
|
algorithm: "SHA-256-PARTS-V1",
|
|
digestHex: "b".repeat(64),
|
|
byteLength: 4,
|
|
partSizeBytes: 4,
|
|
partCount: 1,
|
|
},
|
|
sessionId: "session_abort_not_found",
|
|
sessionExpiresAtEpochMs: 5_000,
|
|
sessionMaxConcurrency: 1,
|
|
acceptedParts: [],
|
|
updatedAtEpochMs: 1_000,
|
|
});
|
|
const runtime = createResumableUploadRuntime({
|
|
controlPlane: harness.control,
|
|
partExecutor: executorFor(harness),
|
|
checkpoints,
|
|
mutationLock: noContentionLock,
|
|
crypto,
|
|
policy: runtimePolicy({ maxRetries: 0 }),
|
|
now: () => 1_000,
|
|
sleep: async () => {},
|
|
});
|
|
|
|
const result = await runtime.abort({
|
|
uploadKey: "upload_key_abort_not_found",
|
|
signal: activeSignal,
|
|
});
|
|
|
|
expect(result).toEqual({
|
|
ok: true,
|
|
value: { state: "ORPHANED" },
|
|
});
|
|
expect(checkpoints.rows.has("upload_key_abort_not_found")).toBe(
|
|
false,
|
|
);
|
|
});
|
|
|
|
it("interrupts a same-runtime active upload before explicit abort waits for the key lock", async () => {
|
|
const checkpoints = new MemoryCheckpointStore();
|
|
const harness = createControlHarness();
|
|
let markEntered!: () => void;
|
|
const entered = new Promise<void>((resolve) => {
|
|
markEntered = resolve;
|
|
});
|
|
const runtime = createResumableUploadRuntime({
|
|
controlPlane: harness.control,
|
|
partExecutor: {
|
|
async uploadPart(input) {
|
|
markEntered();
|
|
return await new Promise((resolve) => {
|
|
const abort = () =>
|
|
resolve(
|
|
browserDataFailure("ABORTED", "UPLOAD_PART"),
|
|
);
|
|
input.signal.addEventListener("abort", abort, {
|
|
once: true,
|
|
});
|
|
if (input.signal.aborted) abort();
|
|
});
|
|
},
|
|
},
|
|
checkpoints,
|
|
mutationLock: createSerialMutationLock(),
|
|
crypto,
|
|
policy: runtimePolicy({ maxRetries: 0 }),
|
|
now: () => 1_000,
|
|
sleep: async () => {},
|
|
});
|
|
const upload = runtime.upload({
|
|
uploadKey: "upload_key_local_interrupt",
|
|
purpose: "attachment",
|
|
mediaType: "application/octet-stream",
|
|
source: rangeSource(new Uint8Array([1, 2, 3, 4])),
|
|
signal: activeSignal,
|
|
});
|
|
await entered;
|
|
|
|
const aborted = runtime.abort({
|
|
uploadKey: "upload_key_local_interrupt",
|
|
signal: activeSignal,
|
|
});
|
|
|
|
await expect(upload).resolves.toMatchObject({
|
|
ok: false,
|
|
error: { code: "ABORTED", operation: "UPLOAD_PART" },
|
|
});
|
|
await expect(aborted).resolves.toEqual({
|
|
ok: true,
|
|
value: { state: "ABORTED" },
|
|
});
|
|
expect(checkpoints.rows.size).toBe(0);
|
|
});
|
|
|
|
it("interrupts another context before explicit abort waits for the shared Web Lock", async () => {
|
|
const checkpoints = new MemoryCheckpointStore();
|
|
const harness = createControlHarness();
|
|
const mutationLock = createSerialMutationLock();
|
|
const [firstCancellation, secondCancellation] =
|
|
createMemoryCancellationPair();
|
|
let markEntered!: () => void;
|
|
const entered = new Promise<void>((resolve) => {
|
|
markEntered = resolve;
|
|
});
|
|
const dependencies = {
|
|
controlPlane: harness.control,
|
|
checkpoints,
|
|
mutationLock,
|
|
crypto,
|
|
policy: runtimePolicy({ maxRetries: 0 }),
|
|
now: () => 1_000,
|
|
sleep: async () => {},
|
|
} as const;
|
|
const firstContext = createResumableUploadRuntime({
|
|
...dependencies,
|
|
crossContextCancellation: firstCancellation,
|
|
partExecutor: {
|
|
async uploadPart(input) {
|
|
markEntered();
|
|
return await new Promise((resolve) => {
|
|
const abort = () =>
|
|
resolve(
|
|
browserDataFailure("ABORTED", "UPLOAD_PART"),
|
|
);
|
|
input.signal.addEventListener("abort", abort, {
|
|
once: true,
|
|
});
|
|
if (input.signal.aborted) abort();
|
|
});
|
|
},
|
|
},
|
|
});
|
|
const secondContext = createResumableUploadRuntime({
|
|
...dependencies,
|
|
crossContextCancellation: secondCancellation,
|
|
partExecutor: executorFor(harness),
|
|
});
|
|
const upload = firstContext.upload({
|
|
uploadKey: "upload_key_cross_context",
|
|
purpose: "attachment",
|
|
mediaType: "application/octet-stream",
|
|
source: rangeSource(new Uint8Array([1, 2, 3, 4])),
|
|
signal: activeSignal,
|
|
});
|
|
await entered;
|
|
const abort = secondContext.abort({
|
|
uploadKey: "upload_key_cross_context",
|
|
signal: activeSignal,
|
|
});
|
|
|
|
await expect(upload).resolves.toMatchObject({
|
|
ok: false,
|
|
error: { code: "ABORTED", operation: "UPLOAD_PART" },
|
|
});
|
|
await expect(abort).resolves.toEqual({
|
|
ok: true,
|
|
value: { state: "ABORTED" },
|
|
});
|
|
expect(checkpoints.rows.size).toBe(0);
|
|
firstContext.close();
|
|
secondContext.close();
|
|
});
|
|
|
|
it("reconciles an ambiguous complete response to QUARANTINED even after session expiry", async () => {
|
|
const checkpoints = new MemoryCheckpointStore();
|
|
const harness = createControlHarness();
|
|
let committed = false;
|
|
const complete = vi.fn(
|
|
async (): ReturnType<
|
|
ResumableUploadControlPlane<TestCapability>["complete"]
|
|
> => {
|
|
committed = true;
|
|
return browserDataFailure("UNAVAILABLE", "UPLOAD_COMPLETE", {
|
|
retryable: true,
|
|
recovery: "RECONCILE",
|
|
});
|
|
},
|
|
);
|
|
harness.control.complete = complete;
|
|
harness.control.getStatus = async () => {
|
|
const session = harness.getSession();
|
|
if (!session) {
|
|
return browserDataFailure("NOT_FOUND", "UPLOAD_RECONCILE");
|
|
}
|
|
return committed
|
|
? browserDataSuccess({
|
|
state: "QUARANTINED",
|
|
session,
|
|
resourceId: "resource_ambiguous_01",
|
|
})
|
|
: browserDataSuccess({
|
|
state: "ACTIVE",
|
|
session,
|
|
acceptedParts: [...harness.accepted.values()].sort(
|
|
(left, right) => left.partNumber - right.partNumber,
|
|
),
|
|
});
|
|
};
|
|
|
|
const firstRuntime = createResumableUploadRuntime({
|
|
controlPlane: harness.control,
|
|
partExecutor: executorFor(harness),
|
|
checkpoints,
|
|
mutationLock: noContentionLock,
|
|
crypto,
|
|
policy: runtimePolicy({ maxRetries: 0 }),
|
|
now: () => 1_000,
|
|
sleep: async () => {},
|
|
});
|
|
const request = {
|
|
uploadKey: "upload_key_ambiguous_complete",
|
|
purpose: "attachment",
|
|
mediaType: "application/octet-stream",
|
|
source: rangeSource(new Uint8Array([1, 2, 3, 4])),
|
|
signal: activeSignal,
|
|
} as const;
|
|
const ambiguous = await firstRuntime.upload(request);
|
|
|
|
expect(ambiguous).toMatchObject({
|
|
ok: false,
|
|
error: {
|
|
code: "UNAVAILABLE",
|
|
operation: "UPLOAD_COMPLETE",
|
|
recovery: "RECONCILE",
|
|
},
|
|
});
|
|
expect(checkpoints.rows.has(request.uploadKey)).toBe(true);
|
|
|
|
const resumedPartExecutor = { uploadPart: vi.fn() };
|
|
const resumedRuntime = createResumableUploadRuntime({
|
|
controlPlane: harness.control,
|
|
partExecutor:
|
|
resumedPartExecutor as unknown as UploadPartExecutor<TestCapability>,
|
|
checkpoints,
|
|
mutationLock: noContentionLock,
|
|
crypto,
|
|
policy: runtimePolicy({ maxRetries: 0 }),
|
|
now: () => 7_000,
|
|
sleep: async () => {},
|
|
});
|
|
const recovered = await resumedRuntime.upload(request);
|
|
|
|
expect(recovered).toMatchObject({
|
|
ok: true,
|
|
value: {
|
|
state: "QUARANTINED",
|
|
resourceId: "resource_ambiguous_01",
|
|
replayed: true,
|
|
},
|
|
});
|
|
expect(complete).toHaveBeenCalledTimes(1);
|
|
expect(resumedPartExecutor.uploadPart).not.toHaveBeenCalled();
|
|
expect(checkpoints.rows.has(request.uploadKey)).toBe(false);
|
|
});
|
|
|
|
it("bridges presigned PUT using the verified response receipt, not the capability receipt", async () => {
|
|
let forwardedUploadBinding = "";
|
|
let forwardedSessionId = "";
|
|
let forwardedRequestBinding = "";
|
|
const port: PresignedUploadPartPort = {
|
|
async put(input) {
|
|
forwardedUploadBinding = input.uploadBindingSha256;
|
|
forwardedSessionId = input.sessionId;
|
|
forwardedRequestBinding = input.requestBindingSha256;
|
|
return browserDataSuccess({
|
|
bytesWritten: input.byteLength,
|
|
checksumSha256: input.checksumSha256,
|
|
receiptToken: "etag.response-01",
|
|
});
|
|
},
|
|
};
|
|
const executor = createPresignedUploadPartExecutor(port, () => 1_000);
|
|
const capability = {
|
|
capabilityReceipt: "capability_receipt_01",
|
|
method: "PUT",
|
|
binding: {
|
|
kind: "UPLOAD_PART",
|
|
protocol: RESUMABLE_UPLOAD_PROTOCOL,
|
|
sessionId: "session_01",
|
|
requestBindingSha256: "c".repeat(64),
|
|
uploadBindingSha256: "b".repeat(64),
|
|
partNumber: 1,
|
|
offset: 0,
|
|
idempotencyKey: "upload-part-idempotency-01",
|
|
},
|
|
mediaType: "application/octet-stream",
|
|
byteLength: 4,
|
|
maxBytes: 4,
|
|
expectedSha256: "a".repeat(64),
|
|
expiresAtEpochMs: 2_000,
|
|
} as unknown as PresignedUploadPartCapability;
|
|
const result = await executor.uploadPart({
|
|
protocol: RESUMABLE_UPLOAD_PROTOCOL,
|
|
capability,
|
|
sessionId: "session_01",
|
|
requestBindingSha256: "c".repeat(64),
|
|
uploadBindingSha256: "b".repeat(64),
|
|
fingerprint: {
|
|
algorithm: "SHA-256-PARTS-V1",
|
|
digestHex: "d".repeat(64),
|
|
byteLength: 4,
|
|
partSizeBytes: 4,
|
|
partCount: 1,
|
|
},
|
|
mediaType: "application/octet-stream",
|
|
part: {
|
|
partNumber: 1,
|
|
offset: 0,
|
|
byteLength: 4,
|
|
checksumSha256: "a".repeat(64),
|
|
},
|
|
bytes: new Uint8Array([1, 2, 3, 4]),
|
|
idempotencyKey: "upload-part-idempotency-01",
|
|
signal: activeSignal,
|
|
});
|
|
expect(result).toMatchObject({
|
|
ok: true,
|
|
value: { receiptToken: "etag.response-01" },
|
|
});
|
|
expect(forwardedUploadBinding).toBe("b".repeat(64));
|
|
expect(forwardedSessionId).toBe("session_01");
|
|
expect(forwardedRequestBinding).toBe("c".repeat(64));
|
|
if (result.ok) {
|
|
expect(result.value.receiptToken).not.toBe(
|
|
capability.capabilityReceipt,
|
|
);
|
|
}
|
|
});
|
|
});
|
|
/**
|
|
* 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<Value>(): Promise<Value> {
|
|
return await new Promise<never>(() => {});
|
|
},
|
|
}),
|
|
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);
|
|
});
|
|
});
|