test(restart-chaos): add ST4 partial multipart and ST7 read-after-write lag storage faults (#20504)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work
> - The restart-chaos harness runs real agent tasks while it injects
infrastructure faults
> - `--storage s3` puts a fault proxy between the server and a loopback
MinIO (ST1–ST3, ST5/ST5R)
> - Remote workspace snapshots over 16 MiB upload as S3 multipart
uploads, and saves are read back right after they are written
> - No fault hit single upload parts or the completion, and no fault
made a just-written object briefly invisible
> - Nothing checked that a wave left no orphaned multipart uploads or
that every committed snapshot reads back
> - This pull request adds ST4 (partial multipart) and ST7
(read-after-write lag), and a teardown store audit
> - The benefit is that waves can prove snapshot saves and restores
survive part failures and read lag without data loss or leaked uploads

## Linked Issues or Issue Description

**What existing behavior does this improve?**
The restart-chaos storage faults treat every request the same. They
cannot target one multipart part or the completion, and they cannot lag
reads behind writes.

**Subsystem affected**
tests/restart-chaos (storage chaos).

**Current behavior**
A wave cannot fail the Nth `UploadPart` of a snapshot upload, cut a part
body mid-transfer, or fail `CompleteMultipartUpload`. A wave cannot make
a `GET` or `HEAD` answer `NoSuchKey` right after a `PUT`. No wave checks
for incomplete multipart uploads or unreadable committed snapshots.

**Proposed behavior**
`--storage-faults ST4` fails a seeded subset of snapshot upload parts
(500/503, reset, or reset after 1–4 MiB of the body), and sometimes the
completion. `--storage-faults ST7` answers 404 `NoSuchKey` to reads of
an object for 5–30 s after its write succeeded. With either fault, the
wave reads every committed workspace snapshot back from MinIO (size and
SHA-256 against its row) and, after the server stops, lists incomplete
multipart uploads under the wave prefix. Either problem fails the wave.

**Reason and benefit**
These are the storage failures that can lose a workspace or leak storage
cost without any visible error.

## What Changed

- `storage-s3-ops.ts` (new): names the S3 operation from the method and
query, and reads the part number.
- `storage-multipart.ts` (new): ST4 rule planning.
- `storage-lag.ts` (new): ST7 rule planning, and a tracker of successful
`PutObject` and `CompleteMultipartUpload` writes per key.
- `storage-audit.ts` (new): teardown audit (committed snapshot
read-back, incomplete multipart uploads), `storage-audit.json`, and a
`Storage audit` section in `SUMMARY.md`. MinIO lists multipart uploads
only for an exact key or no prefix, so the audit lists the wave-owned
bucket and filters by the wave prefix.
- `storage-proxy.ts`: rules can match `operations` and `partNumbers`.
New faults `truncate` (reset after N body bytes) and
`read_after_write_lag`. `storage-proxy.json` reports `lagged`.
- `storage-chaos.ts`, `main.ts`: registry lines for ST4/ST7 and the
audit calls.
- README: ST4/ST7 rows and the audit.

## Verification

- `PAPERCLIP_IN_WORKTREE=false npx vitest run --config
tests/restart-chaos/vitest.config.ts`: 35 files, 423 tests pass. The new
`storage-multipart-lag.test.ts` fails before this change (the modules
and kinds do not exist) and passes after. It includes real MinIO tests
with the AWS SDK: part 2 fails once with 500, a part body is cut after 1
MiB, the completion fails once, the audit finds the incomplete upload
and then finds none after completion; a just-written object answers
`NoSuchKey` to `GET` and 404 to `HEAD` until the lag passes; the
snapshot check reports `sha256_mismatch` and `object_missing`.
- `node scripts/typecheck-test-projects.mjs tests/restart-chaos` passes.
- Live waves run from master after merge (local ST7, Daytona R1 ST4 and
ST7). Results go in a follow-up comment.

## Risks

- Test harness only. No product code changes.
- The audit runs only on waves with ST4 or ST7, so other storage waves
do not change.
- The snapshot check verifies the stored ciphertext against its row. It
does not decrypt, because the key lives in the instance.

## Model Used

- Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, with tool use and code
execution in Claude Code.

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [ ] All Paperclip CI gates are green
- [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

tests/restart-chaos/README.md

@@ -362,8 +362,17 @@ method and status are written to `storage-proxy.json`.
 | ST1 | outage: every request answers 503 `ServiceUnavailable` |
 | ST2 | slow store: 1–10 s added latency with jitter, and a 2–16 MiB/s bandwidth cap shared by every request in the window (each direction) |
 | ST3 | 30% of requests answer 503, half of them `SlowDown` |
+| ST4 | partial multipart: for workspace snapshot uploads, a seeded subset of parts 1–4 fails with 500/503, a connection reset, or a reset after 1–4 MiB of the part body; sometimes `CompleteMultipartUpload` also fails (500 or reset). Each fault hits 1–3 times or for the whole window |
 | ST5 | credentials rejected: every request answers 403 `ExpiredToken`, `InvalidAccessKeyId` or `SignatureDoesNotMatch` (seeded per window) |
 | ST5R | real key rotation: MinIO restarts with a new root key (old key revoked), and the server's credential source keeps serving the old key until the window ends |
+| ST7 | read-after-write lag: a `GET` or `HEAD` of an object answers 404 `NoSuchKey` for 5–30 s (seeded) after its `PUT` or multipart completion succeeded |
+
+With ST4 or ST7 in the wave, the harness audits the store when the wave ends (`storage-audit.json` and a
+`Storage audit` section in `SUMMARY.md`). It reads every committed workspace snapshot back from MinIO and checks
+its size and SHA-256 against the snapshot row, and after the server stops it lists incomplete multipart uploads
+under the wave's prefix. An orphaned upload, a committed snapshot that does not read back, a planned ST4 or ST7 fault
+that never changed a request, or a skipped snapshot check fails the wave. ST4 truncation forwards the first bytes
+of the part to MinIO before it cuts both connections.
 
 With ST5R the server gets no static key: `AWS_PROFILE` names a profile in a wave-owned `AWS_CONFIG_FILE`
 whose `credential_process` prints the key currently in `aws-credentials/current.json` under the instance

tests/restart-chaos/main.ts

@@ -17,6 +17,7 @@ import { BoardUi } from "./board-ui.js";
 import { DEFAULT_MINIO_BINARY, MINIO_RELEASE, minioBinary, objectStoreServerEnvironment, startMinio, type ObjectStore } from "./object-store.js";
 import { StorageChaos, planStorageFaults, replayStorageWindows, type StorageFaultReceipt } from "./storage-chaos.js";
 import { startStorageFaultProxy } from "./storage-proxy.js";
+import { StorageAuditor, auditsStorage, storageAuditMarkdown } from "./storage-audit.js";
 import { exposedAfterLaunch, StorageTrigger, storageTriggerMarkdown, storageTriggerShortfall } from "./storage-trigger.js";
 import { needsCredentialRotation, startCredentialRotation, type CredentialRotation } from "./storage-credentials.js";
 import { describeProviderFault, installProviderFaultSeam, resolveRealCodex, withProviderFaultCheck, type ProviderFaultSeam } from "./provider-faults.js";
@@ -254,6 +255,7 @@ async function main(): Promise<number> {
   let storageTrigger: StorageTrigger | null = null;
   let storageProxy: Awaited<ReturnType<typeof startStorageFaultProxy>> | null = null;
   let credentialRotation: CredentialRotation | null = null;
+  let storageAuditor: StorageAuditor | null = null;
   let providerFaults: ProviderFaultSeam | null = null;
   let acpxSidecarSeam: AcpxSidecarSeam | null = null;
   let speedSection: string | null = null;
@@ -288,6 +290,7 @@ async function main(): Promise<number> {
       if (!binary) throw new Error(`--storage s3 needs MinIO ${MINIO_RELEASE} at ${DEFAULT_MINIO_BINARY} or RESTART_CHAOS_MINIO_BIN`);
       objectStore = await startMinio({ binary, root: supervisor.paths.temporaryRoot });
       storageProxy = await startStorageFaultProxy(objectStore.endpoint);
+      if (auditsStorage(plan.options)) storageAuditor = new StorageAuditor(objectStore, `chaos/${plan.waveId}/`);
       Object.assign(serverEnvironment, objectStoreServerEnvironment(objectStore, storageProxy.endpoint, `chaos/${plan.waveId}/`));
       if (needsCredentialRotation(plan.options)) {
         const rotating = await startCredentialRotation({ store: objectStore, binary, root: supervisor.paths.temporaryRoot, record: (kind, data) => timeline.add(kind, data) });
@@ -623,6 +626,11 @@ async function main(): Promise<number> {
         rules: Object.fromEntries(Object.entries(grade.rules).map(([id, rule]) => [id, rule.verdict])), T: grade.T.verdict,
         universal: Object.fromEntries(Object.entries(grade.universal ?? {}).map(([id, rule]) => [id, `${rule.verdict}${rule.enforced ? "" : " (warning)"}`])) });
     }
+    if (storageAuditor) {
+      storageChaos?.stop();
+      await credentialRotation?.settled();
+      await storageAuditor.checkSnapshots(evidenceDb, lab.companyId);
+    }
     await evidenceDb.$client.end({ timeout: 5 }).catch(() => undefined);
     await writeFile(path.join(waveDir, "faults.json"), `${JSON.stringify(faults, null, 2)}\n`, { mode: 0o600 });
     const speed = await (async () => {
@@ -723,6 +731,13 @@ async function main(): Promise<number> {
       await writeFile(path.join(waveDir, "storage-proxy.json"), `${JSON.stringify(storageProxy.counts(), null, 2)}\n`, { mode: 0o600 }).catch(() => undefined);
       await storageProxy.close().catch(() => undefined);
     }
+    if (storageAuditor) {
+      const failures = await storageAuditor.finish(waveDir, { kinds: storageChaos?.windows.map(window => window.kind) ?? [],
+        injectedBy: storageProxy?.counts().injectedBy }).catch(error => [`storage audit failed: ${String(error)}`]);
+      timeline.add("storage_audit", { ...storageAuditor.audit, failures });
+      if (failures.length) anyFailure = true;
+      speedSection = `${speedSection ?? ""}\n${storageAuditMarkdown(storageAuditor.audit)}`;
+    }
     await objectStore?.stop().catch(error => timeline.add("object_store_stop_failed", { error: String(error) }));
     await acpxSidecarSeam?.remove().catch(error => timeline.add("acpx_stand_in_cleanup_failed", { error: String(error) }));
     const retention = await releaseInstance(supervisor.paths.temporaryRoot, { keep, stopFailed, remoteCleanupPending, owner: plan.waveId })

tests/restart-chaos/storage-audit.ts

@@ -0,0 +1,145 @@
+import { createHash } from "node:crypto";
+import { writeFile } from "node:fs/promises";
+import { createRequire } from "node:module";
+import path from "node:path";
+import type { Readable } from "node:stream";
+import type { Db } from "../../packages/db/src/index.js";
+import type { ObjectStore } from "./object-store.js";
+
+type StoreAccess = Pick<ObjectStore, "endpoint" | "bucket" | "region" | "accessKeyId" | "secretAccessKey">;
+
+export interface IncompleteUpload { key: string; uploadId: string; initiated: string | null }
+export interface SnapshotProblem { id: string; objectKey: string; problem: string }
+export interface CommittedSnapshot { id: string; objectKey: string | null; sha256: string | null; byteSize: number | null }
+export interface StorageAudit {
+  incompleteUploads: IncompleteUpload[];
+  committedSnapshots: number;
+  snapshotProblems: SnapshotProblem[];
+  snapshotsChecked: boolean;
+  unexercisedFaults: string[];
+  errors: string[];
+}
+
+export function unexercisedFaults(kinds: readonly string[], injectedBy: Record<string, number> = {}): string[] {
+  const hit = (prefixes: readonly string[]) => Object.entries(injectedBy).some(([cause, count]) => count > 0 && prefixes.some(prefix => cause.startsWith(prefix)));
+  return [...new Set(kinds)].filter(kind =>
+    (kind === "ST4" && !hit(["UploadPart:", "CompleteMultipartUpload:"]))
+    || (kind === "ST7" && !hit(["GetObject:read_after_write_lag", "HeadObject:read_after_write_lag"])));
+}
+
+export const AUDITED_STORAGE_FAULTS = ["ST4", "ST7"] as const;
+
+export function auditsStorage(options: { storage?: string; storageFaults?: readonly string[];
+  storageWindows?: ReadonlyArray<{ kind: string }> }): boolean {
+  if (options.storage !== "s3") return false;
+  const kinds = [...(options.storageFaults ?? []), ...(options.storageWindows ?? []).map(window => window.kind)];
+  return kinds.some(kind => (AUDITED_STORAGE_FAULTS as readonly string[]).includes(kind));
+}
+
+function s3Client(store: StoreAccess) {
+  const sdk = createRequire(new URL("../../server/package.json", import.meta.url))("@aws-sdk/client-s3");
+  const client = new sdk.S3Client({ endpoint: store.endpoint, region: store.region, forcePathStyle: true,
+    credentials: { accessKeyId: store.accessKeyId, secretAccessKey: store.secretAccessKey } });
+  return { sdk, client };
+}
+
+export function storedKey(prefix: string, objectKey: string): string {
+  const trimmed = prefix.replace(/^\/+|\/+$/g, "");
+  return trimmed ? `${trimmed}/${objectKey}` : objectKey;
+}
+
+export async function listIncompleteUploads(store: StoreAccess, prefix: string): Promise<IncompleteUpload[]> {
+  const { sdk, client } = s3Client(store);
+  const uploads: IncompleteUpload[] = [];
+  try {
+    let keyMarker: string | undefined, uploadIdMarker: string | undefined;
+    for (;;) {
+      const page = await client.send(new sdk.ListMultipartUploadsCommand({ Bucket: store.bucket,
+        ...(keyMarker ? { KeyMarker: keyMarker } : {}), ...(uploadIdMarker ? { UploadIdMarker: uploadIdMarker } : {}) }));
+      for (const upload of page.Uploads ?? []) {
+        if (!String(upload.Key).startsWith(prefix)) continue;
+        uploads.push({ key: String(upload.Key), uploadId: String(upload.UploadId),
+          initiated: upload.Initiated ? new Date(upload.Initiated).toISOString() : null });
+      }
+      if (!page.IsTruncated) return uploads;
+      keyMarker = page.NextKeyMarker; uploadIdMarker = page.NextUploadIdMarker;
+      if (!keyMarker && !uploadIdMarker) return uploads;
+    }
+  } finally { client.destroy(); }
+}
+
+export async function checkCommittedSnapshots(store: StoreAccess, prefix: string,
+  snapshots: readonly CommittedSnapshot[]): Promise<SnapshotProblem[]> {
+  const { sdk, client } = s3Client(store);
+  const problems: SnapshotProblem[] = [];
+  try {
+    for (const snapshot of snapshots) {
+      const objectKey = snapshot.objectKey ?? "";
+      const fail = (problem: string) => problems.push({ id: snapshot.id, objectKey, problem });
+      if (!snapshot.objectKey || !snapshot.sha256 || snapshot.byteSize === null) { fail("committed_without_object"); continue; }
+      try {
+        const object = await client.send(new sdk.GetObjectCommand({ Bucket: store.bucket, Key: storedKey(prefix, snapshot.objectKey) }));
+        const digest = createHash("sha256");
+        let size = 0;
+        for await (const chunk of object.Body as Readable) { size += (chunk as Buffer).length; digest.update(chunk as Buffer); }
+        if (size !== snapshot.byteSize) fail(`size ${size} != ${snapshot.byteSize}`);
+        else if (digest.digest("hex") !== snapshot.sha256) fail("sha256_mismatch");
+      } catch (error) {
+        const name = (error as { name?: string }).name;
+        fail(name === "NoSuchKey" || name === "NotFound" ? "object_missing" : `read_failed: ${String(error)}`);
+      }
+    }
+  } finally { client.destroy(); }
+  return problems;
+}
+
+export function storageAuditFailures(audit: StorageAudit): string[] {
+  return [
+    ...audit.incompleteUploads.map(upload => `orphaned multipart upload ${upload.key} (${upload.uploadId}, initiated ${upload.initiated ?? "unknown"})`),
+    ...audit.snapshotProblems.map(problem => `committed snapshot ${problem.id} does not restore: ${problem.problem}`),
+    ...audit.unexercisedFaults.map(kind => `storage fault ${kind} never changed a request: the wave did not test it`),
+    ...(audit.snapshotsChecked ? [] : ["storage audit incomplete: committed snapshots were not checked"]),
+    ...audit.errors.map(error => `storage audit incomplete: ${error}`),
+  ];
+}
+
+export function storageAuditMarkdown(audit: StorageAudit): string {
+  const failures = storageAuditFailures(audit);
+  return [`## Storage audit`, "",
+    `Incomplete multipart uploads after the server stopped: ${audit.incompleteUploads.length}.`,
+    audit.snapshotsChecked ? `Committed workspace snapshots read back: ${audit.committedSnapshots}, problems: ${audit.snapshotProblems.length}.`
+      : "Committed workspace snapshots: not checked.",
+    `Storage faults that never changed a request: ${audit.unexercisedFaults.join(", ") || "none"}.`,
+    ...(failures.length ? ["", ...failures.map(failure => `- ${failure}`)] : []), ""].join("\n");
+}
+
+export async function readCommittedSnapshots(db: Db, companyId: string): Promise<CommittedSnapshot[]> {
+  const rows = await db.$client.unsafe(
+    `select id, object_key, sha256, byte_size from execution_workspace_snapshots where company_id = $1 and status = 'committed' order by created_at`,
+    [companyId]);
+  return rows.map(row => ({ id: String(row.id), objectKey: row.object_key === null ? null : String(row.object_key),
+    sha256: row.sha256 === null ? null : String(row.sha256), byteSize: row.byte_size === null ? null : Number(row.byte_size) }));
+}
+
+export class StorageAuditor {
+  readonly audit: StorageAudit = { incompleteUploads: [], committedSnapshots: 0, snapshotProblems: [], snapshotsChecked: false, unexercisedFaults: [], errors: [] };
+
+  constructor(private readonly store: StoreAccess, private readonly prefix: string) {}
+
+  async checkSnapshots(db: Db, companyId: string): Promise<void> {
+    try {
+      const snapshots = await readCommittedSnapshots(db, companyId);
+      this.audit.committedSnapshots = snapshots.length;
+      this.audit.snapshotProblems = await checkCommittedSnapshots(this.store, this.prefix, snapshots);
+      this.audit.snapshotsChecked = true;
+    } catch (error) { this.audit.errors.push(`snapshots: ${String(error)}`); }
+  }
+
+  async finish(waveDir: string, exercised: { kinds: readonly string[]; injectedBy?: Record<string, number> }): Promise<string[]> {
+    this.audit.unexercisedFaults = unexercisedFaults(exercised.kinds, exercised.injectedBy);
+    try { this.audit.incompleteUploads = await listIncompleteUploads(this.store, this.prefix); }
+    catch (error) { this.audit.errors.push(`multipart uploads: ${String(error)}`); }
+    await writeFile(path.join(waveDir, "storage-audit.json"), `${JSON.stringify(this.audit, null, 2)}\n`, { mode: 0o600 });
+    return storageAuditFailures(this.audit);
+  }
+}

tests/restart-chaos/storage-chaos.ts

@@ -1,8 +1,10 @@
 import { credentialFaultRules } from "./storage-credentials.js";
 import type { StorageFaultRule } from "./storage-proxy.js";
 import { bandwidthDraw } from "./storage-bandwidth.js";
+import { lagFaultRules } from "./storage-lag.js";
+import { multipartFaultRules } from "./storage-multipart.js";
 
-export const STORAGE_FAULT_KINDS = ["ST1", "ST2", "ST3", "ST5", "ST5R"] as const;
+export const STORAGE_FAULT_KINDS = ["ST1", "ST2", "ST3", "ST4", "ST5", "ST5R", "ST7"] as const;
 export type StorageFaultKind = typeof STORAGE_FAULT_KINDS[number];
 
 export interface StorageFaultWindow { kind: StorageFaultKind; startMs: number; durationMs: number; rules: StorageFaultRule[] }
@@ -25,7 +27,9 @@ export function storageFaultRules(kind: StorageFaultKind, random: () => number):
       return [{ fault: { kind: "latency", ms, jitterMs: Math.round(ms / 4), bytesPerSecond: bandwidthDraw(random) } }];
     }
     case "ST3": return [{ fault: { kind: "errors", rate: 0.3, status: 503, slowDown: random() < 0.5 } }];
+    case "ST4": return multipartFaultRules(random);
     case "ST5": case "ST5R": return credentialFaultRules(kind, random);
+    case "ST7": return lagFaultRules(random);
   }
 }
 
@@ -65,7 +69,7 @@ export class StorageChaos {
   private timer: ReturnType<typeof setInterval> | null = null;
   readonly receipts: StorageFaultReceipt[] = [];
 
-  constructor(private readonly windows: readonly StorageFaultWindow[], private startedAt: number,
+  constructor(readonly windows: readonly StorageFaultWindow[], private startedAt: number,
     private readonly proxy: { setRules(rules: readonly StorageFaultRule[]): void; clear(): void },
     private readonly record: (kind: string, data: Record<string, unknown>) => void) {}
 

tests/restart-chaos/storage-lag.ts

@@ -0,0 +1,34 @@
+import { objectPath, s3Operation } from "./storage-s3-ops.js";
+import type { StorageFaultRule } from "./storage-proxy.js";
+
+export const LAG_MS_RANGE: readonly [number, number] = [5_000, 30_000];
+const FORGET_AFTER_MS = 10 * 60_000;
+
+export function lagFaultRules(random: () => number): StorageFaultRule[] {
+  const [min, max] = LAG_MS_RANGE;
+  return [{ fault: { kind: "read_after_write_lag", lagMs: min + Math.round(random() * (max - min)) },
+    methods: ["GET", "HEAD"], operations: ["GetObject", "HeadObject"] }];
+}
+
+export class ReadAfterWriteTracker {
+  private readonly written = new Map<string, number>();
+  lagged = 0;
+
+  constructor(private readonly now: () => number = Date.now) {}
+
+  note(method: string, url: string, status: number): void {
+    if (status < 200 || status >= 300) return;
+    const operation = s3Operation(method, url);
+    if (operation !== "PutObject" && operation !== "CompleteMultipartUpload") return;
+    const at = this.now();
+    this.written.set(objectPath(url), at);
+    for (const [key, writtenAt] of this.written) if (at - writtenAt > FORGET_AFTER_MS) this.written.delete(key);
+  }
+
+  lagging(url: string, lagMs: number): boolean {
+    const writtenAt = this.written.get(objectPath(url));
+    const lagging = writtenAt !== undefined && this.now() - writtenAt < lagMs;
+    if (lagging) this.lagged += 1;
+    return lagging;
+  }
+}

tests/restart-chaos/storage-multipart-lag.test.ts

@@ -0,0 +1,220 @@
+import { createHash } from "node:crypto";
+import { mkdtemp, rm } from "node:fs/promises";
+import { createRequire } from "node:module";
+import os from "node:os";
+import path from "node:path";
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+import { minioBinary, startMinio, type ObjectStore } from "./object-store.js";
+import { createServer } from "node:http";
+import { once } from "node:events";
+import { auditsStorage, checkCommittedSnapshots, listIncompleteUploads, storageAuditFailures, storedKey, unexercisedFaults } from "./storage-audit.js";
+import { planStorageFaults, storageFaultRules } from "./storage-chaos.js";
+import { LAG_MS_RANGE, ReadAfterWriteTracker, lagFaultRules } from "./storage-lag.js";
+import { SNAPSHOT_PATH, multipartFaultRules } from "./storage-multipart.js";
+import { startStorageFaultProxy } from "./storage-proxy.js";
+import { partNumberOf, s3Operation } from "./storage-s3-ops.js";
+
+const binary = minioBinary();
+const sdk = createRequire(new URL("../../server/package.json", import.meta.url))("@aws-sdk/client-s3");
+const MIB = 1024 * 1024;
+
+function seeded(values: number[]) {
+  let index = 0;
+  return () => values[index++ % values.length]!;
+}
+
+describe("S3 operation classification", () => {
+  it("names multipart, object and bucket operations from method and query", () => {
+    const key = "/b/chaos/w/c/workspace-snapshots/ws/s/u.pcws";
+    expect(s3Operation("POST", `${key}?uploads=&x-id=CreateMultipartUpload`)).toBe("CreateMultipartUpload");
+    expect(s3Operation("PUT", `${key}?partNumber=3&uploadId=abc&x-id=UploadPart`)).toBe("UploadPart");
+    expect(partNumberOf(`${key}?partNumber=3&uploadId=abc`)).toBe(3);
+    expect(s3Operation("POST", `${key}?uploadId=abc`)).toBe("CompleteMultipartUpload");
+    expect(s3Operation("DELETE", `${key}?uploadId=abc`)).toBe("AbortMultipartUpload");
+    expect(s3Operation("GET", "/b?uploads=&prefix=chaos")).toBe("ListMultipartUploads");
+    expect(s3Operation("PUT", key)).toBe("PutObject");
+    expect(s3Operation("GET", `${key}?x-id=GetObject`)).toBe("GetObject");
+    expect(s3Operation("HEAD", key)).toBe("HeadObject");
+    expect(s3Operation("GET", "/b?list-type=2&prefix=x")).toBe("ListObjects");
+    expect(s3Operation("POST", "/b?delete=")).toBe("DeleteObjects");
+  });
+});
+
+describe("ST4 and ST7 planning", () => {
+  it("plans ST4 as seeded faults on snapshot upload parts and sometimes on the completion", () => {
+    const rules = multipartFaultRules(seeded([0.1, 0.9, 0.1, 0.9, 0.2, 0.0, 0.4, 0.1, 0.0, 0.0, 0.0, 0.7]));
+    expect(rules).toEqual([
+      { fault: { kind: "errors", rate: 1, status: 500 }, operations: ["UploadPart"], partNumbers: [1, 3], pathIncludes: SNAPSHOT_PATH, remaining: 1 },
+      { fault: { kind: "errors", rate: 1, status: 500 }, operations: ["CompleteMultipartUpload"], pathIncludes: SNAPSHOT_PATH, remaining: 2 },
+    ]);
+    const kinds = new Set<string>();
+    for (let seed = 1; seed < 200; seed += 1) {
+      for (const window of planStorageFaults({ kinds: ["ST4"], seed, spanMs: 100_000 })) {
+        for (const rule of window.rules) { kinds.add(`${rule.operations?.join()}:${rule.fault.kind}`); expect(rule.pathIncludes).toBe(SNAPSHOT_PATH); }
+      }
+    }
+    expect([...kinds].sort()).toEqual(["CompleteMultipartUpload:errors", "CompleteMultipartUpload:reset", "UploadPart:errors", "UploadPart:reset", "UploadPart:truncate"]);
+  });
+
+  it("plans ST7 as a seeded read-after-write lag on object reads", () => {
+    expect(lagFaultRules(() => 0.5)).toEqual([{ fault: { kind: "read_after_write_lag", lagMs: 17_500 }, methods: ["GET", "HEAD"], operations: ["GetObject", "HeadObject"] }]);
+    const [rule] = storageFaultRules("ST7", () => 1);
+    expect(rule!.fault).toEqual({ kind: "read_after_write_lag", lagMs: LAG_MS_RANGE[1] });
+  });
+
+  it("audits storage only on s3 waves that plan or replay ST4 or ST7", () => {
+    expect(auditsStorage({ storage: "s3", storageFaults: ["ST4"] })).toBe(true);
+    expect(auditsStorage({ storage: "s3", storageWindows: [{ kind: "ST7" }] })).toBe(true);
+    expect(auditsStorage({ storage: "s3", storageFaults: ["ST1"] })).toBe(false);
+    expect(auditsStorage({ storageFaults: ["ST4"] })).toBe(false);
+  });
+
+  it("names each planned ST4 or ST7 fault that never changed a request", () => {
+    expect(unexercisedFaults(["ST4", "ST7", "ST4", "ST1"], { "PutObject:outage": 3 })).toEqual(["ST4", "ST7"]);
+    expect(unexercisedFaults(["ST4", "ST7"], { "UploadPart:truncate": 1, "HeadObject:read_after_write_lag": 2 })).toEqual([]);
+    expect(unexercisedFaults(["ST4"], { "CompleteMultipartUpload:errors": 1 })).toEqual([]);
+  });
+
+  it("forwards a truncated part's first bytes to the store before it cuts both connections", async () => {
+    let received = 0;
+    let closed!: () => void;
+    const upstreamClosed = new Promise<void>(resolve => { closed = resolve; });
+    const server = createServer(incoming => {
+      incoming.on("data", (chunk: Buffer) => { received += chunk.length; });
+      incoming.on("close", () => closed());
+    });
+    server.listen(0, "127.0.0.1"); await once(server, "listening");
+    const proxy = await startStorageFaultProxy(`http://127.0.0.1:${(server.address() as { port: number }).port}`);
+    try {
+      proxy.setRules([{ fault: { kind: "truncate", afterBytes: MIB }, operations: ["UploadPart"] }]);
+      await expect(fetch(`${proxy.endpoint}/b/k?partNumber=1&uploadId=u`, { method: "PUT", body: Buffer.alloc(4 * MIB) })).rejects.toThrow();
+      await upstreamClosed;
+      expect(received).toBeGreaterThan(0);
+      expect(received).toBeLessThan(4 * MIB);
+      expect(proxy.counts()).toMatchObject({ injected: 1, injectedBy: { "UploadPart:truncate": 1 } });
+    } finally {
+      await proxy.close();
+      server.closeAllConnections(); server.close();
+    }
+  });
+
+  it("passes a part shorter than the truncation point through untouched", async () => {
+    let received = 0;
+    const server = createServer((incoming, outgoing) => {
+      incoming.on("data", (chunk: Buffer) => { received += chunk.length; });
+      incoming.on("end", () => outgoing.writeHead(200, { etag: "\"p\"" }).end());
+    });
+    server.listen(0, "127.0.0.1"); await once(server, "listening");
+    const proxy = await startStorageFaultProxy(`http://127.0.0.1:${(server.address() as { port: number }).port}`);
+    try {
+      proxy.setRules([{ fault: { kind: "truncate", afterBytes: MIB }, operations: ["UploadPart"] }]);
+      const response = await fetch(`${proxy.endpoint}/b/k?partNumber=9&uploadId=u`, { method: "PUT", body: Buffer.alloc(1024) });
+      expect([response.status, received]).toEqual([200, 1024]);
+      expect(proxy.counts()).toMatchObject({ injected: 0, injectedBy: {} });
+    } finally {
+      await proxy.close();
+      server.closeAllConnections(); server.close();
+    }
+  });
+
+  it("lags reads of a key only after a successful object write and only for the lag", () => {
+    let now = 1_000;
+    const tracker = new ReadAfterWriteTracker(() => now);
+    tracker.note("PUT", "/b/k?partNumber=1&uploadId=u", 200);
+    tracker.note("PUT", "/b/failed", 503);
+    expect(tracker.lagging("/b/k", 5_000)).toBe(false);
+    expect(tracker.lagging("/b/failed", 5_000)).toBe(false);
+    tracker.note("POST", "/b/k?uploadId=u", 200);
+    tracker.note("PUT", "/b/a%20b", 200);
+    now += 4_000;
+    expect(tracker.lagging("/b/k?x-id=GetObject", 5_000)).toBe(true);
+    expect(tracker.lagging("/b/a b", 5_000)).toBe(true);
+    now += 1_000;
+    expect(tracker.lagging("/b/k", 5_000)).toBe(false);
+    expect(tracker.lagged).toBe(2);
+  });
+
+  it("reports orphaned uploads and unrestorable committed snapshots as wave failures", () => {
+    expect(storageAuditFailures({ incompleteUploads: [{ key: "k", uploadId: "u", initiated: null }], committedSnapshots: 1,
+      snapshotProblems: [{ id: "s", objectKey: "c/k", problem: "object_missing" }], snapshotsChecked: true, unexercisedFaults: [], errors: [] }))
+      .toEqual(["orphaned multipart upload k (u, initiated unknown)", "committed snapshot s does not restore: object_missing"]);
+    expect(storageAuditFailures({ incompleteUploads: [], committedSnapshots: 0, snapshotProblems: [], snapshotsChecked: false,
+      unexercisedFaults: ["ST4"], errors: [] })).toEqual(["storage fault ST4 never changed a request: the wave did not test it",
+      "storage audit incomplete: committed snapshots were not checked"]);
+    expect(storedKey("chaos/w1/", "c/workspace-snapshots/x")).toBe("chaos/w1/c/workspace-snapshots/x");
+  });
+});
+
+describe.skipIf(!binary)("ST4 and ST7 against MinIO", () => {
+  let root: string;
+  let store: ObjectStore;
+  let now = 0;
+  const tracker = new ReadAfterWriteTracker(() => now);
+  let proxy: Awaited<ReturnType<typeof startStorageFaultProxy>>;
+  let client: InstanceType<typeof sdk.S3Client>;
+  const key = `chaos/w1/c/${SNAPSHOT_PATH}/ws/s/u.pcws`;
+
+  beforeAll(async () => {
+    root = await mkdtemp(path.join(os.tmpdir(), "restart-chaos-st4-"));
+    store = await startMinio({ binary: binary!, root });
+    proxy = await startStorageFaultProxy(store.endpoint, Math.random, tracker);
+    client = new sdk.S3Client({ endpoint: proxy.endpoint, region: store.region, forcePathStyle: true, maxAttempts: 1,
+      credentials: { accessKeyId: store.accessKeyId, secretAccessKey: store.secretAccessKey } });
+  }, 60_000);
+  afterAll(async () => {
+    client?.destroy();
+    await proxy?.close();
+    await store?.stop();
+    await rm(root, { recursive: true, force: true });
+  });
+
+  const part = (uploadId: string, number: number, bytes = 5 * MIB) => client.send(new sdk.UploadPartCommand({
+    Bucket: store.bucket, Key: key, UploadId: uploadId, PartNumber: number, Body: Buffer.alloc(bytes, number) }));
+
+  it("fails the chosen upload parts, cuts a part body mid-transfer, fails the completion, and finds the orphaned upload", async () => {
+    const { UploadId: uploadId } = await client.send(new sdk.CreateMultipartUploadCommand({ Bucket: store.bucket, Key: key }));
+    proxy.setRules([{ fault: { kind: "errors", rate: 1, status: 500 }, operations: ["UploadPart"], partNumbers: [2], pathIncludes: SNAPSHOT_PATH, remaining: 1 }]);
+    const first = await part(uploadId, 1);
+    await expect(part(uploadId, 2)).rejects.toMatchObject({ $metadata: { httpStatusCode: 500 } });
+    const second = await part(uploadId, 2);
+    proxy.setRules([{ fault: { kind: "truncate", afterBytes: MIB }, operations: ["UploadPart"], pathIncludes: SNAPSHOT_PATH, remaining: 1 }]);
+    await expect(part(uploadId, 3, 2 * MIB)).rejects.toThrow();
+    const third = await part(uploadId, 3, 2 * MIB);
+    expect(await listIncompleteUploads(store, "chaos/w1/")).toEqual([{ key, uploadId, initiated: expect.any(String) }]);
+    proxy.setRules([{ fault: { kind: "errors", rate: 1, status: 500 }, operations: ["CompleteMultipartUpload"], pathIncludes: SNAPSHOT_PATH, remaining: 1 }]);
+    const complete = () => client.send(new sdk.CompleteMultipartUploadCommand({ Bucket: store.bucket, Key: key, UploadId: uploadId,
+      MultipartUpload: { Parts: [first, second, third].map((result, index) => ({ ETag: result.ETag, PartNumber: index + 1 })) } }));
+    await expect(complete()).rejects.toMatchObject({ $metadata: { httpStatusCode: 500 } });
+    expect(await listIncompleteUploads(store, "chaos/w1/")).toHaveLength(1);
+    proxy.clear();
+    await complete();
+    expect(tracker.lagging(`/${store.bucket}/${key}`, 1)).toBe(true);
+    expect(await listIncompleteUploads(store, "chaos/w1/")).toEqual([]);
+    expect(proxy.counts()).toMatchObject({ injected: 3, byStatus: { 500: 2, reset: 1 },
+      injectedBy: { "UploadPart:errors": 1, "UploadPart:truncate": 1, "CompleteMultipartUpload:errors": 1 } });
+  }, 60_000);
+
+  it("answers NoSuchKey to reads of a just-written object until the lag passes, and verifies committed snapshots", async () => {
+    now = 10_000;
+    const body = Buffer.from("snapshot bytes");
+    const written = `chaos/w1/c/${SNAPSHOT_PATH}/ws/s2/v.pcws`;
+    proxy.setRules(lagFaultRules(() => 0));
+    await client.send(new sdk.PutObjectCommand({ Bucket: store.bucket, Key: written, Body: body }));
+    now += LAG_MS_RANGE[0] - 1;
+    await expect(client.send(new sdk.GetObjectCommand({ Bucket: store.bucket, Key: written }))).rejects.toMatchObject({ name: "NoSuchKey" });
+    await expect(client.send(new sdk.HeadObjectCommand({ Bucket: store.bucket, Key: written }))).rejects.toMatchObject({ $metadata: { httpStatusCode: 404 } });
+    now += 1;
+    const read = await client.send(new sdk.GetObjectCommand({ Bucket: store.bucket, Key: written }));
+    expect(Buffer.from(await read.Body.transformToByteArray()).toString()).toBe("snapshot bytes");
+    proxy.clear();
+    const sha256 = createHash("sha256").update(body).digest("hex");
+    expect(await checkCommittedSnapshots(store, "chaos/w1/", [
+      { id: "good", objectKey: `c/${SNAPSHOT_PATH}/ws/s2/v.pcws`, sha256, byteSize: body.length },
+      { id: "corrupt", objectKey: `c/${SNAPSHOT_PATH}/ws/s2/v.pcws`, sha256: "0".repeat(64), byteSize: body.length },
+      { id: "missing", objectKey: `c/${SNAPSHOT_PATH}/ws/s3/none.pcws`, sha256, byteSize: body.length },
+    ])).toEqual([
+      { id: "corrupt", objectKey: `c/${SNAPSHOT_PATH}/ws/s2/v.pcws`, problem: "sha256_mismatch" },
+      { id: "missing", objectKey: `c/${SNAPSHOT_PATH}/ws/s3/none.pcws`, problem: "object_missing" },
+    ]);
+  }, 60_000);
+});

tests/restart-chaos/storage-multipart.ts

@@ -0,0 +1,26 @@
+import type { StorageFault, StorageFaultRule } from "./storage-proxy.js";
+
+export const SNAPSHOT_PATH = "workspace-snapshots";
+const MIB = 1024 * 1024;
+
+function pick<T>(random: () => number, items: readonly T[]): T {
+  return items[Math.floor(random() * items.length)]!;
+}
+
+export function multipartFaultRules(random: () => number): StorageFaultRule[] {
+  const parts = [1, 2, 3, 4].filter(() => random() < 0.5);
+  if (!parts.length) parts.push(1 + Math.floor(random() * 4));
+  const remaining = random() < 0.5 ? 1 + Math.floor(random() * 3) : undefined;
+  const partFault: StorageFault = pick<StorageFault>(random, [
+    { kind: "errors", rate: 1, status: pick(random, [500, 503]) },
+    { kind: "reset" },
+    { kind: "truncate", afterBytes: (1 + Math.floor(random() * 4)) * MIB },
+  ]);
+  const rules: StorageFaultRule[] = [{ fault: partFault, operations: ["UploadPart"], partNumbers: parts, pathIncludes: SNAPSHOT_PATH,
+    ...(remaining === undefined ? {} : { remaining }) }];
+  if (random() < 0.5) {
+    rules.push({ fault: pick<StorageFault>(random, [{ kind: "errors", rate: 1, status: 500 }, { kind: "reset" }]),
+      operations: ["CompleteMultipartUpload"], pathIncludes: SNAPSHOT_PATH, remaining: 1 + Math.floor(random() * 2) });
+  }
+  return rules;
+}

tests/restart-chaos/storage-proxy.ts

@@ -2,6 +2,8 @@ import { Agent, createServer, request, type IncomingMessage, type ServerResponse
 import { once } from "node:events";
 import type { Socket } from "node:net";
 import { linkRates, paced, releaseLinks, type LinkRates } from "./storage-bandwidth.js";
+import { ReadAfterWriteTracker } from "./storage-lag.js";
+import { partNumberOf, s3Operation, type S3Operation } from "./storage-s3-ops.js";
 
 export type StorageFault =
   | { kind: "pass" }
@@ -10,7 +12,9 @@ export type StorageFault =
   | { kind: "errors"; rate: number; status?: number; slowDown?: boolean }
   | { kind: "reset" }
   | { kind: "expired_credentials"; code?: CredentialErrorCode }
-  | { kind: "credential_rotation" };
+  | { kind: "credential_rotation" }
+  | { kind: "truncate"; afterBytes: number }
+  | { kind: "read_after_write_lag"; lagMs: number };
 
 export const CREDENTIAL_ERROR_CODES = ["ExpiredToken", "InvalidAccessKeyId", "SignatureDoesNotMatch"] as const;
 export type CredentialErrorCode = typeof CREDENTIAL_ERROR_CODES[number];
@@ -20,11 +24,13 @@ const CREDENTIAL_ERROR_MESSAGES: Record<CredentialErrorCode, string> = {
   SignatureDoesNotMatch: "The request signature we calculated does not match the signature you provided.",
 };
 
-export interface StorageFaultRule { fault: StorageFault; methods?: readonly string[]; pathIncludes?: string; remaining?: number }
+export interface StorageFaultRule { fault: StorageFault; methods?: readonly string[]; pathIncludes?: string; remaining?: number;
+  operations?: readonly S3Operation[]; partNumbers?: readonly number[] }
 
 export type StorageRequestListener = (incoming: IncomingMessage, outgoing: ServerResponse) => number | void;
 
-export interface StorageProxyCounts { requests: number; injected: number; byMethod: Record<string, number>; byStatus: Record<string, number> }
+export interface StorageProxyCounts { requests: number; injected: number; byMethod: Record<string, number>; byStatus: Record<string, number>;
+  injectedBy?: Record<string, number> }
 
 const XML = "application/xml";
 function s3Error(code: string, message: string) {
@@ -46,36 +52,42 @@ export function faultResponse(fault: StorageFault): { status: number; body: stri
 export function matches(rule: StorageFaultRule, method: string, url: string): boolean {
   if (rule.remaining !== undefined && rule.remaining <= 0) return false;
   if (rule.methods && !rule.methods.includes(method)) return false;
+  if (rule.operations && !rule.operations.includes(s3Operation(method, url))) return false;
+  if (rule.partNumbers && !rule.partNumbers.includes(partNumberOf(url) ?? 0)) return false;
   return !rule.pathIncludes || url.includes(rule.pathIncludes);
 }
 
-export async function startStorageFaultProxy(endpoint: string, random: () => number = Math.random) {
+export async function startStorageFaultProxy(endpoint: string, random: () => number = Math.random,
+  writes: ReadAfterWriteTracker = new ReadAfterWriteTracker()) {
   const target = new URL(endpoint);
   if (target.protocol !== "http:" || !["127.0.0.1", "localhost", "[::1]"].includes(target.hostname)) {
     throw new Error("Storage fault injection requires a loopback store");
   }
   let rules: StorageFaultRule[] = [];
   let listener: StorageRequestListener | null = null;
-  const counts: StorageProxyCounts = { requests: 0, injected: 0, byMethod: {}, byStatus: {} };
+  const counts: StorageProxyCounts = { requests: 0, injected: 0, byMethod: {}, byStatus: {}, injectedBy: {} };
   const sockets = new Set<Socket>();
   const delayed = new Set<ReturnType<typeof setTimeout>>();
   const agent = new Agent({ keepAlive: true });
   const links = new Map<StorageFaultRule, LinkRates>();
   const endLinks = () => { releaseLinks(links.values()); links.clear(); };
-  const note = (method: string, status: number | "reset", injected: boolean) => {
+  const note = (method: string, status: number | "reset", injected: boolean, cause?: string) => {
     counts.requests += 1;
     counts.byMethod[method] = (counts.byMethod[method] ?? 0) + 1;
     counts.byStatus[String(status)] = (counts.byStatus[String(status)] ?? 0) + 1;
     if (injected) counts.injected += 1;
+    if (cause) counts.injectedBy![cause] = (counts.injectedBy![cause] ?? 0) + 1;
   };
   const forward = (incoming: IncomingMessage, outgoing: ServerResponse, link?: LinkRates) => {
     const upstream = request(target, { method: incoming.method, path: incoming.url, headers: incoming.headers, agent }, response => {
       response.on("error", () => outgoing.destroy());
       note(incoming.method ?? "GET", response.statusCode ?? 502, false);
+      writes.note(incoming.method ?? "GET", incoming.url ?? "/", response.statusCode ?? 502);
       outgoing.writeHead(response.statusCode ?? 502, response.headers);
       (link ? response.pipe(paced(link.down)) : response).pipe(outgoing);
     });
     upstream.on("error", () => {
+      if (outgoing.destroyed) return;
       if (outgoing.headersSent) { outgoing.destroy(); return; }
       note(incoming.method ?? "GET", 502, false);
       outgoing.writeHead(502).end();
@@ -105,13 +117,23 @@ export async function startStorageFaultProxy(endpoint: string, random: () => num
       return;
     }
     if (fault.kind === "errors" && random() >= fault.rate) { forward(incoming, outgoing); return; }
+    if (fault.kind === "read_after_write_lag" && !writes.lagging(incoming.url ?? "/", fault.lagMs)) { forward(incoming, outgoing); return; }
+    const cause = `${s3Operation(method, incoming.url ?? "/")}:${fault.kind}`;
+    if (fault.kind === "truncate") {
+      let seen = 0;
+      const cut = () => { if (incoming.socket.destroyed) return; note(method, "reset", true, cause); incoming.socket.destroy(); };
+      incoming.on("data", (chunk: Buffer) => { seen += chunk.length; if (seen >= fault.afterBytes) cut(); });
+      forward(incoming, outgoing);
+      return;
+    }
     const injected = fault.kind === "errors"
       ? { status: fault.status ?? 503, body: fault.slowDown ? s3Error("SlowDown", "Please reduce your request rate.") : s3Error("InternalError", "Injected storage error") }
+      : fault.kind === "read_after_write_lag" ? { status: 404, body: s3Error("NoSuchKey", "The specified key does not exist.") }
       : faultResponse(fault);
     if (!injected) { forward(incoming, outgoing); return; }
     incoming.resume();
-    if (injected === "reset") { note(method, "reset", true); incoming.socket.destroy(); return; }
-    note(method, injected.status, true);
+    if (injected === "reset") { note(method, "reset", true, cause); incoming.socket.destroy(); return; }
+    note(method, injected.status, true, cause);
     outgoing.writeHead(injected.status, { "content-type": XML, "content-length": Buffer.byteLength(injected.body) }).end(injected.body);
   });
   server.on("connection", socket => { sockets.add(socket); socket.on("close", () => sockets.delete(socket)); });
@@ -120,7 +142,7 @@ export async function startStorageFaultProxy(endpoint: string, random: () => num
   if (!address || typeof address === "string") throw new Error("Storage proxy did not bind");
   return {
     endpoint: `http://127.0.0.1:${address.port}`,
-    counts: () => structuredClone(counts),
+    counts: () => ({ ...structuredClone(counts), lagged: writes.lagged }),
     setRules(next: readonly StorageFaultRule[]) { endLinks(); rules = next.map(rule => ({ ...rule })); },
     clear() { endLinks(); rules = []; },
     observe(next: StorageRequestListener | null) { listener = next; },

tests/restart-chaos/storage-s3-ops.ts

@@ -0,0 +1,41 @@
+export type S3Operation =
+  | "CreateMultipartUpload" | "UploadPart" | "CompleteMultipartUpload" | "AbortMultipartUpload" | "ListParts" | "ListMultipartUploads"
+  | "PutObject" | "GetObject" | "HeadObject" | "DeleteObject" | "DeleteObjects" | "ListObjects" | "Other";
+
+function parse(url: string) {
+  return new URL(url, "http://store.invalid");
+}
+
+export function objectPath(url: string): string {
+  return decodeURIComponent(parse(url).pathname);
+}
+
+export function partNumberOf(url: string): number | null {
+  const value = parse(url).searchParams.get("partNumber");
+  return value === null ? null : Number(value);
+}
+
+export function s3Operation(method: string, url: string): S3Operation {
+  const parsed = parse(url);
+  const query = parsed.searchParams;
+  const isBucket = parsed.pathname.split("/").filter(Boolean).length <= 1;
+  if (query.has("uploadId")) {
+    switch (method) {
+      case "PUT": return "UploadPart";
+      case "POST": return "CompleteMultipartUpload";
+      case "DELETE": return "AbortMultipartUpload";
+      case "GET": return "ListParts";
+      default: return "Other";
+    }
+  }
+  if (query.has("uploads")) return method === "POST" ? "CreateMultipartUpload" : method === "GET" ? "ListMultipartUploads" : "Other";
+  if (method === "POST" && query.has("delete")) return "DeleteObjects";
+  if (isBucket) return method === "GET" ? "ListObjects" : "Other";
+  switch (method) {
+    case "PUT": return "PutObject";
+    case "GET": return "GetObject";
+    case "HEAD": return "HeadObject";
+    case "DELETE": return "DeleteObject";
+    default: return "Other";
+  }
+}