diff --git a/CHANGELOG.md b/CHANGELOG.md index d4dcb75..b79d3da 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,15 @@ ## Upcoming +- Reclaim object storage on stream expiry and deletion: a crash-safe retention + reaper deletes a doomed stream's remote objects (data first, manifest last) + and hard-deletes local rows once the prefix is verifiably empty; a periodic + object-store scan reaps doomed manifests whose creating node is gone; + restore-from-R2 recovers tombstoned/expired manifests as row-only tombstones; + recreating a deleted or expired stream waits for the old incarnation's + cleanup; the uploader skips segments of deleted streams. New env knobs: + `DS_RETENTION_REAP_LIMIT`, `DS_RETENTION_DELETE_CONCURRENCY`, + `DS_RETENTION_SCAN_MS`. - Add an `otel-traces` profile with OTLP JSON/protobuf/gzip ingest, canonical trace-span normalization, search aliases, privacy controls, and trace rollups. - Add request observability pairing descriptors and `POST /v1/observe/request` diff --git a/docs/architecture.md b/docs/architecture.md index 04d128a..a1d52d8 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -193,20 +193,44 @@ and if it is missing a background reconciliation pass can rebuild it from published segments plus retained WAL. Profiles and schemas only shape how a stream is interpreted. -## Stream Deletion Enforcement +## Stream Deletion Enforcement and Retention `DELETE /v1/stream/{name}` is enforced as a tombstone plus local acceleration scrub: -- the stream row stays in SQLite with the deleted flag set +- the stream row stays in SQLite with the deleted flag set until the reap below + completes - the same local delete transaction removes all stream-owned acceleration state: - routing index state and runs - exact secondary index state and runs - routing-key lexicon state and runs - bundled search companion plans and per-segment companion rows +- before acking, the manifest is republished carrying the deleted flag, so the + stream is a tombstone in remote object storage - the request path does not synchronously delete already-published remote segment, manifest, schema, or index objects +Remote cleanup is owned by the retention sweeper and the `StreamReaper` +(`src/retention.ts`, `src/retention_sweeper.ts`), which run as background loops +alongside the uploader and reconciler: + +- the expiry phase soft-deletes streams whose `expires_at` has passed + (`Stream-TTL` / `Stream-Expires-At`), converging expiry and DELETE on one + reapable state: the deleted-flagged local row +- the reap deletes the stream's remote objects data-first with `manifest.json` + strictly last, then hard-deletes local rows once the prefix is verifiably + empty; every step is idempotent and resumes after a crash +- restore-from-R2 recovers tombstoned or expired manifests as row-only + tombstones (no segment head checks), so a half-reaped prefix re-arms the reap + instead of aborting bootstrap +- a low-frequency object-store scan restores row-only tombstones for doomed + manifests that have no local row, which keeps retention converging on + deployments whose local SQLite is ephemeral across restarts +- recreating a deleted or expired stream name reaps the old incarnation inline + before the create, so two incarnations never interleave under one prefix +- the uploader skips segments of deleted streams, so an in-flight reap cannot + be re-populated by late uploads + Startup re-enforces the same invariant before background loops start. On boot, the server scans tombstoned streams and re-runs the acceleration scrub so older builds, crashes, or manual SQLite edits cannot leave orphaned async-index state diff --git a/docs/durable-streams-spec.md b/docs/durable-streams-spec.md index 481c35b..a9502f1 100644 --- a/docs/durable-streams-spec.md +++ b/docs/durable-streams-spec.md @@ -116,6 +116,8 @@ System streams (reserved names): Rules: - At most one of `Stream-TTL` and `Stream-Expires-At` may be provided. - If provided, the stream becomes unavailable for reads/appends after expiry. +- After expiry, the stream's local and remote storage is reclaimed + asynchronously (same ordering as DELETE: data objects first, manifest last). ### 3.3 Read request headers @@ -1112,8 +1114,12 @@ Headers: - exact secondary index state and runs - routing-key lexicon state and runs - bundled search companion plans and per-segment companion catalog rows -- Does not synchronously delete already-published segment, manifest, schema, or - index objects from remote object storage. +- Before acking, republishes the manifest carrying the deleted flag, so the + stream is a tombstone in remote object storage. +- Remote objects are then deleted asynchronously and idempotently: data objects + (segments, indexes, companions, schema) first, `manifest.json` strictly last, + and local state is fully removed only once the stream's remote prefix is + verifiably empty. Recreating the same name completes only after that cleanup. - Must be idempotent. --- diff --git a/docs/metrics.md b/docs/metrics.md index a36c86b..e8040cc 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -271,6 +271,17 @@ This implementation emits interval summaries for: - `tieredstore.backpressure.limit.bytes` - `tieredstore.backpressure.pressure` +### Retention + +- `tieredstore.retention.streams_reaped` (tag `trigger`) +- `tieredstore.retention.objects_deleted` +- `tieredstore.retention.list_passes` +- `tieredstore.retention.reap.latency` (tag `outcome`) +- `tieredstore.retention.reap.failures` (tag `kind`) +- `tieredstore.retention.scan.latency` +- `tieredstore.retention.scan.manifests_checked` +- `tieredstore.retention.scan.tombstones_restored` + ### Process memory - `process.rss.bytes` diff --git a/docs/operational-notes.md b/docs/operational-notes.md index c7b3e2c..7f5e5d5 100644 --- a/docs/operational-notes.md +++ b/docs/operational-notes.md @@ -121,6 +121,9 @@ Companion-cache note: - `DS_OBJECTSTORE_RETRY_MAX_MS`: max backoff for retries (default 2s) - `DS_EXPIRY_SWEEP_MS`: expired stream sweep interval (default 60s; 0 disables) - `DS_EXPIRY_SWEEP_LIMIT`: expired streams per sweep tick (default 100) +- `DS_RETENTION_REAP_LIMIT`: soft-deleted streams reaped per sweep tick (default 20) +- `DS_RETENTION_DELETE_CONCURRENCY`: parallel object deletes within one stream's reap (default 4) +- `DS_RETENTION_SCAN_MS`: object-store scan interval for doomed manifests without a local row (default 6h; 0 disables) - `DS_METRICS_FLUSH_MS`: metrics flush interval (default 10s; 0 disables) - `DS_STATS_INTERVAL_MS`: stats log interval when using `--stats` (default 60s) - `DS_BACKPRESSURE_BUDGET_MS`: per-request queue-wait budget used by stats (default `DS_INGEST_FLUSH_MS + 1`) diff --git a/docs/recovery-integrity-runbook.md b/docs/recovery-integrity-runbook.md index 24da006..be16680 100644 --- a/docs/recovery-integrity-runbook.md +++ b/docs/recovery-integrity-runbook.md @@ -45,6 +45,24 @@ Correctness depends on explicit commit points in the write path: - applies chunked WAL GC up to that bound - This is the remote durability/visibility commit point. +5. Deletion commit points (expiry and DELETE): +- Local soft delete (`STREAM_FLAG_DELETED` on the stream row) records the + delete intent; the row is the durable resume token for cleanup. +- Tombstone manifest publish (deleted flag, or a past `expires_at` already in + the manifest) is the remote delete-intent commit: from here, restore from + object storage recovers the stream only as a row-only tombstone (no segment + head checks), so a half-cleaned prefix can never abort bootstrap. +- Remote data objects are deleted before `manifest.json`; the manifest going + away is the remote invisibility point. +- Local hard delete runs only after the stream's remote prefix is verifiably + empty; this is the final commit. A crash anywhere earlier re-runs the reap + idempotently on the next sweep tick. +- An un-acked DELETE (5xx on the tombstone publish) still converges: the local + soft-deleted row drives the reaper, which republishes the tombstone first. +- A periodic scan restores row-only tombstones for doomed manifests that have + no local row (local SQLite is ephemeral across redeploys), so retention holds + even when the node that created a stream is gone. + ### 1.2 Core safety invariants For every stream, these invariants must hold: diff --git a/docs/sqlite-schema.md b/docs/sqlite-schema.md index 855262d..5e608bf 100644 --- a/docs/sqlite-schema.md +++ b/docs/sqlite-schema.md @@ -78,6 +78,10 @@ Additional columns present in the current implementation: - Retention/flags: - `expires_at_ms` - `stream_flags` + - Lifecycle: expiry or DELETE sets the deleted flag (soft delete); the row + survives as the durable resume token while the reaper clears the stream's + object-store prefix, and is hard-deleted (all per-stream rows removed) only + once the prefix is verifiably empty. - WAL accounting: - `logical_size_bytes` - `wal_rows` diff --git a/src/app_core.ts b/src/app_core.ts index ce6da79..f007aa0 100644 --- a/src/app_core.ts +++ b/src/app_core.ts @@ -22,7 +22,9 @@ import { } from "./schema/registry"; import { decodeJsonPayloadResult } from "./schema/read_json"; import { resolvePointerResult } from "./util/json_pointer"; -import { ExpirySweeper } from "./expiry_sweeper"; +import { NullObjectStore } from "./objectstore/null"; +import { StreamReaper } from "./retention"; +import { RetentionSweeper } from "./retention_sweeper"; import type { StatsCollector } from "./stats"; import { BackpressureGate } from "./backpressure"; import { MemoryPressureMonitor } from "./memory"; @@ -208,6 +210,14 @@ function unavailable(msg = "server shutting down"): Response { return json(503, { error: { code: "unavailable", message: msg } }, retryAfterHeaders(UNAVAILABLE_RETRY_AFTER_SECONDS)); } +function retentionInProgress(): Response { + return json( + 503, + { error: { code: "retention_in_progress", message: "stream cleanup in progress; retry" } }, + retryAfterHeaders("1") + ); +} + function overloaded(msg = "ingest queue full", code = "overloaded"): Response { return json(429, { error: { code, message: msg } }, retryAfterHeaders(OVERLOAD_RETRY_AFTER_SECONDS)); } @@ -465,6 +475,7 @@ export type App = { registry: SchemaRegistryStore; profiles: StreamProfileStore; touch: TouchProcessorManager; + reaper: StreamReaper; stats?: StatsCollector; backpressure?: BackpressureGate; memory?: MemoryPressureMonitor; @@ -1105,7 +1116,18 @@ export function createAppCore(cfg: Config, opts: CreateAppCoreOptions): App { }, collectRuntimeMetrics, }); - const expirySweeper = new ExpirySweeper(cfg, db); + // Without a remote store a stream's only state is local, so a reap is just + // the local cleanup and there is nothing for the retention scan to list. + const remoteRetention = !(store instanceof NullObjectStore); + const reaper = new StreamReaper(cfg, db, store, (stream) => uploader.publishManifest(stream), { + metrics, + diskCache: runtime.segmentDiskCache, + localOnly: !remoteRetention, + }); + const retentionSweeper = new RetentionSweeper(cfg, db, store, reaper, { + metrics, + scanEnabled: remoteRetention, + }); const streamSizeReconciler = new StreamSizeReconciler( db, store, @@ -1131,7 +1153,7 @@ export function createAppCore(cfg: Config, opts: CreateAppCoreOptions): App { })(); } metricsEmitter.start(); - expirySweeper.start(); + retentionSweeper.start(); touch.start(); streamSizeReconciler.start(); @@ -2608,12 +2630,15 @@ export function createAppCore(cfg: Config, opts: CreateAppCoreOptions): App { const bodyBytes = new Uint8Array(ab); let srow = db.getStream(stream); - if (srow && db.isDeleted(srow)) { - db.hardDeleteStream(stream); - srow = null; - } - if (srow && srow.expires_at_ms != null && db.nowMs() > srow.expires_at_ms) { - db.hardDeleteStream(stream); + const doomed = + srow != null && (db.isDeleted(srow) || (srow.expires_at_ms != null && db.nowMs() > srow.expires_at_ms)); + if (doomed) { + // The old incarnation's objects must be gone before the name is + // reused: new uploads reuse the same segment keys, so creating + // over remnants would interleave two incarnations under one + // prefix (and serve stale cached segment bytes). + const reapRes = await reaper.reapForRecreate(stream); + if (Result.isError(reapRes)) return retentionInProgress(); srow = null; } @@ -2721,6 +2746,7 @@ export function createAppCore(cfg: Config, opts: CreateAppCoreOptions): App { notifier.notifyDetailsChanged(stream); notifier.notifyClose(stream); await uploader.publishManifest(stream); + retentionSweeper.nudge(); return new Response(null, { status: 204, headers: withNosniff() }); } @@ -3341,7 +3367,7 @@ export function createAppCore(cfg: Config, opts: CreateAppCoreOptions): App { uploader.stop(true); await indexer?.stop(); metricsEmitter.stop(); - expirySweeper.stop(); + retentionSweeper.stop(); streamSizeReconciler.stop(); ingest.stop(); memorySampler?.stop(); @@ -3365,6 +3391,7 @@ export function createAppCore(cfg: Config, opts: CreateAppCoreOptions): App { metrics, registry, profiles, + reaper, touch, stats, backpressure, diff --git a/src/bootstrap.ts b/src/bootstrap.ts index 377fffc..dcaf873 100644 --- a/src/bootstrap.ts +++ b/src/bootstrap.ts @@ -1,7 +1,7 @@ import { mkdirSync, rmSync } from "node:fs"; import { dirname } from "node:path"; import type { Config } from "./config"; -import { SqliteDurableStore } from "./db/db"; +import { SqliteDurableStore, STREAM_FLAG_DELETED } from "./db/db"; import type { ObjectStore } from "./objectstore/interface"; import { zstdDecompressSync } from "./util/zstd"; import { localSegmentPath, schemaObjectKey, segmentObjectKey, streamHash16Hex } from "./util/stream_paths"; @@ -72,6 +72,16 @@ export async function bootstrapFromR2(cfg: Config, store: ObjectStore, opts: { c const ttlSeconds = typeof manifest.ttl_seconds === "number" ? manifest.ttl_seconds : null; const streamFlags = typeof manifest.stream_flags === "number" ? manifest.stream_flags : 0; + // Tombstoned (deleted) and expired manifests restore as a row-only + // tombstone: the retention reaper may already have removed segment and + // index objects under this prefix, so head-checking them here would abort + // the whole bootstrap. The restored row re-arms the reaper, which resumes + // the cleanup and hard-deletes the row once the prefix is empty. + if (manifestIsTombstone(manifest, nowMs)) { + restoreTombstoneRow(db, manifest, nowMs); + continue; + } + const segmentOffsetsBytes = decodeZstdBase64(manifest.segment_offsets ?? ""); const segmentBlocksBytes = decodeZstdBase64(manifest.segment_blocks ?? ""); const segmentLastTsBytes = decodeZstdBase64(manifest.segment_last_ts ?? ""); @@ -355,6 +365,53 @@ export async function bootstrapFromR2(cfg: Config, store: ObjectStore, opts: { c } } +/** A manifest whose stream can never serve again: deleted or past expiry. */ +export function manifestIsTombstone(manifest: Manifest, nowMs: bigint): boolean { + const streamFlags = typeof manifest.stream_flags === "number" ? manifest.stream_flags : 0; + const expiresAtMs = parseIsoMs(manifest.expires_at); + return (streamFlags & STREAM_FLAG_DELETED) !== 0 || (expiresAtMs != null && expiresAtMs <= nowMs); +} + +/** + * Restores a doomed stream as a row-only tombstone: enough identity for reads + * to answer 404/410 and for the retention reaper to resume cleanup, without + * head-checking segment objects that may already be deleted. + */ +export function restoreTombstoneRow(db: SqliteDurableStore, manifest: Manifest, nowMs: bigint): void { + const stream = String(manifest.name ?? ""); + if (!stream) return; + const nextOffsetNum = typeof manifest.next_offset === "number" ? manifest.next_offset : 0; + db.restoreStreamRow({ + stream, + created_at_ms: parseIsoMs(manifest.created_at) ?? nowMs, + updated_at_ms: nowMs, + content_type: typeof manifest.content_type === "string" ? manifest.content_type : "application/octet-stream", + profile: typeof manifest.profile === "string" && manifest.profile !== "" ? manifest.profile : "generic", + stream_seq: typeof manifest.stream_seq === "string" ? manifest.stream_seq : null, + closed: typeof manifest.closed === "number" ? manifest.closed : 0, + closed_producer_id: typeof manifest.closed_producer_id === "string" ? manifest.closed_producer_id : null, + closed_producer_epoch: typeof manifest.closed_producer_epoch === "number" ? manifest.closed_producer_epoch : null, + closed_producer_seq: typeof manifest.closed_producer_seq === "number" ? manifest.closed_producer_seq : null, + ttl_seconds: typeof manifest.ttl_seconds === "number" ? manifest.ttl_seconds : null, + epoch: typeof manifest.epoch === "number" ? manifest.epoch : 0, + next_offset: BigInt(nextOffsetNum), + sealed_through: -1n, + uploaded_through: -1n, + uploaded_segment_count: 0, + pending_rows: 0n, + pending_bytes: 0n, + logical_size_bytes: parseManifestBigInt(manifest.logical_size_bytes) ?? 0n, + wal_rows: 0n, + wal_bytes: 0n, + last_append_ms: nowMs, + last_segment_cut_ms: nowMs, + segment_in_progress: 0, + expires_at_ms: parseIsoMs(manifest.expires_at), + stream_flags: typeof manifest.stream_flags === "number" ? manifest.stream_flags : 0, + }); + db.upsertManifestRow(stream, Number(manifest.generation ?? 0), Number(manifest.generation ?? 0), nowMs, null, null); +} + function parseManifestBigInt(value: unknown): bigint | null { if (typeof value === "bigint") return value; if (typeof value === "number" && Number.isFinite(value)) return BigInt(Math.trunc(value)); diff --git a/src/config.ts b/src/config.ts index 2364078..fe42f50 100644 --- a/src/config.ts +++ b/src/config.ts @@ -63,6 +63,9 @@ export type Config = { objectStoreMaxDelayMs: number; expirySweepIntervalMs: number; expirySweepBatchLimit: number; + retentionReapBatchLimit: number; + retentionDeleteConcurrency: number; + retentionScanIntervalMs: number; metricsFlushIntervalMs: number; touchWorkers: number; touchCheckIntervalMs: number; @@ -140,6 +143,9 @@ const KNOWN_DS_ENVS = new Set([ "DS_LOCAL_DATA_ROOT", "DS_EXPIRY_SWEEP_MS", "DS_EXPIRY_SWEEP_LIMIT", + "DS_RETENTION_REAP_LIMIT", + "DS_RETENTION_DELETE_CONCURRENCY", + "DS_RETENTION_SCAN_MS", "DS_METRICS_FLUSH_MS", "DS_TOUCH_WORKERS", "DS_TOUCH_CHECK_MS", @@ -362,6 +368,9 @@ export function loadConfig(): Config { objectStoreMaxDelayMs: envNum("DS_OBJECTSTORE_RETRY_MAX_MS", 2000), expirySweepIntervalMs: envNum("DS_EXPIRY_SWEEP_MS", 60_000), expirySweepBatchLimit: envNum("DS_EXPIRY_SWEEP_LIMIT", 100), + retentionReapBatchLimit: envNum("DS_RETENTION_REAP_LIMIT", 20), + retentionDeleteConcurrency: envNum("DS_RETENTION_DELETE_CONCURRENCY", 4), + retentionScanIntervalMs: envNum("DS_RETENTION_SCAN_MS", 6 * 60 * 60 * 1000), metricsFlushIntervalMs: envNum("DS_METRICS_FLUSH_MS", 10_000), touchWorkers: envNum("DS_TOUCH_WORKERS", 1), touchCheckIntervalMs: envNum("DS_TOUCH_CHECK_MS", 250), diff --git a/src/db/db.ts b/src/db/db.ts index 19892ac..5d42997 100644 --- a/src/db/db.ts +++ b/src/db/db.ts @@ -177,6 +177,7 @@ export class SqliteDurableStore { candidateStreams: SqliteStatement; candidateStreamsNoInterval: SqliteStatement; listExpiredStreams: SqliteStatement; + listReapableStreams: SqliteStatement; streamWalRange: SqliteStatement; streamWalRangeByKey: SqliteStatement; @@ -376,6 +377,13 @@ export class SqliteDurableStore { ORDER BY expires_at_ms ASC LIMIT ?;` ), + listReapableStreams: this.db.query( + `SELECT stream + FROM streams + WHERE (stream_flags & ?) != 0 + ORDER BY updated_at_ms ASC + LIMIT ?;` + ), streamWalRange: this.db.query( `SELECT offset, ts_ms, routing_key, content_type, payload @@ -432,9 +440,10 @@ export class SqliteDurableStore { `UPDATE segments SET r2_etag=?, uploaded_at_ms=? WHERE segment_id=?;` ), pendingUploadHeads: this.db.query( - `SELECT segment_id, stream, segment_index, start_offset, end_offset, block_count, last_append_ms, payload_bytes, size_bytes, - local_path, created_at_ms, uploaded_at_ms, r2_etag + `SELECT s.segment_id, s.stream, s.segment_index, s.start_offset, s.end_offset, s.block_count, s.last_append_ms, s.payload_bytes, s.size_bytes, + s.local_path, s.created_at_ms, s.uploaded_at_ms, s.r2_etag FROM segments s + JOIN streams st ON st.stream = s.stream AND (st.stream_flags & ?) = 0 WHERE s.uploaded_at_ms IS NULL AND s.segment_index = ( SELECT MIN(s2.segment_index) @@ -1058,6 +1067,13 @@ export class SqliteDurableStore { return rows.map((r) => String(r.stream)); } + /** Soft-deleted streams awaiting object-store cleanup, oldest first. TOUCH + * streams are not excluded: an explicitly deleted touch stream must reap. */ + listReapableStreams(limit: number): string[] { + const rows = this.stmts.listReapableStreams.all(STREAM_FLAG_DELETED, limit) as any[]; + return rows.map((r) => String(r.stream)); + } + deleteAccelerationState(stream: string): void { const tx = this.db.transaction(() => { this.stmts.deleteIndexRunsForStream.run(stream); @@ -1582,7 +1598,9 @@ export class SqliteDurableStore { } pendingUploadHeads(limit: number): SegmentRow[] { - const rows = this.stmts.pendingUploadHeads.all(limit) as any[]; + // Deleted streams are excluded so a reap in progress cannot be re-populated + // by late segment uploads (which would also republish the manifest). + const rows = this.stmts.pendingUploadHeads.all(STREAM_FLAG_DELETED, limit) as any[]; return rows.map((r) => this.coerceSegmentRow(r)); } diff --git a/src/expiry_sweeper.ts b/src/expiry_sweeper.ts deleted file mode 100644 index 234185d..0000000 --- a/src/expiry_sweeper.ts +++ /dev/null @@ -1,44 +0,0 @@ -import type { Config } from "./config"; -import type { SqliteDurableStore } from "./db/db"; - -export class ExpirySweeper { - private readonly cfg: Config; - private readonly db: SqliteDurableStore; - private timer: any | null = null; - private running = false; - - constructor(cfg: Config, db: SqliteDurableStore) { - this.cfg = cfg; - this.db = db; - } - - start(): void { - if (this.timer || this.cfg.expirySweepIntervalMs <= 0) return; - this.timer = setInterval(() => { - void this.tick(); - }, this.cfg.expirySweepIntervalMs); - } - - stop(): void { - if (this.timer) clearInterval(this.timer); - this.timer = null; - } - - private async tick(): Promise { - if (this.running) return; - this.running = true; - try { - const expired = this.db.listExpiredStreams(this.cfg.expirySweepBatchLimit); - if (expired.length === 0) return; - for (const stream of expired) { - try { - this.db.deleteStream(stream); - } catch { - // ignore deletion errors - } - } - } finally { - this.running = false; - } - } -} diff --git a/src/objectstore/mock_r2.ts b/src/objectstore/mock_r2.ts index e4bd065..d92011f 100644 --- a/src/objectstore/mock_r2.ts +++ b/src/objectstore/mock_r2.ts @@ -9,11 +9,13 @@ export type MockR2Faults = { getDelayMs?: number; headDelayMs?: number; listDelayMs?: number; + deleteDelayMs?: number; failPutPrefix?: string; // fail PUT when key starts with prefix failPutEvery?: number; // fail every Nth PUT failGetEvery?: number; // fail every Nth GET failHeadEvery?: number; failListEvery?: number; + failDeleteEvery?: number; timeoutPutEvery?: number; timeoutGetEvery?: number; timeoutHeadEvery?: number; @@ -49,6 +51,7 @@ export class MockR2Store implements ObjectStore { private getCount = 0; private headCount = 0; private listCount = 0; + private deleteCount = 0; private memBytes = 0; constructor(opts: MockR2Options | MockR2Faults = {}) { @@ -215,6 +218,10 @@ export class MockR2Store implements ObjectStore { } async delete(key: string): Promise { + this.deleteCount++; + this.maybeFail(this.deleteCount, this.faults.failDeleteEvery, `MockR2: injected DELETE failure for ${key}`); + await sleep(this.faults.deleteDelayMs ?? 0); + const v = this.data.get(key); if (v?.bytes) this.memBytes -= v.bytes.byteLength; if (v?.path) { @@ -250,12 +257,13 @@ export class MockR2Store implements ObjectStore { return this.memBytes; } - stats(): { puts: number; gets: number; heads: number; lists: number; memoryBytes: number } { + stats(): { puts: number; gets: number; heads: number; lists: number; deletes: number; memoryBytes: number } { return { puts: this.putCount, gets: this.getCount, heads: this.headCount, lists: this.listCount, + deletes: this.deleteCount, memoryBytes: this.memBytes, }; } @@ -265,5 +273,6 @@ export class MockR2Store implements ObjectStore { this.getCount = 0; this.headCount = 0; this.listCount = 0; + this.deleteCount = 0; } } diff --git a/src/retention.ts b/src/retention.ts new file mode 100644 index 0000000..c831da2 --- /dev/null +++ b/src/retention.ts @@ -0,0 +1,263 @@ +import { rmSync } from "node:fs"; +import { Result } from "better-result"; +import type { Config } from "./config"; +import { ConcurrencyGate } from "./concurrency_gate"; +import { SqliteDurableStore, STREAM_FLAG_DELETED } from "./db/db"; +import type { Metrics } from "./metrics"; +import type { ObjectStore } from "./objectstore/interface"; +import type { SegmentDiskCache } from "./segment/cache"; +import { retry, type RetryOptions } from "./util/retry"; +import { manifestObjectKey, streamHash16Hex } from "./util/stream_paths"; + +export type ReapSummary = { + stream: string; + objectsDeleted: number; + listPasses: number; + skipped: boolean; +}; + +export type ReapError = { + kind: "tombstone_publish_failed" | "list_failed" | "delete_failed" | "prefix_not_stable" | "stopped"; + message: string; + key?: string; +}; + +// A reap that needs more passes than this is being raced by a writer; back off +// to the next sweep tick instead of spinning against it. +const MAX_LIST_PASSES = 5; + +/** + * Deletes a doomed stream's object-store data and, once the prefix is + * verifiably empty, its local rows. A stream is doomed when its local row + * carries STREAM_FLAG_DELETED; the row is the durable resume token, so a crash + * at any step re-runs the reap on the next sweep tick. + * + * Ordering is the crash-safety contract: data objects are deleted first and + * `manifest.json` strictly last. While the manifest exists the stream is + * visible to restore-from-R2 as a tombstone (deleted flag or past expiry), so + * a restore mid-reap re-arms the reap instead of resurrecting the stream. + */ +export class StreamReaper { + private readonly cfg: Config; + private readonly db: SqliteDurableStore; + private readonly store: ObjectStore; + private readonly publishTombstone: (stream: string) => Promise; + private readonly metrics?: Metrics; + private readonly diskCache?: SegmentDiskCache; + private readonly localOnly: boolean; + private readonly retryOpts: RetryOptions; + private readonly locks = new Map>(); + private stopped = false; + + constructor( + cfg: Config, + db: SqliteDurableStore, + store: ObjectStore, + publishTombstone: (stream: string) => Promise, + opts: { + metrics?: Metrics; + diskCache?: SegmentDiskCache; + /** Local mode has no remote objects; a reap is just the local cleanup. */ + localOnly?: boolean; + } = {} + ) { + this.cfg = cfg; + this.db = db; + this.store = store; + this.publishTombstone = publishTombstone; + this.metrics = opts.metrics; + this.diskCache = opts.diskCache; + this.localOnly = opts.localOnly ?? false; + this.retryOpts = { + retries: cfg.objectStoreRetries, + baseDelayMs: cfg.objectStoreBaseDelayMs, + maxDelayMs: cfg.objectStoreMaxDelayMs, + timeoutMs: cfg.objectStoreTimeoutMs, + }; + } + + stop(): void { + this.stopped = true; + } + + /** + * Serializes reaping and (in the resolution layer) hydration per stream, so + * a reap can never interleave with a restore or recreate of the same name. + */ + async withStreamLock(stream: string, fn: () => Promise): Promise { + const prev = this.locks.get(stream) ?? Promise.resolve(); + const run = prev.catch(() => undefined).then(fn); + const tail = run.catch(() => undefined); + this.locks.set(stream, tail); + try { + return await run; + } finally { + // Only clear when no later caller chained behind us. + if (this.locks.get(stream) === tail) this.locks.delete(stream); + } + } + + /** Idempotent; safe for non-reapable streams (returns ok skipped). */ + async reapStream(stream: string): Promise> { + return this.withStreamLock(stream, () => this.reapLocked(stream)); + } + + /** + * PUT-recreate entry: flags an expired-but-not-deleted row so the reap owns + * it, then reaps. A row that turned live again under the lock is left alone. + */ + async reapForRecreate(stream: string): Promise> { + return this.withStreamLock(stream, async () => { + const row = this.db.getStream(stream); + if (!row) return Result.ok(skippedSummary(stream)); + const expired = row.expires_at_ms != null && this.db.nowMs() > row.expires_at_ms; + if (!this.db.isDeleted(row) && !expired) return Result.ok(skippedSummary(stream)); + if (!this.db.isDeleted(row)) this.db.deleteStream(stream); + return this.reapLocked(stream); + }); + } + + private async reapLocked(stream: string): Promise> { + if (this.stopped) return Result.err({ kind: "stopped" as const, message: "reaper stopped" }); + const row = this.db.getStream(stream); + if (!row || !this.db.isDeleted(row)) return Result.ok(skippedSummary(stream)); + + const startedNs = process.hrtime.bigint(); + const res = await this.reapDoomed(stream); + if (this.metrics) { + const elapsedNs = Number(process.hrtime.bigint() - startedNs); + this.metrics.record("tieredstore.retention.reap.latency", elapsedNs, "ns", { + outcome: Result.isError(res) ? "error" : "ok", + }); + if (Result.isError(res)) { + this.metrics.record("tieredstore.retention.reap.failures", 1, "count", { kind: res.error.kind }); + } else if (!res.value.skipped) { + this.metrics.record("tieredstore.retention.objects_deleted", res.value.objectsDeleted, "count"); + this.metrics.record("tieredstore.retention.list_passes", res.value.listPasses, "count"); + } + } + return res; + } + + private async reapDoomed(stream: string): Promise> { + const shash = streamHash16Hex(stream); + const prefix = `streams/${shash}/`; + const mkey = manifestObjectKey(shash); + + if (this.localOnly) { + this.cleanupLocal(stream, shash, []); + return Result.ok({ stream, objectsDeleted: 0, listPasses: 0, skipped: false }); + } + + // The manifest must be terminal (deleted flag or past expiry) in R2 before + // any data object disappears: restore-from-R2 short-circuits terminal + // manifests, so a crash mid-reap can never brick bootstrap on a + // head-check of an already-deleted segment. + const terminal = await this.ensureTerminalManifest(stream, mkey); + if (Result.isError(terminal)) return terminal; + + let listPasses = 0; + let objectsDeleted = 0; + const deletedKeys: string[] = []; + for (;;) { + if (this.stopped) return Result.err({ kind: "stopped" as const, message: "reaper stopped" }); + listPasses++; + if (listPasses > MAX_LIST_PASSES) { + return Result.err({ + kind: "prefix_not_stable" as const, + message: `prefix ${prefix} still non-empty after ${MAX_LIST_PASSES} passes`, + }); + } + let keys: string[]; + try { + keys = await retry(() => this.store.list(prefix), this.retryOpts); + } catch (e) { + return Result.err({ kind: "list_failed" as const, message: errorMessage(e) }); + } + if (keys.length === 0) break; + + // Data objects first; the manifest goes only when it is the last key + // left, and the follow-up list pass verifies the prefix emptied. + const dataKeys = keys.filter((k) => k !== mkey); + const doomedKeys = dataKeys.length > 0 ? dataKeys : keys; + const failure = await this.deleteKeys(doomedKeys); + if (failure) return failure; + objectsDeleted += doomedKeys.length; + deletedKeys.push(...doomedKeys); + } + + this.cleanupLocal(stream, shash, deletedKeys); + return Result.ok({ stream, objectsDeleted, listPasses, skipped: false }); + } + + /** + * Local commit point: drops cached segment bytes (recreated streams reuse + * the same object keys), any never-uploaded local segment files, and finally + * every local row. + */ + private cleanupLocal(stream: string, shash: string, deletedKeys: string[]): void { + if (this.diskCache) { + for (const key of deletedKeys) this.diskCache.remove(key); + } + try { + rmSync(`${this.cfg.rootDir}/local/streams/${shash}`, { recursive: true, force: true }); + } catch { + // ignore + } + this.db.hardDeleteStream(stream); + } + + private async ensureTerminalManifest(stream: string, mkey: string): Promise> { + const readManifest = async (): Promise<{ terminal: boolean } | null> => { + const bytes = await retry(() => this.store.get(mkey), this.retryOpts); + if (!bytes) return null; + try { + const manifest = JSON.parse(new TextDecoder().decode(bytes)) as Record; + const flags = typeof manifest.stream_flags === "number" ? manifest.stream_flags : 0; + const expiresAt = typeof manifest.expires_at === "string" ? Date.parse(manifest.expires_at) : Number.NaN; + const expired = Number.isFinite(expiresAt) && BigInt(Math.trunc(expiresAt)) <= this.db.nowMs(); + return { terminal: (flags & STREAM_FLAG_DELETED) !== 0 || expired }; + } catch { + // An unparseable manifest cannot resurrect anything on restore; treat + // it as terminal so the reap can remove it. + return { terminal: true }; + } + }; + + try { + const current = await readManifest(); + if (!current || current.terminal) return Result.ok(undefined); + await this.publishTombstone(stream); + const republished = await readManifest(); + if (!republished || republished.terminal) return Result.ok(undefined); + return Result.err({ + kind: "tombstone_publish_failed" as const, + message: `manifest for ${stream} is still live after tombstone publish`, + }); + } catch (e) { + return Result.err({ kind: "tombstone_publish_failed" as const, message: errorMessage(e) }); + } + } + + private async deleteKeys(keys: string[]): Promise | null> { + const gate = new ConcurrencyGate(Math.max(1, this.cfg.retentionDeleteConcurrency)); + const settled = await Promise.allSettled( + keys.map((key) => gate.run(() => retry(() => this.store.delete(key), this.retryOpts))) + ); + for (let i = 0; i < settled.length; i++) { + const s = settled[i]; + if (s.status === "rejected") { + return Result.err({ kind: "delete_failed" as const, message: errorMessage(s.reason), key: keys[i] }); + } + } + return null; + } +} + +function skippedSummary(stream: string): ReapSummary { + return { stream, objectsDeleted: 0, listPasses: 0, skipped: true }; +} + +function errorMessage(e: unknown): string { + return e instanceof Error ? e.message : String(e); +} diff --git a/src/retention_sweeper.ts b/src/retention_sweeper.ts new file mode 100644 index 0000000..c8f7d61 --- /dev/null +++ b/src/retention_sweeper.ts @@ -0,0 +1,165 @@ +import { Result } from "better-result"; +import { manifestIsTombstone, restoreTombstoneRow } from "./bootstrap"; +import { ConcurrencyGate } from "./concurrency_gate"; +import type { Config } from "./config"; +import type { SqliteDurableStore } from "./db/db"; +import type { Metrics } from "./metrics"; +import type { ObjectStore } from "./objectstore/interface"; +import type { StreamReaper } from "./retention"; +import { FailureTracker } from "./uploader"; +import { retry, type RetryOptions } from "./util/retry"; + +/** + * Drives stream retention: flags expired streams (soft delete), feeds + * soft-deleted streams to the reaper, and periodically scans the object store + * for doomed manifests that no longer have a local row (the local SQLite is + * ephemeral on redeploy, so the object store is the only durable record). + */ +export class RetentionSweeper { + private readonly cfg: Config; + private readonly db: SqliteDurableStore; + private readonly store: ObjectStore; + private readonly reaper: StreamReaper; + private readonly metrics?: Metrics; + private readonly failures = new FailureTracker(1024); + private readonly retryOpts: RetryOptions; + private timer: any | null = null; + private scanTimer: any | null = null; + private running = false; + private scanRunning = false; + + private readonly scanEnabled: boolean; + + constructor( + cfg: Config, + db: SqliteDurableStore, + store: ObjectStore, + reaper: StreamReaper, + opts: { metrics?: Metrics; scanEnabled?: boolean } = {} + ) { + this.cfg = cfg; + this.db = db; + this.store = store; + this.reaper = reaper; + this.metrics = opts.metrics; + this.scanEnabled = opts.scanEnabled ?? true; + this.retryOpts = { + retries: cfg.objectStoreRetries, + baseDelayMs: cfg.objectStoreBaseDelayMs, + maxDelayMs: cfg.objectStoreMaxDelayMs, + timeoutMs: cfg.objectStoreTimeoutMs, + }; + } + + start(): void { + if (!this.timer && this.cfg.expirySweepIntervalMs > 0) { + this.timer = setInterval(() => { + void this.tick(); + }, this.cfg.expirySweepIntervalMs); + } + if (this.scanEnabled && !this.scanTimer && this.cfg.retentionScanIntervalMs > 0) { + // Jitter keeps a fleet of replicas from listing the store in lockstep. + const jitter = 0.9 + Math.random() * 0.2; + this.scanTimer = setInterval( + () => { + void this.scanTick(); + }, + Math.round(this.cfg.retentionScanIntervalMs * jitter) + ); + } + } + + stop(): void { + if (this.timer) clearInterval(this.timer); + this.timer = null; + if (this.scanTimer) clearInterval(this.scanTimer); + this.scanTimer = null; + this.reaper.stop(); + } + + /** Schedules an immediate tick (e.g. right after an HTTP DELETE). */ + nudge(): void { + void this.tick(); + } + + private async tick(): Promise { + if (this.running) return; + this.running = true; + try { + const expired = this.db.listExpiredStreams(this.cfg.expirySweepBatchLimit); + for (const stream of expired) { + try { + this.db.deleteStream(stream); + } catch { + // ignore deletion errors + } + } + + const reapable = this.db.listReapableStreams(this.cfg.retentionReapBatchLimit); + for (const stream of reapable) { + if (this.failures.shouldSkip(stream)) continue; + const res = await this.reaper.reapStream(stream); + if (Result.isError(res)) { + if (res.error.kind === "stopped") return; + this.failures.recordFailure(stream); + } else { + this.failures.recordSuccess(stream); + if (!res.value.skipped) { + this.metrics?.record("tieredstore.retention.streams_reaped", 1, "count", { trigger: "sweep" }); + } + } + } + } finally { + this.running = false; + } + } + + private async scanTick(): Promise { + if (this.scanRunning) return; + this.scanRunning = true; + const startedNs = process.hrtime.bigint(); + let manifestsChecked = 0; + let tombstonesRestored = 0; + try { + const keys = await retry(() => this.store.list("streams/"), this.retryOpts); + const manifestKeys = keys.filter((k) => k.endsWith("/manifest.json")); + const gate = new ConcurrencyGate(Math.max(1, this.cfg.retentionDeleteConcurrency)); + await Promise.allSettled( + manifestKeys.map((mkey) => + gate.run(async () => { + manifestsChecked++; + const bytes = await retry(() => this.store.get(mkey), this.retryOpts); + if (!bytes) return; + let manifest: Record; + try { + manifest = JSON.parse(new TextDecoder().decode(bytes)); + } catch { + return; + } + const stream = String(manifest.name ?? ""); + if (!stream) return; + if (!manifestIsTombstone(manifest, this.db.nowMs())) return; + // Serialize with reaps/hydrations of the same name; a local row of + // any kind wins and already drives its own lifecycle. + await this.reaper.withStreamLock(stream, async () => { + if (this.db.getStream(stream)) return; + restoreTombstoneRow(this.db, manifest, this.db.nowMs()); + tombstonesRestored++; + }); + }) + ) + ); + if (tombstonesRestored > 0) this.nudge(); + } catch { + // Transient list failures are retried on the next scan interval. + } finally { + this.scanRunning = false; + if (this.metrics) { + const elapsedNs = Number(process.hrtime.bigint() - startedNs); + this.metrics.record("tieredstore.retention.scan.latency", elapsedNs, "ns"); + this.metrics.record("tieredstore.retention.scan.manifests_checked", manifestsChecked, "count"); + this.metrics.record("tieredstore.retention.scan.tombstones_restored", tombstonesRestored, "count"); + } + } + } +} diff --git a/src/uploader.ts b/src/uploader.ts index 9fbb874..0a87bf0 100644 --- a/src/uploader.ts +++ b/src/uploader.ts @@ -356,7 +356,7 @@ export class Uploader { } } -class FailureTracker { +export class FailureTracker { private readonly cache: LruCache; constructor(maxEntries: number) { diff --git a/test/assumptions.test.ts b/test/assumptions.test.ts index 388bd86..03d1a4f 100644 --- a/test/assumptions.test.ts +++ b/test/assumptions.test.ts @@ -477,7 +477,7 @@ describe("assumptions", () => { rmSync(root, { recursive: true, force: true }); }); - test("delete is tombstone, listing excludes deleted, and acceleration state is scrubbed", async () => { + test("delete tombstones, hides the stream, scrubs acceleration state, and reaps", async () => { const root = mkdtempSync(join(tmpdir(), "ds-assume-")); const cfg = makeConfig(root); const app = createApp(cfg, new MockR2Store()); @@ -498,10 +498,16 @@ describe("assumptions", () => { const list = await fetch(`${baseUrl}/v1/streams`); const arr = await list.json(); expect(arr.find((x: any) => x.name === "del")).toBeUndefined(); + // The row survives only as a tombstone until the nudged reap clears the + // object-store prefix and hard-deletes it. const deletedRow = app.deps.db.getStream("del"); - expect(deletedRow).not.toBeNull(); - expect(deletedRow && app.deps.db.isDeleted(deletedRow)).toBe(true); + if (deletedRow) expect(app.deps.db.isDeleted(deletedRow)).toBe(true); expectAccelerationStateCleared(app.deps.db, "del"); + const deadline = Date.now() + 10_000; + while (Date.now() < deadline && app.deps.db.getStream("del") !== null) { + await sleep(25); + } + expect(app.deps.db.getStream("del")).toBeNull(); server.stop(); await app.close(); diff --git a/test/retention_reaper.test.ts b/test/retention_reaper.test.ts new file mode 100644 index 0000000..22a05de --- /dev/null +++ b/test/retention_reaper.test.ts @@ -0,0 +1,581 @@ +import { existsSync, mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { describe, expect, test } from "bun:test"; +import { Result } from "better-result"; +import { createApp } from "../src/app"; +import { bootstrapFromR2 } from "../src/bootstrap"; +import { loadConfig, type Config } from "../src/config"; +import { SqliteDurableStore } from "../src/db/db"; +import { MockR2Store, type MockR2Faults } from "../src/objectstore/mock_r2"; +import { manifestObjectKey, streamHash16Hex } from "../src/util/stream_paths"; + +const METRICS_STREAM = "__stream_metrics__"; + +function makeConfig(rootDir: string, overrides: Partial): Config { + const base = loadConfig(); + return { + ...base, + rootDir, + dbPath: `${rootDir}/wal.sqlite`, + port: 0, + segmentMaxBytes: 128, + segmentCheckIntervalMs: 25, + uploadIntervalMs: 25, + uploadConcurrency: 2, + segmentCacheMaxBytes: 0, + segmentFooterCacheEntries: 0, + expirySweepIntervalMs: 0, + retentionScanIntervalMs: 0, + objectStoreRetries: 0, + objectStoreBaseDelayMs: 5, + ...overrides, + }; +} + +function streamUrl(stream: string, suffix = ""): string { + return `http://local/v1/stream/${encodeURIComponent(stream)}${suffix}`; +} + +function prefixOf(stream: string): string { + return `streams/${streamHash16Hex(stream)}/`; +} + +function manifestKeyOf(stream: string): string { + return manifestObjectKey(streamHash16Hex(stream)); +} + +async function sleep(ms: number): Promise { + return new Promise((res) => setTimeout(res, ms)); +} + +async function until(cond: () => boolean | Promise, timeoutMs = 15_000, stepMs = 25): Promise { + const deadline = Date.now() + timeoutMs; + for (;;) { + if (await cond()) return true; + if (Date.now() >= deadline) return false; + await sleep(stepMs); + } +} + +type SeededApp = Awaited>; + +async function seedUploadedStream( + app: SeededApp, + store: MockR2Store, + stream: string, + opts: { records?: number; ttl?: string } = {} +): Promise { + const records = opts.records ?? 6; + const headers: Record = { "content-type": "application/json" }; + if (opts.ttl) headers["stream-ttl"] = opts.ttl; + const createRes = await app.fetch(new Request(streamUrl(stream), { method: "PUT", headers })); + expect([200, 201, 204]).toContain(createRes.status); + for (let i = 0; i < records; i++) { + const r = await app.fetch( + new Request(streamUrl(stream), { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ i, pad: "x".repeat(48) }), + }) + ); + expect(r.status).toBe(204); + } + const uploaded = await until(() => { + const row = app.deps.db.getStream(stream); + return ( + !!row && + row.uploaded_through >= row.sealed_through && + app.deps.db.countSegmentsForStream(stream) > 0 && + store.has(manifestKeyOf(stream)) + ); + }); + expect(uploaded).toBe(true); +} + +describe("retention reaper", () => { + test( + "expiry sweep reaps every object and hard-deletes the row", + async () => { + const root = mkdtempSync(join(tmpdir(), "ds-reap-expiry-")); + const store = new MockR2Store(); + const app = createApp(makeConfig(root, { expirySweepIntervalMs: 50 }), store); + const stream = "expires"; + try { + await seedUploadedStream(app, store, stream, { ttl: "2" }); + expect((await store.list(prefixOf(stream))).length).toBeGreaterThan(1); + + const reaped = await until( + async () => (await store.list(prefixOf(stream))).length === 0 && app.deps.db.getStream(stream) === null, + 20_000 + ); + expect(reaped).toBe(true); + expect(existsSync(`${root}/local/streams/${streamHash16Hex(stream)}`)).toBe(false); + } finally { + await app.close(); + rmSync(root, { recursive: true, force: true }); + } + }, + 30_000 + ); + + test( + "DELETE tombstones then the background reap empties the prefix", + async () => { + const root = mkdtempSync(join(tmpdir(), "ds-reap-delete-")); + const store = new MockR2Store(); + const app = createApp(makeConfig(root, { expirySweepIntervalMs: 50 }), store); + const stream = "deleted"; + try { + await seedUploadedStream(app, store, stream); + const delRes = await app.fetch(new Request(streamUrl(stream), { method: "DELETE" })); + expect(delRes.status).toBe(204); + + const reaped = await until( + async () => (await store.list(prefixOf(stream))).length === 0 && app.deps.db.getStream(stream) === null, + 20_000 + ); + expect(reaped).toBe(true); + + // Reap deletes flow through the accounting wrapper like any other op. + const counted = app.deps.db.db + .query(`SELECT COALESCE(SUM(count), 0) AS c FROM objectstore_request_counts WHERE op='delete';`) + .get() as { c: number }; + expect(Number(counted.c)).toBeGreaterThan(0); + } finally { + await app.close(); + rmSync(root, { recursive: true, force: true }); + } + }, + 30_000 + ); + + test( + "manifest.json outlives every data object during a reap", + async () => { + const root = mkdtempSync(join(tmpdir(), "ds-reap-order-")); + const faults: MockR2Faults = {}; + const store = new MockR2Store({ faults }); + const app = createApp( + makeConfig(root, { expirySweepIntervalMs: 50, retentionDeleteConcurrency: 1 }), + store + ); + const stream = "ordered"; + try { + await seedUploadedStream(app, store, stream, { records: 10 }); + const mkey = manifestKeyOf(stream); + faults.deleteDelayMs = 25; + + const delRes = await app.fetch(new Request(streamUrl(stream), { method: "DELETE" })); + expect(delRes.status).toBe(204); + + let violated = false; + const emptied = await until(async () => { + const keys = await store.list(prefixOf(stream)); + if (!keys.includes(mkey) && keys.length > 0) violated = true; + return keys.length === 0; + }, 20_000, 5); + expect(emptied).toBe(true); + expect(violated).toBe(false); + } finally { + await app.close(); + rmSync(root, { recursive: true, force: true }); + } + }, + 30_000 + ); + + test( + "a crashed reap resumes after restore-from-R2 on a fresh root", + async () => { + const root = mkdtempSync(join(tmpdir(), "ds-reap-crash-src-")); + const root2 = mkdtempSync(join(tmpdir(), "ds-reap-crash-dst-")); + const faults: MockR2Faults = {}; + const store = new MockR2Store({ faults }); + const stream = "crashy"; + + const app = createApp(makeConfig(root, { expirySweepIntervalMs: 50 }), store); + try { + await seedUploadedStream(app, store, stream, { records: 10 }); + const before = (await store.list(prefixOf(stream))).length; + expect(before).toBeGreaterThan(2); + + faults.failDeleteEvery = 2; + const delRes = await app.fetch(new Request(streamUrl(stream), { method: "DELETE" })); + expect(delRes.status).toBe(204); + + // A partial reap happened (some objects gone) but could not finish. + const partial = await until(async () => { + const left = (await store.list(prefixOf(stream))).length; + return left > 0 && left < before && store.stats().deletes > 0; + }); + expect(partial).toBe(true); + expect(app.deps.db.getStream(stream)).not.toBeNull(); + } finally { + await app.close(); + } + + faults.failDeleteEvery = undefined; + // Restore-from-R2 must not abort on the half-reaped prefix; the + // tombstone manifest restores a row that re-arms the reap. + await bootstrapFromR2(makeConfig(root2, {}), store, { clearLocal: true }); + const app2 = createApp(makeConfig(root2, { expirySweepIntervalMs: 50 }), store); + try { + const reaped = await until( + async () => (await store.list(prefixOf(stream))).length === 0 && app2.deps.db.getStream(stream) === null, + 20_000 + ); + expect(reaped).toBe(true); + } finally { + await app2.close(); + rmSync(root, { recursive: true, force: true }); + rmSync(root2, { recursive: true, force: true }); + } + }, + 30_000 + ); + + test( + "bootstrap restores a tombstone manifest without head-checking segments", + async () => { + const root = mkdtempSync(join(tmpdir(), "ds-reap-boot-src-")); + const root2 = mkdtempSync(join(tmpdir(), "ds-reap-boot-dst-")); + const store = new MockR2Store(); + const stream = "tombstoned"; + + const app = createApp(makeConfig(root, {}), store); + try { + await seedUploadedStream(app, store, stream); + // Soft-delete without the HTTP route so no reap is nudged: the data + // objects stay in place next to a tombstoned manifest. + app.deps.db.deleteStream(stream); + await app.deps.uploader.publishManifest(stream); + } finally { + await app.close(); + } + expect((await store.list(prefixOf(stream))).length).toBeGreaterThan(1); + + store.resetStats(); + await bootstrapFromR2(makeConfig(root2, {}), store, { clearLocal: true }); + const metricsManifests = store.has(manifestKeyOf(METRICS_STREAM)) ? 1 : 0; + expect(store.stats().heads).toBe(metricsManifests); + + const db = new SqliteDurableStore(`${root2}/wal.sqlite`); + try { + const row = db.getStream(stream); + expect(row).not.toBeNull(); + expect(db.isDeleted(row!)).toBe(true); + expect(db.countSegmentsForStream(stream)).toBe(0); + } finally { + db.close(); + rmSync(root, { recursive: true, force: true }); + rmSync(root2, { recursive: true, force: true }); + } + }, + 30_000 + ); + + test( + "PUT recreates a deleted stream on a clean prefix", + async () => { + const root = mkdtempSync(join(tmpdir(), "ds-reap-recreate-")); + const store = new MockR2Store(); + const app = createApp(makeConfig(root, {}), store); + const stream = "recreated"; + try { + await seedUploadedStream(app, store, stream, { records: 6 }); + const delRes = await app.fetch(new Request(streamUrl(stream), { method: "DELETE" })); + expect(delRes.status).toBe(204); + + const putRes = await app.fetch( + new Request(streamUrl(stream), { method: "PUT", headers: { "content-type": "application/json" } }) + ); + expect(putRes.status).toBe(201); + + const appendRes = await app.fetch( + new Request(streamUrl(stream), { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ generation: "second" }), + }) + ); + expect(appendRes.status).toBe(204); + + const readRes = await app.fetch(new Request(`${streamUrl(stream)}?offset=-1`, { method: "GET" })); + expect(readRes.status).toBe(200); + const body = await readRes.text(); + expect(body).toContain("second"); + expect(body).not.toContain('"pad"'); + } finally { + await app.close(); + rmSync(root, { recursive: true, force: true }); + } + }, + 30_000 + ); + + test( + "PUT recreates an expired stream after reaping the old incarnation", + async () => { + const root = mkdtempSync(join(tmpdir(), "ds-reap-expired-put-")); + const store = new MockR2Store(); + // Sweeper off: the expired row is discovered by the PUT itself. + const app = createApp(makeConfig(root, {}), store); + const stream = "expired-put"; + try { + await seedUploadedStream(app, store, stream, { ttl: "1" }); + const oldObjects = (await store.list(prefixOf(stream))).length; + expect(oldObjects).toBeGreaterThan(1); + await sleep(1_100); + + const putRes = await app.fetch( + new Request(streamUrl(stream), { method: "PUT", headers: { "content-type": "application/json" } }) + ); + expect(putRes.status).toBe(201); + + const row = app.deps.db.getStream(stream); + expect(row).not.toBeNull(); + expect(row!.expires_at_ms).toBeNull(); + expect(row!.next_offset).toBe(0n); + // Only the fresh incarnation may remain under the prefix. + const keys = await store.list(prefixOf(stream)); + expect(keys.filter((k) => k.endsWith(".bin"))).toEqual([]); + } finally { + await app.close(); + rmSync(root, { recursive: true, force: true }); + } + }, + 30_000 + ); + + test( + "recreate serializes with an in-flight background reap", + async () => { + const root = mkdtempSync(join(tmpdir(), "ds-reap-race-")); + const faults: MockR2Faults = {}; + const store = new MockR2Store({ faults }); + const app = createApp( + makeConfig(root, { expirySweepIntervalMs: 50, retentionDeleteConcurrency: 1 }), + store + ); + const stream = "raced"; + try { + await seedUploadedStream(app, store, stream, { records: 10 }); + faults.deleteDelayMs = 30; + + const delRes = await app.fetch(new Request(streamUrl(stream), { method: "DELETE" })); + expect(delRes.status).toBe(204); + // The nudged background reap is now crawling through the prefix. + const putRes = await app.fetch( + new Request(streamUrl(stream), { method: "PUT", headers: { "content-type": "application/json" } }) + ); + expect(putRes.status).toBe(201); + faults.deleteDelayMs = undefined; + + // Enough appends to seal and upload a segment, so the new incarnation + // publishes a manifest — which the background reap must not eat. + for (let i = 0; i < 6; i++) { + const r = await app.fetch( + new Request(streamUrl(stream), { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ generation: "second", i, pad: "x".repeat(48) }), + }) + ); + expect(r.status).toBe(204); + } + const healthy = await until(async () => { + const row = app.deps.db.getStream(stream); + return ( + !!row && + !app.deps.db.isDeleted(row) && + row.uploaded_through >= row.sealed_through && + app.deps.db.countSegmentsForStream(stream) > 0 && + store.has(manifestKeyOf(stream)) + ); + }); + expect(healthy).toBe(true); + const readRes = await app.fetch(new Request(`${streamUrl(stream)}?offset=-1`, { method: "GET" })); + expect(readRes.status).toBe(200); + expect(await readRes.text()).toContain("second"); + } finally { + await app.close(); + rmSync(root, { recursive: true, force: true }); + } + }, + 30_000 + ); + + test( + "a late-landing object is swept by the verify pass", + async () => { + const root = mkdtempSync(join(tmpdir(), "ds-reap-straggler-")); + const faults: MockR2Faults = {}; + const store = new MockR2Store({ faults }); + const app = createApp(makeConfig(root, { retentionDeleteConcurrency: 1 }), store); + const stream = "straggler"; + try { + await seedUploadedStream(app, store, stream, { records: 10 }); + app.deps.db.deleteStream(stream); + faults.deleteDelayMs = 40; + + const reapPromise = app.deps.reaper.reapStream(stream); + const midReap = await until(() => store.stats().deletes >= 1, 10_000, 5); + expect(midReap).toBe(true); + const strayKey = `${prefixOf(stream)}segments/9999999999999999-stray.bin`; + await store.put(strayKey, new TextEncoder().encode("stray")); + + const res = await reapPromise; + expect(Result.isError(res)).toBe(false); + if (!Result.isError(res)) { + expect(res.value.skipped).toBe(false); + expect(res.value.listPasses).toBeGreaterThanOrEqual(2); + } + expect(await store.list(prefixOf(stream))).toEqual([]); + expect(app.deps.db.getStream(stream)).toBeNull(); + } finally { + await app.close(); + rmSync(root, { recursive: true, force: true }); + } + }, + 30_000 + ); + + test( + "delete failures back off and converge once the store recovers", + async () => { + const root = mkdtempSync(join(tmpdir(), "ds-reap-backoff-")); + const faults: MockR2Faults = {}; + const store = new MockR2Store({ faults }); + const app = createApp(makeConfig(root, { expirySweepIntervalMs: 50 }), store); + const stream = "flaky"; + try { + await seedUploadedStream(app, store, stream); + faults.failDeleteEvery = 1; + + const delRes = await app.fetch(new Request(streamUrl(stream), { method: "DELETE" })); + expect(delRes.status).toBe(204); + + await sleep(1_200); + // Every reap attempt failed, so the row and prefix survive; the + // failure tracker keeps the sweeper from hammering the store on + // every 50ms tick. + expect(app.deps.db.getStream(stream)).not.toBeNull(); + expect((await store.list(prefixOf(stream))).length).toBeGreaterThan(0); + + faults.failDeleteEvery = undefined; + const reaped = await until( + async () => (await store.list(prefixOf(stream))).length === 0 && app.deps.db.getStream(stream) === null, + 20_000 + ); + expect(reaped).toBe(true); + } finally { + await app.close(); + rmSync(root, { recursive: true, force: true }); + } + }, + 30_000 + ); + + test( + "a deleted stream's sealed segments stop uploading", + async () => { + const root = mkdtempSync(join(tmpdir(), "ds-reap-upload-stop-")); + const store = new MockR2Store(); + // Uploads effectively off: segments seal locally and stay pending. + const app = createApp(makeConfig(root, { uploadIntervalMs: 60_000 }), store); + const stream = "never-uploads"; + try { + const createRes = await app.fetch( + new Request(streamUrl(stream), { method: "PUT", headers: { "content-type": "application/json" } }) + ); + expect([200, 201, 204]).toContain(createRes.status); + for (let i = 0; i < 6; i++) { + const r = await app.fetch( + new Request(streamUrl(stream), { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ i, pad: "x".repeat(48) }), + }) + ); + expect(r.status).toBe(204); + } + const sealed = await until(() => app.deps.db.countSegmentsForStream(stream) > 0); + expect(sealed).toBe(true); + expect(app.deps.db.pendingUploadHeads(10).map((r) => r.stream)).toContain(stream); + + app.deps.db.deleteStream(stream); + expect(app.deps.db.pendingUploadHeads(10).map((r) => r.stream)).not.toContain(stream); + + const res = await app.deps.reaper.reapStream(stream); + expect(Result.isError(res)).toBe(false); + expect(await store.list(prefixOf(stream))).toEqual([]); + expect(app.deps.db.getStream(stream)).toBeNull(); + expect(existsSync(`${root}/local/streams/${streamHash16Hex(stream)}`)).toBe(false); + } finally { + await app.close(); + rmSync(root, { recursive: true, force: true }); + } + }, + 30_000 + ); + + test( + "the retention scan reaps doomed manifests with no local row", + async () => { + const rootA = mkdtempSync(join(tmpdir(), "ds-reap-scan-src-")); + const rootB = mkdtempSync(join(tmpdir(), "ds-reap-scan-dst-")); + const store = new MockR2Store(); + const stream = "orphaned-by-redeploy"; + + const appA = createApp(makeConfig(rootA, {}), store); + try { + await seedUploadedStream(appA, store, stream, { ttl: "1" }); + } finally { + await appA.close(); + } + await sleep(1_100); + expect((await store.list(prefixOf(stream))).length).toBeGreaterThan(1); + + // Fresh node, empty SQLite, no eager bootstrap: only the scan can + // discover the doomed manifest. + const appB = createApp( + makeConfig(rootB, { expirySweepIntervalMs: 50, retentionScanIntervalMs: 150 }), + store + ); + try { + expect(appB.deps.db.getStream(stream)).toBeNull(); + const reaped = await until(async () => (await store.list(prefixOf(stream))).length === 0, 20_000); + expect(reaped).toBe(true); + expect(appB.deps.db.getStream(stream)).toBeNull(); + } finally { + await appB.close(); + rmSync(rootA, { recursive: true, force: true }); + rmSync(rootB, { recursive: true, force: true }); + } + }, + 30_000 + ); + + test( + "the internal metrics stream survives sweeping", + async () => { + const root = mkdtempSync(join(tmpdir(), "ds-reap-metrics-")); + const store = new MockR2Store(); + const app = createApp( + makeConfig(root, { expirySweepIntervalMs: 50, retentionScanIntervalMs: 150 }), + store + ); + try { + await sleep(500); + const row = app.deps.db.getStream(METRICS_STREAM); + expect(row).not.toBeNull(); + expect(app.deps.db.isDeleted(row!)).toBe(false); + } finally { + await app.close(); + rmSync(root, { recursive: true, force: true }); + } + }, + 30_000 + ); +});