tech-log-frontend/tests/unit/resumable-upload-runtime.test.ts

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");
  });
});