From e17d8e29b8b5e5b462e0301dd5b6326ae09d7413 Mon Sep 17 00:00:00 2001 From: Kristof Siket Date: Thu, 2 Jul 2026 13:30:53 +0200 Subject: [PATCH 1/2] Reap expired and deleted streams from the object store A stream row with STREAM_FLAG_DELETED is the single reapable state: the expiry sweeper flags expired rows, HTTP DELETE flags explicitly, and the StreamReaper deletes the stream's objects data-first with manifest.json strictly last, hard-deleting local rows only once the prefix is verifiably empty. The manifest is kept terminal (deleted flag or past expiry) in the object store before any data object disappears, so restore-from-R2 mid-reap restores a row-only tombstone that re-arms the reap instead of aborting boot on a missing segment head. A periodic scan restores row-only tombstones for doomed manifests that no longer have a local row (local SQLite is ephemeral on redeploy), making the object store the durable retention driver. PUT-recreate now reaps the old incarnation before creating, closing the orphaned-objects and stale segment-cache holes; the uploader stops picking segments of deleted streams so a reap cannot be re-populated mid-flight. --- src/app_core.ts | 47 +++++-- src/bootstrap.ts | 59 ++++++++- src/config.ts | 9 ++ src/db/db.ts | 24 +++- src/expiry_sweeper.ts | 44 ------- src/objectstore/mock_r2.ts | 11 +- src/retention.ts | 263 +++++++++++++++++++++++++++++++++++++ src/retention_sweeper.ts | 165 +++++++++++++++++++++++ src/uploader.ts | 2 +- 9 files changed, 564 insertions(+), 60 deletions(-) delete mode 100644 src/expiry_sweeper.ts create mode 100644 src/retention.ts create mode 100644 src/retention_sweeper.ts 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) { From 0f7e31ab38093eacbbc9a7aa7229db28a180849e Mon Sep 17 00:00:00 2001 From: Kristof Siket Date: Thu, 2 Jul 2026 13:52:31 +0200 Subject: [PATCH 2/2] Test and document the retention reaper Thirteen lifecycle tests cover expiry and DELETE reaps, manifest-last ordering, crash-resume across restore-from-R2, tombstone-lite bootstrap (zero segment head checks), recreate on deleted and expired names, recreate racing an in-flight reap, verify-pass stragglers, failure backoff, the upload filter for deleted streams, the no-local-row scan path, and internal-stream safety. MockR2 gains delete fault injection and counters. The assumptions suite's delete test now asserts the new contract: tombstone until the reap completes, then fully gone. Docs: spec DELETE/TTL semantics, recovery-runbook deletion commit points, architecture retention section, operational env knobs, metrics series, sqlite-schema lifecycle note, CHANGELOG. --- CHANGELOG.md | 9 + docs/architecture.md | 28 +- docs/durable-streams-spec.md | 10 +- docs/metrics.md | 11 + docs/operational-notes.md | 3 + docs/recovery-integrity-runbook.md | 18 + docs/sqlite-schema.md | 4 + test/assumptions.test.ts | 12 +- test/retention_reaper.test.ts | 581 +++++++++++++++++++++++++++++ 9 files changed, 669 insertions(+), 7 deletions(-) create mode 100644 test/retention_reaper.test.ts 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/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 + ); +});