From d75b5fc06dd740728b66a3e008e5aa4c1a8b69f2 Mon Sep 17 00:00:00 2001 From: MasterPtato Date: Thu, 9 Jul 2026 10:59:15 -0700 Subject: [PATCH] feat(kitchen-sink): add churnDb actor and seed driver for compaction read-hotshard repro --- .../kitchen-sink/scripts/seed-churn-dbs.ts | 279 ++++++++++++++++++ .../src/actors/testing/churn-db.ts | 268 +++++++++++++++++ examples/kitchen-sink/src/index.ts | 2 + 3 files changed, 549 insertions(+) create mode 100644 examples/kitchen-sink/scripts/seed-churn-dbs.ts create mode 100644 examples/kitchen-sink/src/actors/testing/churn-db.ts diff --git a/examples/kitchen-sink/scripts/seed-churn-dbs.ts b/examples/kitchen-sink/scripts/seed-churn-dbs.ts new file mode 100644 index 0000000000..d32975f411 --- /dev/null +++ b/examples/kitchen-sink/scripts/seed-churn-dbs.ts @@ -0,0 +1,279 @@ +#!/usr/bin/env -S pnpm exec tsx + +// Seeds churnDb actors with deep overwrite delta chains for the compaction +// read-hot-shard load test (spec H2). Each actor rewrites a small bounded +// working set in place until it accumulates a target committed transaction +// (delta) count. +// +// Read-hotspot shape (see churn-db.ts): keep each branch's total DELTA under +// one FDB shard (< ~500 MB) so the compactor stage read cannot fan out across +// shards, while the chain stays deep (25k+ txids). Seed MANY such branches +// (CHURN_COUNT in the hundreds), not a few whales, so many single-shard stage +// reads run concurrently once compaction is enabled. Confirm the shape with +// fdb_subspace_sizes.py: each seeded branch should show meaningful DELTA on a +// single-digit shard count. +// +// The engine under test has hot compaction disabled while seeding, so every +// commit accumulates as an uncompacted delta chain in FDB. This driver only +// issues action calls; the actual SQLite writes happen on the serverful +// kitchen-sink runners the engine routes to (pool "k8s"). +// +// Seed only a few of these (COUNT small) alongside the H1 append-only herd +// from seed-large-dbs.ts. After deploying the compaction-enabled build, re-run +// with a slightly higher CHURN_TARGET_TXIDS to "bump" each whale (force one +// commit) and spawn the depot_db_manager -> hot compactor workflows. +// +// Usage (in-cluster Job or laptop): +// RIVET_ENDPOINT="http://default:@rivet-engine.rivet-engine.svc.cluster.local:6420" \ +// CHURN_COUNT=3 CHURN_TARGET_TXIDS=18000 CHURN_CONCURRENCY=3 \ +// node --import tsx scripts/seed-churn-dbs.ts +// +// All knobs are env vars (flags also accepted): +// RIVET_ENDPOINT engine guard endpoint incl. namespace:token userinfo +// CHURN_COUNT number of distinct whale DBs to create (default 3) +// CHURN_TARGET_TXIDS target committed-txn (delta) count per DB (default 18000) +// CHURN_WORKING_SET_ROWS rows in the bounded working set (default 64) +// CHURN_ROW_BYTES payload bytes per row (default 8192) +// CHURN_CONCURRENCY max DBs churned in parallel (default = count) +// CHURN_KEY_PREFIX actor key prefix (default "churn") +// CHURN_RUN_ID run tag mixed into keys (default timestamp) +// CHURN_MAX_CALLS max churn() calls per DB before giving up (default 1000) + +import { createClient } from "rivetkit/client"; +import type { registry } from "../src/index.ts"; + +interface Args { + endpoint: string; + count: number; + targetTxids: number; + workingSetRows: number; + rowBytes: number; + concurrency: number; + keyPrefix: string; + runId: string; + maxCalls: number; + commitDelayMs: number; + budgetMs?: number; +} + +function envNumNonNeg(name: string, fallback: number): number { + const raw = process.env[name]; + if (raw === undefined || raw === "") return fallback; + const value = Number(raw); + if (!Number.isFinite(value) || value < 0) { + throw new Error(`${name} must be a non-negative number, got ${raw}`); + } + return Math.floor(value); +} + +function envNum(name: string, fallback: number): number { + const raw = process.env[name]; + if (raw === undefined || raw === "") return fallback; + const value = Number(raw); + if (!Number.isFinite(value) || value <= 0) { + throw new Error(`${name} must be a positive number, got ${raw}`); + } + return Math.floor(value); +} + +function flag(argv: string[], name: string): string | undefined { + const prefix = `${name}=`; + const inline = argv.find((a) => a.startsWith(prefix)); + if (inline) return inline.slice(prefix.length); + const idx = argv.indexOf(name); + if (idx >= 0) return argv[idx + 1]; + return undefined; +} + +function parseArgs(argv: string[]): Args { + const endpoint = flag(argv, "--endpoint") ?? process.env.RIVET_ENDPOINT; + if (!endpoint) { + throw new Error( + "missing endpoint: set RIVET_ENDPOINT or pass --endpoint", + ); + } + const runId = + flag(argv, "--run-id") ?? + process.env.CHURN_RUN_ID ?? + new Date().toISOString().replace(/[:.]/g, "-"); + const count = Number(flag(argv, "--count")) || envNum("CHURN_COUNT", 3); + return { + endpoint, + count, + targetTxids: + Number(flag(argv, "--target-txids")) || + envNum("CHURN_TARGET_TXIDS", 25_000), + workingSetRows: + Number(flag(argv, "--working-set-rows")) || + envNum("CHURN_WORKING_SET_ROWS", 16), + rowBytes: + Number(flag(argv, "--row-bytes")) || + envNum("CHURN_ROW_BYTES", 256), + concurrency: + Number(flag(argv, "--concurrency")) || + envNum("CHURN_CONCURRENCY", count), + keyPrefix: + flag(argv, "--key-prefix") ?? + process.env.CHURN_KEY_PREFIX ?? + "churn", + runId, + maxCalls: + Number(flag(argv, "--max-calls")) || + envNum("CHURN_MAX_CALLS", 1000), + commitDelayMs: envNumNonNeg("CHURN_COMMIT_DELAY_MS", 25), + budgetMs: process.env.CHURN_BUDGET_MS + ? envNum("CHURN_BUDGET_MS", 120_000) + : undefined, + }; +} + +function fmtBytes(bytes: number): string { + return `${(bytes / 1024 / 1024).toFixed(1)}MiB`; +} + +interface DbOutcome { + key: string; + ok: boolean; + sizeBytes: number; + pageCount: number; + txidCount: number; + calls: number; + error?: string; +} + +async function churnOne( + client: ReturnType>, + args: Args, + index: number, +): Promise { + const key = `${args.keyPrefix}-${args.runId}-${String(index).padStart(6, "0")}`; + const handle = client.churnDb.getOrCreate(key); + let calls = 0; + let last = { done: false, sizeBytes: 0, pageCount: 0, txidCount: 0 }; + try { + while (!last.done && calls < args.maxCalls) { + const result = await handle.churn({ + targetTxids: args.targetTxids, + workingSetRows: args.workingSetRows, + rowBytes: args.rowBytes, + commitDelayMs: args.commitDelayMs, + ...(args.budgetMs !== undefined + ? { budgetMs: args.budgetMs } + : {}), + }); + calls += 1; + last = { + done: result.done, + sizeBytes: result.sizeBytes, + pageCount: result.pageCount, + txidCount: result.txidCount, + }; + } + return { + key, + ok: last.done, + sizeBytes: last.sizeBytes, + pageCount: last.pageCount, + txidCount: last.txidCount, + calls, + ...(last.done ? {} : { error: `not done after ${calls} calls` }), + }; + } catch (err) { + return { + key, + ok: false, + sizeBytes: last.sizeBytes, + pageCount: last.pageCount, + txidCount: last.txidCount, + calls, + error: err instanceof Error ? err.message : String(err), + }; + } +} + +async function main(): Promise { + const args = parseArgs(process.argv.slice(2)); + const client = createClient(args.endpoint); + + const endpointForLog = args.endpoint.replace(/\/\/[^@]*@/, "//@"); + console.log( + JSON.stringify({ + event: "churn_start", + endpoint: endpointForLog, + count: args.count, + targetTxids: args.targetTxids, + workingSetRows: args.workingSetRows, + rowBytes: args.rowBytes, + commitDelayMs: args.commitDelayMs, + concurrency: args.concurrency, + runId: args.runId, + }), + ); + + const outcomes: DbOutcome[] = new Array(args.count); + let next = 0; + let completed = 0; + let totalTxids = 0; + const startedAt = Date.now(); + + async function worker(): Promise { + while (true) { + const index = next; + if (index >= args.count) return; + next += 1; + const outcome = await churnOne(client, args, index); + outcomes[index] = outcome; + completed += 1; + totalTxids += outcome.txidCount; + console.log( + JSON.stringify({ + event: "churn_db_done", + key: outcome.key, + ok: outcome.ok, + size: fmtBytes(outcome.sizeBytes), + pageCount: outcome.pageCount, + txidCount: outcome.txidCount, + calls: outcome.calls, + completed, + total: args.count, + ...(outcome.error ? { error: outcome.error } : {}), + }), + ); + } + } + + await Promise.all( + new Array(Math.min(args.concurrency, args.count)) + .fill(0) + .map(() => worker()), + ); + + const ok = outcomes.filter((o) => o.ok).length; + const failed = outcomes.filter((o) => !o.ok); + console.log( + JSON.stringify({ + event: "churn_end", + ok, + failed: failed.length, + total: args.count, + totalTxids, + elapsedSec: Math.round((Date.now() - startedAt) / 1000), + failures: failed + .slice(0, 20) + .map((o) => ({ key: o.key, error: o.error })), + }), + ); + + if (failed.length > 0) process.exitCode = 1; +} + +main().catch((err) => { + console.error( + JSON.stringify({ + event: "churn_fatal", + error: + err instanceof Error ? (err.stack ?? err.message) : String(err), + }), + ); + process.exit(1); +}); diff --git a/examples/kitchen-sink/src/actors/testing/churn-db.ts b/examples/kitchen-sink/src/actors/testing/churn-db.ts new file mode 100644 index 0000000000..627d5452d9 --- /dev/null +++ b/examples/kitchen-sink/src/actors/testing/churn-db.ts @@ -0,0 +1,268 @@ +import { actor } from "rivetkit"; +import { db } from "rivetkit/db"; + +// Seeding vessel for the "deep overwrite delta chain" compaction read-hot-shard +// load test (spec H2). +// +// Unlike growDb (append-only, live size == delta total), churnDb repeatedly +// UPDATEs a small, bounded working set in place. Every commit rewrites the same +// few pages, so the live SHARD/PIDX range stays tiny while each commit still +// encodes a fresh DELTA/{txid} blob. +// +// The read-hotspot shape (proven necessary by the 2026-07-10 prod subspace +// scan): the branch's whole contiguous DELTA range must stay SMALL enough to +// live on ONE FDB shard (< ~500 MB) while the chain is DEEP (tens of thousands +// of txids). Then the compactor's stage read of that branch cannot fan out +// across shards. Every slice lands on the same shard's replica procs. The +// earlier 18 GiB "whale" shape was wrong: DD split it across ~450 shards and +// the reads spread, so it never pinned a proc. The fix here is small per-commit +// DELTA (tiny working set, small rows) plus high targetTxids for depth, which +// keeps the total contiguous DELTA under one shard. +// +// It deliberately avoids VACUUM, DELETE, and whole-database scans so the +// database never grows and each churn() call costs O(working set), not +// O(delta chain depth). + +interface ChurnInput { + // Number of rows in the bounded working set that is rewritten every commit. + // Small on purpose: the whole live database stays within one FDB shard. + workingSetRows?: number; + // Payload bytes per row, regenerated server-side via randomblob() on every + // commit so the pages are always dirtied (no no-op page elision). + rowBytes?: number; + // Target cumulative committed-transaction count. churn() is resumable: it + // persists the running count and stops once this target is reached, so the + // driver can drive a whale to a fixed delta-chain depth across many calls. + // Raising the target on a later call resumes churning (this is the "bump" + // that fires compaction after the compaction-enabled build is deployed). + targetTxids?: number; + // Wall-clock budget for a single churn() call in milliseconds. When the + // budget is exceeded the call returns done=false so the driver can resume. + budgetMs?: number; + // Delay in milliseconds between commits. Throttles the per-whale write rate + // so the depot sqlite_commit path (which waits on FDB durability) does not + // outrun the storage servers. Each whale appends DELTA/{txid} to one + // contiguous txid-ordered tail, so unthrottled rapid churn is an + // append-hotspot whose durability lag exceeds the commit timeout. A small + // delay keeps the tail shard's durability lag under that bound. + commitDelayMs?: number; +} + +interface StorageStats { + pageCount: number; + pageSize: number; + freelistCount: number; + sizeBytes: number; +} + +interface ChurnResult extends StorageStats { + done: boolean; + txidCount: number; + targetTxids: number; + commitsThisCall: number; + elapsedMs: number; +} + +// Sub-shard-deep shape: a tiny working set (few dirty pages per commit) over a +// deep chain keeps the whole DELTA range under one FDB shard. At ~12-16 KiB of +// DELTA per commit, 25k txids is ~300-400 MB, i.e. one shard. Verify the +// resulting per-branch DELTA/shard-count with fdb_subspace_sizes.py after +// seeding and tune targetTxids to stay single-shard. +const DEFAULT_WORKING_SET_ROWS = 16; +const DEFAULT_ROW_BYTES = 256; +const DEFAULT_TARGET_TXIDS = 25_000; +const DEFAULT_BUDGET_MS = 120_000; +const DEFAULT_COMMIT_DELAY_MS = 25; + +function nonNegativeInt(value: number | undefined, fallback: number): number { + if (value === undefined) return fallback; + if (!Number.isFinite(value) || value < 0) { + throw new Error(`expected a non-negative finite number, got ${value}`); + } + return Math.floor(value); +} + +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +function finiteInt(value: number | undefined, fallback: number): number { + if (value === undefined) return fallback; + if (!Number.isFinite(value) || value <= 0) { + throw new Error(`expected a positive finite number, got ${value}`); + } + return Math.floor(value); +} + +async function queryOne( + database: { + execute: (sql: string, ...args: unknown[]) => Promise; + }, + sql: string, +): Promise { + const rows = await database.execute(sql); + if (!rows[0]) throw new Error(`query returned no rows: ${sql}`); + return rows[0] as T; +} + +async function storageStats(database: { + execute: (sql: string, ...args: unknown[]) => Promise; +}): Promise { + const [pageCount, pageSize, freelistCount] = await Promise.all([ + queryOne<{ page_count: number }>(database, "PRAGMA page_count"), + queryOne<{ page_size: number }>(database, "PRAGMA page_size"), + queryOne<{ freelist_count: number }>(database, "PRAGMA freelist_count"), + ]); + return { + pageCount: pageCount.page_count, + pageSize: pageSize.page_size, + freelistCount: freelistCount.freelist_count, + sizeBytes: pageCount.page_count * pageSize.page_size, + }; +} + +// Lazily seed the bounded working set. Runs once per database as a single small +// commit; subsequent calls are cheap no-ops once the rows exist. +async function ensureWorkingSet( + database: { + execute: (sql: string, ...args: unknown[]) => Promise; + }, + workingSetRows: number, + rowBytes: number, +): Promise { + const existing = await queryOne<{ n: number }>( + database, + "SELECT COUNT(*) AS n FROM churn_rows", + ); + if (existing.n >= workingSetRows) return; + + await database.execute("BEGIN"); + try { + for (let id = existing.n + 1; id <= workingSetRows; id += 1) { + await database.execute( + "INSERT INTO churn_rows (id, payload, rev) VALUES (?, randomblob(?), 0)", + id, + rowBytes, + ); + } + await database.execute("COMMIT"); + } catch (err) { + await database.execute("ROLLBACK").catch(() => undefined); + throw err; + } +} + +export const churnDb = actor({ + options: { + // Generous per-action timeout. churn() self-limits with budgetMs and + // returns done=false before this fires. + actionTimeout: 300_000, + }, + db: db({ + onMigrate: async (database) => { + // Bounded working set rewritten in place on every commit. + await database.execute(` + CREATE TABLE IF NOT EXISTS churn_rows ( + id INTEGER PRIMARY KEY, + payload BLOB NOT NULL, + rev INTEGER NOT NULL + ) + `); + // Single-row durable counter of committed churn transactions. Bumped + // inside each churn commit so the delta-chain depth is recoverable + // across calls without scanning history. + await database.execute(` + CREATE TABLE IF NOT EXISTS churn_meta ( + id INTEGER PRIMARY KEY CHECK (id = 0), + txid_count INTEGER NOT NULL + ) + `); + await database.execute( + "INSERT OR IGNORE INTO churn_meta (id, txid_count) VALUES (0, 0)", + ); + }, + }), + actions: { + churn: async (c, input: ChurnInput = {}): Promise => { + const startedAt = performance.now(); + const workingSetRows = finiteInt( + input.workingSetRows, + DEFAULT_WORKING_SET_ROWS, + ); + const rowBytes = finiteInt(input.rowBytes, DEFAULT_ROW_BYTES); + const targetTxids = finiteInt( + input.targetTxids, + DEFAULT_TARGET_TXIDS, + ); + const budgetMs = finiteInt(input.budgetMs, DEFAULT_BUDGET_MS); + const commitDelayMs = nonNegativeInt( + input.commitDelayMs, + DEFAULT_COMMIT_DELAY_MS, + ); + + await ensureWorkingSet(c.db, workingSetRows, rowBytes); + + let { txid_count: txidCount } = await queryOne<{ + txid_count: number; + }>(c.db, "SELECT txid_count FROM churn_meta WHERE id = 0"); + let commitsThisCall = 0; + + while (txidCount < targetTxids) { + if (performance.now() - startedAt >= budgetMs) { + const stats = await storageStats(c.db); + return { + ...stats, + done: false, + txidCount, + targetTxids, + commitsThisCall, + elapsedMs: Math.round(performance.now() - startedAt), + }; + } + + // One BEGIN/UPDATE/COMMIT rewrites the entire working set and + // bumps the counter, producing exactly one uncompacted DELTA in + // FDB over the same fixed page range. + await c.db.execute("BEGIN"); + try { + await c.db.execute( + "UPDATE churn_rows SET payload = randomblob(?), rev = rev + 1", + rowBytes, + ); + await c.db.execute( + "UPDATE churn_meta SET txid_count = txid_count + 1 WHERE id = 0", + ); + await c.db.execute("COMMIT"); + } catch (err) { + await c.db.execute("ROLLBACK").catch(() => undefined); + throw err; + } + + txidCount += 1; + commitsThisCall += 1; + + // Throttle the write rate so the depot commit path does not + // outrun FDB durability on the whale's append-hotspot tail shard. + if (commitDelayMs > 0) await sleep(commitDelayMs); + } + + const stats = await storageStats(c.db); + return { + ...stats, + done: true, + txidCount, + targetTxids, + commitsThisCall, + elapsedMs: Math.round(performance.now() - startedAt), + }; + }, + stats: async (c): Promise => { + const stats = await storageStats(c.db); + const meta = await queryOne<{ txid_count: number }>( + c.db, + "SELECT txid_count FROM churn_meta WHERE id = 0", + ); + return { ...stats, txidCount: meta.txid_count }; + }, + }, +}); diff --git a/examples/kitchen-sink/src/index.ts b/examples/kitchen-sink/src/index.ts index 0712f2446d..6196e5f3a4 100644 --- a/examples/kitchen-sink/src/index.ts +++ b/examples/kitchen-sink/src/index.ts @@ -89,6 +89,7 @@ import { uniqueVarActor, } from "./actors/state/vars.ts"; // Testing +import { churnDb } from "./actors/testing/churn-db.ts"; import { growDb } from "./actors/testing/grow-db.ts"; import { inlineClientActor } from "./actors/testing/inline-client.ts"; import { loadTestAgent } from "./actors/testing/load-test-agent.ts"; @@ -291,6 +292,7 @@ export const registry = setup({ rawSqliteFuzzer, sqliteMemoryPressure, growDb, + churnDb, mockAgenticLoop, sleepCloseFuzz, loadTestAgent,