feat: lazy on-demand R2 index restore (skip eager bootstrap)

docs/overview.md

@@ -50,6 +50,20 @@ object storage. It does not restore transient local SQLite state
 such as the unuploaded WAL tail, producer dedupe state, or runtime live/template
 state.
 
+`--lazy-restore` is an alternative to eager bootstrap for large backlogs. Instead
+of rebuilding every stream's index into local SQLite before the server accepts
+traffic, the server starts immediately and hydrates a stream's index from its R2
+manifest on the first read that misses local SQLite. A read for a stream with no
+R2 manifest is a genuine `404`. The rows a lazy read writes are identical to what
+`--bootstrap-from-r2` would have written for that stream, so the reader's
+segment/WAL merge stays correct either way. This keeps `/health` responsive
+regardless of backlog size, which matters when boot time is gated by a deploy
+health check. When both flags are passed, `--lazy-restore` wins and the eager
+pass is skipped. Concurrent first reads of the same cold stream share one
+hydration. Hydrated rows are not evicted; on a host where local SQLite resets per
+deploy the working set stays bounded, but a long-lived node that reads a very
+large stream set will accumulate index rows (bounded eviction is future work).
+
 A stream becomes recoverable from object storage after its first manifest is
 published.
 
@@ -189,6 +203,8 @@ Optional flags:
 - `--stats`
 - `--hist`
 - `--bootstrap-from-r2`
+- `--lazy-restore` (skip the eager bootstrap; hydrate each stream's index from R2
+  on the first read miss instead; also settable via `DS_LAZY_RESTORE=1`)
 - `--auto-tune[=MB]`
 
 Optional OTLP trace receiver configuration:

docs/recovery-integrity-runbook.md

@@ -123,6 +123,13 @@ Full mode:
 - `--bootstrap-from-r2` does not restore transient local SQLite state such as
   the unuploaded WAL tail, producer dedupe state, or runtime live/template
   state.
+- `--lazy-restore` is an alternative to eager `--bootstrap-from-r2`: the server
+  serves `/health` immediately and hydrates each stream's index from its R2
+  manifest on the first read miss, producing the same SQLite rows the eager pass
+  would. Prefer it when eager restore of a large backlog would blow a startup
+  health check. It restores the same published state and, like eager bootstrap,
+  does not restore the unuploaded WAL tail or other transient local state. A read
+  for a stream with no published manifest is a genuine not-found.
 - Deleting a node without a local snapshot is only safe after the streams you
   care about have uploaded through the tail you need and published a manifest
   for that state.

package.json

@@ -48,6 +48,7 @@
     "test": "bun test",
     "test:ci": "node scripts/run-bun-tests-sequential.mjs",
     "test:large-index-filter": "DS_LARGE_INDEX_FILTER=1 DS_LARGE_INDEX_FILTER_SEGMENT_BYTES=16777216 bun test test/large_index_filter.test.ts",
+    "test:bootstrap-scaling": "DS_BOOTSTRAP_SCALING=1 bun test test/bootstrap_restore_scaling.test.ts",
     "typecheck": "tsc --noEmit",
     "test:node-local-package": "node scripts/test-node-local-package.mjs",
     "test:bun-local-package": "node scripts/test-bun-local-package.mjs",

src/app_core.ts

@@ -11,6 +11,7 @@ import { parseDurationMsResult } from "./util/duration";
 import { Metrics } from "./metrics";
 import { parseTimestampMsResult } from "./util/time";
 import { cleanupTempSegments } from "./util/cleanup";
+import { hydrateStreamFromR2 } from "./bootstrap";
 import { MetricsEmitter } from "./metrics_emitter";
 import {
   SchemaRegistryStore,
@@ -629,6 +630,32 @@ export function createAppCore(cfg: Config, opts: CreateAppCoreOptions): App {
     memorySampler,
   });
   const { store, reader, segmenter, uploader, indexer, uploadSchemaRegistry, getRuntimeMemorySnapshot, getLocalStorageUsage } = runtime;
+
+  const hydrateInFlight = new Map<string, Promise<"hydrated" | "absent">>();
+  /**
+   * Lazy-restore lookup: resolves a stream row for the read path. When the eager
+   * R2 bootstrap was skipped (`--lazy-restore`) and the stream's index rows are
+   * not yet in local SQLite, this hydrates them from that stream's R2 manifest on
+   * demand, then re-reads the row. Concurrent readers of the same cold stream
+   * share one hydration (single-flight, evicted on settle). A stream with no
+   * manifest in R2 resolves to `null`, i.e. a genuine not-found.
+   *
+   * When lazy restore is off this is a single synchronous `getStream`, so the
+   * eager and local modes keep their existing hot-path behavior.
+   */
+  const getStreamForRead = async (stream: string): Promise<StreamRow | null> => {
+    const existing = db.getStream(stream);
+    if (existing || !cfg.lazyRestore) return existing;
+    let pending = hydrateInFlight.get(stream);
+    if (!pending) {
+      pending = hydrateStreamFromR2(cfg, store, db, stream).finally(() => {
+        hydrateInFlight.delete(stream);
+      });
+      hydrateInFlight.set(stream, pending);
+    }
+    await pending;
+    return db.getStream(stream);
+  };
   const runtimeHighWater: RuntimeMemoryHighWaterSnapshot = {
     process: {},
     process_breakdown: {},
@@ -1830,6 +1857,8 @@ export function createAppCore(cfg: Config, opts: CreateAppCoreOptions): App {
           return capability;
         };
 
+        if (observeReq.include.events && observeReq.streams.events) await getStreamForRead(observeReq.streams.events);
+        if (observeReq.include.trace && observeReq.streams.traces) await getStreamForRead(observeReq.streams.traces);
         const eventCorrelation =
           observeReq.include.events && observeReq.streams.events ? loadCorrelationCapability(observeReq.streams.events, "events") : null;
         if (eventCorrelation instanceof Response) return eventCorrelation;
@@ -2191,7 +2220,7 @@ export function createAppCore(cfg: Config, opts: CreateAppCoreOptions): App {
         }
 
         if (isSchema) {
-          const srow = db.getStream(stream);
+          const srow = await getStreamForRead(stream);
           if (!srow || db.isDeleted(srow)) return notFound();
           if (srow.expires_at_ms != null && db.nowMs() > srow.expires_at_ms) return notFound("stream expired");
 
@@ -2254,7 +2283,7 @@ export function createAppCore(cfg: Config, opts: CreateAppCoreOptions): App {
         }
 
         if (isProfile) {
-          const srow = db.getStream(stream);
+          const srow = await getStreamForRead(stream);
           if (!srow || db.isDeleted(srow)) return notFound();
           if (srow.expires_at_ms != null && db.nowMs() > srow.expires_at_ms) return notFound("stream expired");
 
@@ -2293,6 +2322,7 @@ export function createAppCore(cfg: Config, opts: CreateAppCoreOptions): App {
 
         if (isDetails || isIndexStatus) {
           if (req.method !== "GET") return badRequest("unsupported method");
+          await getStreamForRead(stream);
           const liveParam = url.searchParams.get("live") ?? "";
           let longPoll = false;
           if (liveParam === "" || liveParam === "false" || liveParam === "0") longPoll = false;
@@ -2382,7 +2412,7 @@ export function createAppCore(cfg: Config, opts: CreateAppCoreOptions): App {
         }
 
         if (isRoutingKeys) {
-          const srow = db.getStream(stream);
+          const srow = await getStreamForRead(stream);
           if (!srow || db.isDeleted(srow)) return notFound();
           if (srow.expires_at_ms != null && db.nowMs() > srow.expires_at_ms) return notFound("stream expired");
           if (req.method !== "GET") return badRequest("unsupported method");
@@ -2429,7 +2459,7 @@ export function createAppCore(cfg: Config, opts: CreateAppCoreOptions): App {
         }
 
         if (isSearch) {
-          const srow = db.getStream(stream);
+          const srow = await getStreamForRead(stream);
           if (!srow || db.isDeleted(srow)) return notFound();
           if (srow.expires_at_ms != null && db.nowMs() > srow.expires_at_ms) return notFound("stream expired");
 
@@ -2507,7 +2537,7 @@ export function createAppCore(cfg: Config, opts: CreateAppCoreOptions): App {
         }
 
         if (isAggregate) {
-          const srow = db.getStream(stream);
+          const srow = await getStreamForRead(stream);
           if (!srow || db.isDeleted(srow)) return notFound();
           if (srow.expires_at_ms != null && db.nowMs() > srow.expires_at_ms) return notFound("stream expired");
           if (req.method !== "POST") return badRequest("unsupported method");
@@ -2725,7 +2755,7 @@ export function createAppCore(cfg: Config, opts: CreateAppCoreOptions): App {
         }
 
         if (req.method === "HEAD") {
-          const srow = db.getStream(stream);
+          const srow = await getStreamForRead(stream);
           if (!srow || db.isDeleted(srow)) return notFound();
           if (srow.expires_at_ms != null && db.nowMs() > srow.expires_at_ms) return notFound("stream expired");
           const tailOffset = encodeOffset(srow.epoch, srow.next_offset - 1n);
@@ -2891,7 +2921,7 @@ export function createAppCore(cfg: Config, opts: CreateAppCoreOptions): App {
         }
 
         if (req.method === "GET") {
-          const srow = db.getStream(stream);
+          const srow = await getStreamForRead(stream);
           if (!srow || db.isDeleted(srow)) return notFound();
           if (srow.expires_at_ms != null && db.nowMs() > srow.expires_at_ms) return notFound("stream expired");
 

src/bootstrap.ts

@@ -4,12 +4,21 @@ import type { Config } from "./config";
 import { SqliteDurableStore } from "./db/db";
 import type { ObjectStore } from "./objectstore/interface";
 import { zstdDecompressSync } from "./util/zstd";
-import { localSegmentPath, schemaObjectKey, segmentObjectKey, streamHash16Hex } from "./util/stream_paths";
-import { retry } from "./util/retry";
+import { localSegmentPath, manifestObjectKey, schemaObjectKey, segmentObjectKey, streamHash16Hex } from "./util/stream_paths";
+import { retry, type RetryOptions } from "./util/retry";
 import { dsError } from "./util/ds_error.ts";
 
 type Manifest = Record<string, any>;
 
+function objectStoreRetryOpts(cfg: Config): RetryOptions {
+  return {
+    retries: cfg.objectStoreRetries,
+    baseDelayMs: cfg.objectStoreBaseDelayMs,
+    maxDelayMs: cfg.objectStoreMaxDelayMs,
+    timeoutMs: cfg.objectStoreTimeoutMs,
+  };
+}
+
 export async function bootstrapFromR2(cfg: Config, store: ObjectStore, opts: { clearLocal?: boolean } = {}): Promise<void> {
   if (opts.clearLocal !== false) {
     try {
@@ -33,12 +42,7 @@ export async function bootstrapFromR2(cfg: Config, store: ObjectStore, opts: { c
 
   const db = new SqliteDurableStore(cfg.dbPath, { cacheBytes: cfg.sqliteCacheBytes });
   try {
-    const retryOpts = {
-      retries: cfg.objectStoreRetries,
-      baseDelayMs: cfg.objectStoreBaseDelayMs,
-      maxDelayMs: cfg.objectStoreMaxDelayMs,
-      timeoutMs: cfg.objectStoreTimeoutMs,
-    };
+    const retryOpts = objectStoreRetryOpts(cfg);
     const keys = await retry(() => store.list("streams/"), retryOpts);
     const manifestKeys = keys.filter((k) => k.endsWith("/manifest.json"));
     for (const mkey of manifestKeys) {
@@ -48,8 +52,51 @@ export async function bootstrapFromR2(cfg: Config, store: ObjectStore, opts: { c
         return data;
       }, retryOpts);
       const manifest = JSON.parse(new TextDecoder().decode(mbytes)) as Manifest;
+      await restoreManifestIntoDb(cfg, store, db, manifest, mkey, retryOpts);
+    }
+  } finally {
+    db.close();
+  }
+}
+
+/**
+ * Hydrate one stream's SQLite index rows from its published R2 manifest, into an
+ * already-open store. Returns `"hydrated"` when the manifest exists and was
+ * applied, or `"absent"` when no manifest exists for that stream (a genuine
+ * not-found). Throws only on object-store/decode failures.
+ *
+ * The rows produced are identical to what {@link bootstrapFromR2} writes for the
+ * same stream, so the reader's segment/WAL merge stays correct whether the index
+ * was restored eagerly on boot or lazily on the first read miss.
+ */
+export async function hydrateStreamFromR2(
+  cfg: Config,
+  store: ObjectStore,
+  db: SqliteDurableStore,
+  streamName: string
+): Promise<"hydrated" | "absent"> {
+  const shash = streamHash16Hex(streamName);
+  const mkey = manifestObjectKey(shash);
+  const retryOpts = objectStoreRetryOpts(cfg);
+  const mbytes = await retry(() => store.get(mkey), retryOpts);
+  if (!mbytes) return "absent";
+  const manifest = JSON.parse(new TextDecoder().decode(mbytes)) as Manifest;
+  const stream = String(manifest.name ?? "");
+  if (!stream) return "absent";
+  await restoreManifestIntoDb(cfg, store, db, manifest, mkey, retryOpts);
+  return "hydrated";
+}
+
+async function restoreManifestIntoDb(
+  cfg: Config,
+  store: ObjectStore,
+  db: SqliteDurableStore,
+  manifest: Manifest,
+  mkey: string,
+  retryOpts: RetryOptions
+): Promise<void> {
       const stream = String(manifest.name ?? "");
-      if (!stream) continue;
+      if (!stream) return;
 
       const shash = streamHash16Hex(stream);
       const nowMs = db.nowMs();
@@ -349,10 +396,6 @@ export async function bootstrapFromR2(cfg: Config, store: ObjectStore, opts: { c
         db.upsertSchemaRegistry(stream, new TextDecoder().decode(schemaBytes));
         db.setSchemaUploadedSizeBytes(stream, schemaBytes.byteLength);
       }
-    }
-  } finally {
-    db.close();
-  }
 }
 
 function parseManifestBigInt(value: unknown): bigint | null {

src/compute/demo_entry.ts

@@ -273,6 +273,10 @@ async function main(): Promise<void> {
       }
       throw error;
     }
+    const lazyRestoreEnabled = process.argv.slice(2).includes("--lazy-restore");
+    if (lazyRestoreEnabled) {
+      process.env.DS_LAZY_RESTORE = "1";
+    }
     cfg = loadConfig();
     const args = process.argv.slice(2);
 
@@ -354,7 +358,7 @@ async function main(): Promise<void> {
       });
     }
 
-    if (bootstrapEnabled) {
+    if (bootstrapEnabled && !lazyRestoreEnabled) {
       await bootstrapFromR2(cfg, store, { clearLocal: true });
     }
 

src/config.ts

@@ -70,6 +70,7 @@ export type Config = {
   touchMaxBatchBytes: number;
   otlpTracesStream: string | null;
   otlpAutoCreate: boolean;
+  lazyRestore: boolean;
   port: number;
 };
 
@@ -147,6 +148,7 @@ const KNOWN_DS_ENVS = new Set<string>([
   "DS_TOUCH_MAX_BATCH_BYTES",
   "DS_OTLP_TRACES_STREAM",
   "DS_OTLP_AUTO_CREATE",
+  "DS_LAZY_RESTORE",
   "DS_AUTO_TUNE_REQUESTED_MB",
   "DS_AUTO_TUNE_PRESET_MB",
   "DS_AUTO_TUNE_EFFECTIVE_MEMORY_LIMIT_MB",
@@ -369,6 +371,7 @@ export function loadConfig(): Config {
     touchMaxBatchBytes: envNum("DS_TOUCH_MAX_BATCH_BYTES", 4 * 1024 * 1024),
     otlpTracesStream: process.env.DS_OTLP_TRACES_STREAM?.trim() || null,
     otlpAutoCreate: envBool("DS_OTLP_AUTO_CREATE", false),
+    lazyRestore: envBool("DS_LAZY_RESTORE", false),
     port: envNum("PORT", 8080),
   };
 }

src/objectstore/mock_r2.ts

@@ -50,6 +50,7 @@ export class MockR2Store implements ObjectStore {
   private headCount = 0;
   private listCount = 0;
   private memBytes = 0;
+  private readonly getCountByKey = new Map<string, number>();
 
   constructor(opts: MockR2Options | MockR2Faults = {}) {
     if ("failPutEvery" in opts || "putDelayMs" in opts) {
@@ -160,6 +161,7 @@ export class MockR2Store implements ObjectStore {
 
   async get(key: string, opts: GetOptions = {}): Promise<Uint8Array | null> {
     this.getCount++;
+    this.getCountByKey.set(key, (this.getCountByKey.get(key) ?? 0) + 1);
     this.maybeFail(this.getCount, this.faults.failGetEvery, `MockR2: injected GET failure for ${key}`);
     this.maybeTimeout(this.getCount, this.faults.timeoutGetEvery, `MockR2: injected GET timeout for ${key}`);
     await sleep(this.faults.getDelayMs ?? 0);
@@ -238,10 +240,21 @@ export class MockR2Store implements ObjectStore {
   }
 
   // Helpers for tests
+  setDelays(delays: Pick<MockR2Faults, "getDelayMs" | "headDelayMs" | "listDelayMs" | "putDelayMs">): void {
+    if (delays.getDelayMs !== undefined) this.faults.getDelayMs = delays.getDelayMs;
+    if (delays.headDelayMs !== undefined) this.faults.headDelayMs = delays.headDelayMs;
+    if (delays.listDelayMs !== undefined) this.faults.listDelayMs = delays.listDelayMs;
+    if (delays.putDelayMs !== undefined) this.faults.putDelayMs = delays.putDelayMs;
+  }
+
   has(key: string): boolean {
     return this.data.has(key);
   }
 
+  getCountFor(key: string): number {
+    return this.getCountByKey.get(key) ?? 0;
+  }
+
   size(): number {
     return this.data.size;
   }
@@ -265,5 +278,6 @@ export class MockR2Store implements ObjectStore {
     this.getCount = 0;
     this.headCount = 0;
     this.listCount = 0;
+    this.getCountByKey.clear();
   }
 }

src/server.ts

@@ -33,6 +33,11 @@ if (autoTune.enabled) {
   }
 }
 
+const lazyRestoreEnabled = args.includes("--lazy-restore");
+if (lazyRestoreEnabled) {
+  process.env.DS_LAZY_RESTORE = "1";
+}
+
 const cfg = loadConfig();
 
 const statsEnabled = args.includes("--stats");
@@ -124,7 +129,7 @@ if (storeChoice === "local") {
   });
 }
 
-if (bootstrapEnabled) {
+if (bootstrapEnabled && !lazyRestoreEnabled) {
   await bootstrapFromR2(cfg, store, { clearLocal: true });
 }