From 7eae4bc2eaf5f93e79450e227e05536ecd987c95 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Tue, 10 Nov 2026 03:49:17 -0600 Subject: [PATCH] test(restart-chaos): add ST4 partial multipart and ST7 read-after-write lag storage faults (#20504) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 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 --- tests/restart-chaos/README.md | 9 + tests/restart-chaos/main.ts | 15 ++ tests/restart-chaos/storage-audit.ts | 145 ++++++++++++ tests/restart-chaos/storage-chaos.ts | 8 +- tests/restart-chaos/storage-lag.ts | 34 +++ .../storage-multipart-lag.test.ts | 220 ++++++++++++++++++ tests/restart-chaos/storage-multipart.ts | 26 +++ tests/restart-chaos/storage-proxy.ts | 40 +++- tests/restart-chaos/storage-s3-ops.ts | 41 ++++ 9 files changed, 527 insertions(+), 11 deletions(-) create mode 100644 tests/restart-chaos/storage-audit.ts create mode 100644 tests/restart-chaos/storage-lag.ts create mode 100644 tests/restart-chaos/storage-multipart-lag.test.ts create mode 100644 tests/restart-chaos/storage-multipart.ts create mode 100644 tests/restart-chaos/storage-s3-ops.ts diff --git a/tests/restart-chaos/README.md b/tests/restart-chaos/README.md index c863de70826..423d8ee0af1 100644 --- a/tests/restart-chaos/README.md +++ b/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 diff --git a/tests/restart-chaos/main.ts b/tests/restart-chaos/main.ts index 09e566b9fd8..861a3a9a686 100644 --- a/tests/restart-chaos/main.ts +++ b/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 { let storageTrigger: StorageTrigger | null = null; let storageProxy: Awaited> | 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 { 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 { 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 { 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 }) diff --git a/tests/restart-chaos/storage-audit.ts b/tests/restart-chaos/storage-audit.ts new file mode 100644 index 00000000000..452d77eb9ff --- /dev/null +++ b/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; + +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[] { + 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 { + 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 { + 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 { + 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 { + 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 }): Promise { + 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); + } +} diff --git a/tests/restart-chaos/storage-chaos.ts b/tests/restart-chaos/storage-chaos.ts index fd65297de12..a3e731783f1 100644 --- a/tests/restart-chaos/storage-chaos.ts +++ b/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 | 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) => void) {} diff --git a/tests/restart-chaos/storage-lag.ts b/tests/restart-chaos/storage-lag.ts new file mode 100644 index 00000000000..9720e1b7183 --- /dev/null +++ b/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(); + 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; + } +} diff --git a/tests/restart-chaos/storage-multipart-lag.test.ts b/tests/restart-chaos/storage-multipart-lag.test.ts new file mode 100644 index 00000000000..7518eb78b29 --- /dev/null +++ b/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(); + 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(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>; + let client: InstanceType; + 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); +}); diff --git a/tests/restart-chaos/storage-multipart.ts b/tests/restart-chaos/storage-multipart.ts new file mode 100644 index 00000000000..f3d9523ba3f --- /dev/null +++ b/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(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(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(random, [{ kind: "errors", rate: 1, status: 500 }, { kind: "reset" }]), + operations: ["CompleteMultipartUpload"], pathIncludes: SNAPSHOT_PATH, remaining: 1 + Math.floor(random() * 2) }); + } + return rules; +} diff --git a/tests/restart-chaos/storage-proxy.ts b/tests/restart-chaos/storage-proxy.ts index 42d95da93dd..fdaa529e218 100644 --- a/tests/restart-chaos/storage-proxy.ts +++ b/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 = { 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; byStatus: Record } +export interface StorageProxyCounts { requests: number; injected: number; byMethod: Record; byStatus: Record; + injectedBy?: Record } 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(); const delayed = new Set>(); const agent = new Agent({ keepAlive: true }); const links = new Map(); 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; }, diff --git a/tests/restart-chaos/storage-s3-ops.ts b/tests/restart-chaos/storage-s3-ops.ts new file mode 100644 index 00000000000..c573935a707 --- /dev/null +++ b/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"; + } +}