DongHyeonkaandClaude Opus 5
39a4a973a8
The capability vault checked an issuer's registration and then read it again to store it, including its nested header rows. A stateful issuer could show an allowed header set to the forbidden-header check and hand `Authorization` to the copy, so the vault stored — and the executor sent — a credential no rule had ever seen. The registration and everything nested in it is now snapshotted once, and only that snapshot is validated, frozen and stored. The upload control plane had the same shape one level down: a `sessionId` that answered `session_01` to the regex and `../../unsafe` to the result snapshot reached a success receipt. Two lifetimes were also unowned. A download source lease that resolved after the caller's abort never reached the holder, so nothing closed it and its fetch reader and capability lease outlived the terminal result; a compensator sharing the holder's close-once latch now closes it exactly once. And `dispose()` proved quiescence from the wrapper registry alone, so a provider that ignored its attempt deadline let teardown report a drained runtime and close the checkpoint store while the provider was still running. Raw provider promises are now their own registry and the drain must prove both. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
1457 lines
45 KiB
TypeScript
1457 lines
45 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);
|
|
});
|
|
});
|
|
|
|
/**
|
|
* TR-04. `dispose()` proved quiescence from the wrapper registry only. A
|
|
* provider that ignored its attempt deadline let the wrapper settle first and
|
|
* leave the set empty, so teardown reported a drained runtime — and closed the
|
|
* checkpoint store — while the provider was still running.
|
|
*/
|
|
describe("TR-04 teardown waits for raw provider work, not only its wrapper", () => {
|
|
it("refuses to report a drained runtime while a provider is still running", async () => {
|
|
const harness = createControlHarness();
|
|
const checkpoints = new MemoryCheckpointStore();
|
|
const closeStore = vi.spyOn(checkpoints, "close");
|
|
let releaseProvider: (() => void) | undefined;
|
|
// Ignores the attempt signal entirely and outlives its own deadline.
|
|
harness.control.createSession = () =>
|
|
new Promise((resolve) => {
|
|
releaseProvider = () =>
|
|
resolve(
|
|
browserDataFailure("UNAVAILABLE", "UPLOAD_SESSION", {
|
|
retryable: true,
|
|
recovery: "RESUME",
|
|
}),
|
|
);
|
|
});
|
|
|
|
const runtime = createResumableUploadRuntime({
|
|
controlPlane: harness.control,
|
|
partExecutor: executorFor(harness),
|
|
checkpoints,
|
|
mutationLock: noContentionLock,
|
|
crypto,
|
|
policy: runtimePolicy({
|
|
providerAttemptTimeoutMs: 5,
|
|
cleanupDeadlineMs: 25,
|
|
maxRetries: 0,
|
|
}),
|
|
now: () => 1_000,
|
|
random: () => 0,
|
|
sleep: async () => {},
|
|
});
|
|
|
|
const uploading = runtime.upload({
|
|
uploadKey: "upload_key_raw",
|
|
purpose: "attachment",
|
|
mediaType: "application/octet-stream",
|
|
source: byteStreamSource(new Uint8Array([1, 2, 3, 4, 5])),
|
|
signal: activeSignal,
|
|
});
|
|
await vi.waitFor(() => expect(releaseProvider).toBeDefined());
|
|
// The wrapper has already given up on the attempt.
|
|
await uploading;
|
|
|
|
const disposed = await runtime.dispose();
|
|
expect(disposed.ok).toBe(false);
|
|
if (!disposed.ok) {
|
|
expect(disposed.error.code).toBe("UNAVAILABLE");
|
|
expect(disposed.error.recovery).toBe("RESUME");
|
|
}
|
|
// The store stays open while something could still write a checkpoint.
|
|
expect(closeStore).not.toHaveBeenCalled();
|
|
|
|
releaseProvider?.();
|
|
});
|
|
|
|
it("reports a drained runtime once the raw provider settles", async () => {
|
|
const harness = createControlHarness();
|
|
const checkpoints = new MemoryCheckpointStore();
|
|
let releaseProvider: (() => void) | undefined;
|
|
harness.control.createSession = () =>
|
|
new Promise((resolve) => {
|
|
releaseProvider = () =>
|
|
resolve(
|
|
browserDataFailure("UNAVAILABLE", "UPLOAD_SESSION", {
|
|
retryable: true,
|
|
recovery: "RESUME",
|
|
}),
|
|
);
|
|
});
|
|
|
|
const runtime = createResumableUploadRuntime({
|
|
controlPlane: harness.control,
|
|
partExecutor: executorFor(harness),
|
|
checkpoints,
|
|
mutationLock: noContentionLock,
|
|
crypto,
|
|
policy: runtimePolicy({
|
|
providerAttemptTimeoutMs: 5,
|
|
cleanupDeadlineMs: 1_000,
|
|
maxRetries: 0,
|
|
}),
|
|
now: () => 1_000,
|
|
random: () => 0,
|
|
sleep: async () => {},
|
|
});
|
|
|
|
const uploading = runtime.upload({
|
|
uploadKey: "upload_key_raw_2",
|
|
purpose: "attachment",
|
|
mediaType: "application/octet-stream",
|
|
source: byteStreamSource(new Uint8Array([1, 2, 3, 4, 5])),
|
|
signal: activeSignal,
|
|
});
|
|
await vi.waitFor(() => expect(releaseProvider).toBeDefined());
|
|
await uploading;
|
|
|
|
const disposing = runtime.dispose();
|
|
releaseProvider?.();
|
|
await expect(disposing).resolves.toMatchObject({ ok: true });
|
|
expect(runtime.lifecycle()).toBe("CLOSED");
|
|
});
|
|
});
|