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.

Author: kristof-siket

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`

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

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.
 
 ---

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`

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`)

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:

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`

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();