Reap expired and deleted streams from the object store

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`

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,

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

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<string>([
   "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),