diff --git a/docs/overview.md b/docs/overview.md index 62c2349..62d8430 100644 --- a/docs/overview.md +++ b/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: diff --git a/docs/recovery-integrity-runbook.md b/docs/recovery-integrity-runbook.md index 24da006..a885332 100644 --- a/docs/recovery-integrity-runbook.md +++ b/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. diff --git a/package.json b/package.json index 2308d85..0615769 100644 --- a/package.json +++ b/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", diff --git a/src/app_core.ts b/src/app_core.ts index ce6da79..298cb7b 100644 --- a/src/app_core.ts +++ b/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>(); + /** + * 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 => { + 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"); diff --git a/src/bootstrap.ts b/src/bootstrap.ts index 377fffc..f205382 100644 --- a/src/bootstrap.ts +++ b/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; +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 { 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 { 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 { diff --git a/src/compute/demo_entry.ts b/src/compute/demo_entry.ts index d3a96bd..0796123 100644 --- a/src/compute/demo_entry.ts +++ b/src/compute/demo_entry.ts @@ -273,6 +273,10 @@ async function main(): Promise { } 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 { }); } - if (bootstrapEnabled) { + if (bootstrapEnabled && !lazyRestoreEnabled) { await bootstrapFromR2(cfg, store, { clearLocal: true }); } diff --git a/src/config.ts b/src/config.ts index 2364078..2a75ff3 100644 --- a/src/config.ts +++ b/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([ "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), }; } diff --git a/src/objectstore/mock_r2.ts b/src/objectstore/mock_r2.ts index e4bd065..45e341f 100644 --- a/src/objectstore/mock_r2.ts +++ b/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(); 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 { 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): 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(); } } diff --git a/src/server.ts b/src/server.ts index 1cb07ae..a4eba2b 100644 --- a/src/server.ts +++ b/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 }); } diff --git a/test/bootstrap_restore_scaling.test.ts b/test/bootstrap_restore_scaling.test.ts new file mode 100644 index 0000000..51df6ad --- /dev/null +++ b/test/bootstrap_restore_scaling.test.ts @@ -0,0 +1,220 @@ +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { describe, expect, test } from "bun:test"; +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 } from "../src/objectstore/mock_r2"; +import { streamHash16Hex } from "../src/util/stream_paths"; + +/** + * Reproduces the blocking eager-bootstrap wall that motivates lazy restore. + * + * On boot with `--bootstrap-from-r2` the server rebuilds the entire segment/stream + * index from R2 before it serves `/health`: LIST once, then per stream GET+HEAD the + * manifest, HEAD every segment, and GET the schema. That cost is + * `(#streams · per-stream-ops + #segments · HEAD) × object-store-round-trip-latency` + * and grows with the whole R2 backlog. With realistic R2 round-trips (~25 ms) it + * crosses the 60 s deploy health gate well before the backlog is large, which is + * what rolls the deploy back on Compute. + * + * The harness seeds a small authentic corpus through the real append→segment→upload + * path (so the manifest/segment layout is genuine), replicates those R2 objects + * under distinct stream names to reach N, injects a fixed per-op latency to model R2 + * round-trips, and times `bootstrapFromR2` at N = 100 / 1k / 5k. + * + * It is opt-in (like `test:large-index-filter`) because at production-like latency + * it deliberately runs for minutes. Enable with `DS_BOOTSTRAP_SCALING=1`; override + * the modeled per-op latency with `DS_BOOTSTRAP_SCALING_DELAY_MS` (default 25). The + * printed table always projects the measured op count to a 25 ms round-trip, which + * is the number cited as the production wall. + */ + +const SCALING_ENABLED = process.env.DS_BOOTSTRAP_SCALING === "1"; +const MODELED_R2_ROUND_TRIP_MS = 25; +const measuredDelayMs = process.env.DS_BOOTSTRAP_SCALING_DELAY_MS + ? Number(process.env.DS_BOOTSTRAP_SCALING_DELAY_MS) + : MODELED_R2_ROUND_TRIP_MS; + +function makeConfig(rootDir: string, overrides: Partial): Config { + const base = loadConfig(); + return { + ...base, + rootDir, + dbPath: `${rootDir}/wal.sqlite`, + port: 0, + ...overrides, + }; +} + +function sleep(ms: number): Promise { + return new Promise((res) => setTimeout(res, ms)); +} + +/** Seed one authentic base stream with `>= minSegments` uploaded segments in R2. */ +async function seedBaseStream(store: MockR2Store, baseStream: string, minSegments: number): Promise { + const root = mkdtempSync(join(tmpdir(), "ds-scale-base-")); + const cfg = makeConfig(root, { + segmentMaxBytes: 200, + segmentMaxIntervalMs: 30, + segmentCheckIntervalMs: 15, + uploadIntervalMs: 15, + uploadConcurrency: 8, + segmentCacheMaxBytes: 0, + segmentFooterCacheEntries: 0, + }); + const app = createApp(cfg, store); + try { + await app.fetch( + new Request(`http://local/v1/stream/${encodeURIComponent(baseStream)}`, { + method: "PUT", + headers: { "content-type": "application/json" }, + }) + ); + for (let i = 0; i < minSegments * 3; i++) { + const r = await app.fetch( + new Request(`http://local/v1/stream/${encodeURIComponent(baseStream)}`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ i, pad: "z".repeat(96) }), + }) + ); + expect(r.status).toBe(204); + } + const deadline = Date.now() + 20_000; + while (Date.now() < deadline) { + const row = app.deps.db.getStream(baseStream); + const segs = app.deps.db.countSegmentsForStream(baseStream); + if (row && row.uploaded_through >= row.sealed_through && segs >= minSegments) break; + await sleep(25); + } + expect(app.deps.db.countSegmentsForStream(baseStream)).toBeGreaterThanOrEqual(minSegments); + } finally { + await app.close(); + rmSync(root, { recursive: true, force: true }); + } +} + +/** + * Replicate the base stream's authentic R2 objects (manifest + segments) under a + * fresh stream name. The manifest `name` field is rewritten so bootstrap restores a + * genuine, distinct stream. Object bytes and key layout are otherwise identical to a + * stream that was actually appended, segmented, and uploaded, so the restore drives + * exactly the LIST/GET/HEAD pattern a real backlog would. + */ +async function cloneStream(store: MockR2Store, baseStream: string, targetStream: string): Promise { + const baseHash = streamHash16Hex(baseStream); + const targetHash = streamHash16Hex(targetStream); + const keys = await store.list(`streams/${baseHash}/`); + for (const key of keys) { + const bytes = await store.get(key); + if (!bytes) continue; + const targetKey = key.replace(`streams/${baseHash}/`, `streams/${targetHash}/`); + if (key.endsWith("/manifest.json")) { + const manifest = JSON.parse(new TextDecoder().decode(bytes)); + manifest.name = targetStream; + const rewritten = new TextEncoder().encode(JSON.stringify(manifest)); + await store.put(targetKey, rewritten, { contentType: "application/json", contentLength: rewritten.byteLength }); + } else { + await store.put(targetKey, bytes, { contentLength: bytes.byteLength }); + } + } +} + +type Row = { + streams: number; + segments: number; + objects: number; + ops: number; + measuredDelayMs: number; + measuredRestoreMs: number; + projected25msMs: number; +}; + +function formatCurve(rows: Row[]): string { + const header = "streams | segments | r2 objects | store ops | @" + measuredDelayMs + "ms restore | @25ms projected"; + const body = rows + .map( + (r) => + `${String(r.streams).padStart(7)} | ${String(r.segments).padStart(8)} | ${String(r.objects).padStart(10)} | ${String( + r.ops + ).padStart(9)} | ${(String(Math.round(r.measuredRestoreMs)) + "ms").padStart(String("@" + measuredDelayMs + "ms restore").length)} | ${( + (r.projected25msMs / 1000).toFixed(1) + "s" + ).padStart(16)}` + ) + .join("\n"); + return `\n${header}\n${body}\n`; +} + +describe.skipIf(!SCALING_ENABLED)("eager bootstrap restore scaling", () => { + test( + "eager bootstrapFromR2 wall-clock scales with backlog and crosses the 60s deploy gate", + async () => { + const store = new MockR2Store(); + const baseStream = "__scale_base__"; + await seedBaseStream(store, baseStream, 1); + + const baseHash = streamHash16Hex(baseStream); + const baseKeys = await store.list(`streams/${baseHash}/`); + const segmentsPerStream = baseKeys.filter((k) => k.includes("/segments/")).length; + expect(segmentsPerStream).toBeGreaterThanOrEqual(1); + + const sizes = [100, 1000, 5000]; + const rows: Row[] = []; + let built = 0; + + for (const target of sizes) { + for (let i = built; i < target; i++) { + await cloneStream(store, baseStream, `svc-${i}`); + } + built = target; + + const restoreRoot = mkdtempSync(join(tmpdir(), "ds-scale-restore-")); + const cfg = makeConfig(restoreRoot, { segmentCacheMaxBytes: 0, segmentFooterCacheEntries: 0 }); + store.setDelays({ getDelayMs: measuredDelayMs, headDelayMs: measuredDelayMs, listDelayMs: measuredDelayMs }); + store.resetStats(); + + const start = Date.now(); + await bootstrapFromR2(cfg, store, { clearLocal: true }); + const measuredRestoreMs = Date.now() - start; + + const opsStats = store.stats(); + const ops = opsStats.gets + opsStats.heads + opsStats.lists; + store.setDelays({ getDelayMs: 0, headDelayMs: 0, listDelayMs: 0 }); + + const verifyDb = new SqliteDurableStore(cfg.dbPath); + try { + expect(verifyDb.getStream("svc-0")).not.toBeNull(); + expect(verifyDb.getStream(`svc-${target - 1}`)).not.toBeNull(); + } finally { + verifyDb.close(); + } + rmSync(restoreRoot, { recursive: true, force: true }); + + rows.push({ + streams: target, + segments: target * segmentsPerStream, + objects: store.size(), + ops, + measuredDelayMs, + measuredRestoreMs, + projected25msMs: ops * MODELED_R2_ROUND_TRIP_MS, + }); + } + + // eslint-disable-next-line no-console + console.log(`\n[eager-bootstrap restore curve]${formatCurve(rows)}`); + + const at100 = rows.find((r) => r.streams === 100)!; + const at5000 = rows.find((r) => r.streams === 5000)!; + // Wall-clock is linear in the backlog: ~50x the streams => ~50x the ops/time. + expect(at5000.ops).toBeGreaterThan(at100.ops * 40); + expect(at5000.measuredRestoreMs).toBeGreaterThan(at100.measuredRestoreMs * 20); + // At a realistic 25ms R2 round-trip, the 5000-stream restore blows the 60s gate. + expect(at5000.projected25msMs).toBeGreaterThan(60_000); + }, + 900_000 + ); +}); diff --git a/test/lazy_r2_restore.test.ts b/test/lazy_r2_restore.test.ts new file mode 100644 index 0000000..badf11d --- /dev/null +++ b/test/lazy_r2_restore.test.ts @@ -0,0 +1,237 @@ +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { createApp } from "../src/app"; +import { bootstrapFromR2 } from "../src/bootstrap"; +import { loadConfig, type Config } from "../src/config"; +import { MockR2Store } from "../src/objectstore/mock_r2"; +import { manifestObjectKey, streamHash16Hex } from "../src/util/stream_paths"; + +function makeConfig(rootDir: string, overrides: Partial): Config { + const base = loadConfig(); + return { + ...base, + rootDir, + dbPath: `${rootDir}/wal.sqlite`, + port: 0, + ...overrides, + }; +} + +function sleep(ms: number): Promise { + return new Promise((res) => setTimeout(res, ms)); +} + +const seedOverrides: Partial = { + segmentMaxBytes: 128, + segmentMaxIntervalMs: 40, + segmentCheckIntervalMs: 20, + uploadIntervalMs: 20, + uploadConcurrency: 4, + segmentCacheMaxBytes: 0, + segmentFooterCacheEntries: 0, +}; + +const restoreOverrides: Partial = { + segmentCacheMaxBytes: 0, + segmentFooterCacheEntries: 0, +}; + +/** Seed streams into `store`, each with one or more uploaded segments in R2. */ +async function seedStreams(cfg: Config, store: MockR2Store, streams: string[], recordsPerStream: number): Promise { + const app = createApp(cfg, store); + try { + for (const stream of streams) { + const createRes = await app.fetch( + new Request(`http://local/v1/stream/${encodeURIComponent(stream)}`, { + method: "PUT", + headers: { "content-type": "application/json" }, + }) + ); + expect([201, 204]).toContain(createRes.status); + for (let i = 0; i < recordsPerStream; i++) { + const r = await app.fetch( + new Request(`http://local/v1/stream/${encodeURIComponent(stream)}`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ i, stream, pad: "x".repeat(48) }), + }) + ); + expect(r.status).toBe(204); + } + } + + const deadline = Date.now() + 20_000; + while (Date.now() < deadline) { + let allUploaded = true; + for (const stream of streams) { + const row = app.deps.db.getStream(stream); + const uploaded = row != null && row.uploaded_through >= row.sealed_through && app.deps.db.countSegmentsForStream(stream) > 0; + if (!uploaded) { + allUploaded = false; + break; + } + } + if (allUploaded) break; + await sleep(25); + } + for (const stream of streams) { + expect(app.deps.db.countSegmentsForStream(stream)).toBeGreaterThan(0); + } + } finally { + await app.close(); + } +} + +/** Read a whole stream from the beginning and return the response body text. */ +async function readWholeStream(app: ReturnType, stream: string): Promise<{ status: number; body: string }> { + const res = await app.fetch( + new Request(`http://local/v1/stream/${encodeURIComponent(stream)}?offset=-1&format=json`, { method: "GET" }) + ); + return { status: res.status, body: await res.text() }; +} + +/** How many of `streams` currently have a hydrated row in local SQLite. */ +function hydratedCount(app: ReturnType, streams: string[]): number { + return streams.filter((s) => app.deps.db.getStream(s) != null).length; +} + +describe("lazy R2 restore", () => { + let srcRoot: string; + let dstRoot: string; + + beforeEach(() => { + srcRoot = mkdtempSync(join(tmpdir(), "ds-lazy-src-")); + dstRoot = mkdtempSync(join(tmpdir(), "ds-lazy-dst-")); + }); + + afterEach(() => { + rmSync(srcRoot, { recursive: true, force: true }); + rmSync(dstRoot, { recursive: true, force: true }); + }); + + test( + "boots without eager restore and hydrates a cold stream on read with byte-identical records", + async () => { + const store = new MockR2Store(); + const streams = ["alpha", "beta", "gamma"]; + await seedStreams(makeConfig(srcRoot, seedOverrides), store, streams, 6); + + // Reference: eager-restored node reads every stream. + const eagerRoot = mkdtempSync(join(tmpdir(), "ds-lazy-eager-")); + await bootstrapFromR2(makeConfig(eagerRoot, restoreOverrides), store, { clearLocal: true }); + const eagerApp = createApp(makeConfig(eagerRoot, restoreOverrides), store); + const eagerBodies = new Map(); + try { + for (const stream of streams) { + const { status, body } = await readWholeStream(eagerApp, stream); + expect(status).toBe(200); + eagerBodies.set(stream, body); + } + } finally { + await eagerApp.close(); + rmSync(eagerRoot, { recursive: true, force: true }); + } + + // Lazy node: eager restore skipped, SQLite starts empty. + const lazyApp = createApp(makeConfig(dstRoot, { lazyRestore: true, ...restoreOverrides }), store); + try { + expect(hydratedCount(lazyApp, streams)).toBe(0); + + const cold = await readWholeStream(lazyApp, "beta"); + expect(cold.status).toBe(200); + expect(cold.body).toBe(eagerBodies.get("beta")); + + // Only the one read stream was hydrated into SQLite. + expect(hydratedCount(lazyApp, streams)).toBe(1); + expect(lazyApp.deps.db.getStream("beta")).not.toBeNull(); + expect(lazyApp.deps.db.getStream("alpha")).toBeNull(); + } finally { + await lazyApp.close(); + } + }, + 60_000 + ); + + test( + "a never-written stream returns 404 in lazy mode", + async () => { + const store = new MockR2Store(); + await seedStreams(makeConfig(srcRoot, seedOverrides), store, ["present"], 4); + + const lazyApp = createApp(makeConfig(dstRoot, { lazyRestore: true, ...restoreOverrides }), store); + try { + const missing = await readWholeStream(lazyApp, "never-written"); + expect(missing.status).toBe(404); + expect(lazyApp.deps.db.getStream("never-written")).toBeNull(); + + const present = await readWholeStream(lazyApp, "present"); + expect(present.status).toBe(200); + } finally { + await lazyApp.close(); + } + }, + 60_000 + ); + + test( + "reading K of N cold streams hydrates ~K SQLite rows, not N", + async () => { + const store = new MockR2Store(); + const streams = Array.from({ length: 8 }, (_, i) => `stream-${i}`); + await seedStreams(makeConfig(srcRoot, seedOverrides), store, streams, 4); + + const lazyApp = createApp(makeConfig(dstRoot, { lazyRestore: true, ...restoreOverrides }), store); + try { + expect(hydratedCount(lazyApp, streams)).toBe(0); + + const readStreams = streams.slice(0, 3); + for (const stream of readStreams) { + const { status } = await readWholeStream(lazyApp, stream); + expect(status).toBe(200); + } + + expect(hydratedCount(lazyApp, streams)).toBe(readStreams.length); + for (let i = readStreams.length; i < streams.length; i++) { + expect(lazyApp.deps.db.getStream(streams[i])).toBeNull(); + } + } finally { + await lazyApp.close(); + } + }, + 60_000 + ); + + test( + "concurrent reads of the same cold stream trigger a single hydration", + async () => { + const store = new MockR2Store(); + await seedStreams(makeConfig(srcRoot, seedOverrides), store, ["hot"], 6); + + // Slow the object store so all concurrent readers arrive before the first + // hydration settles, exercising the single-flight collapse. + store.setDelays({ getDelayMs: 40, headDelayMs: 40, listDelayMs: 40 }); + + const lazyApp = createApp(makeConfig(dstRoot, { lazyRestore: true, ...restoreOverrides }), store); + try { + store.resetStats(); + const manifestKey = manifestObjectKey(streamHash16Hex("hot")); + expect(store.getCountFor(manifestKey)).toBe(0); + + const results = await Promise.all( + Array.from({ length: 12 }, () => readWholeStream(lazyApp, "hot")) + ); + for (const { status } of results) expect(status).toBe(200); + + // Single-flight: the manifest fetch that drives hydration fires exactly + // once for the whole burst, not once per concurrent reader. + expect(store.getCountFor(manifestKey)).toBe(1); + expect(lazyApp.deps.db.getStream("hot")).not.toBeNull(); + } finally { + await lazyApp.close(); + } + }, + 60_000 + ); +});