From 931229909a54203e80690e6f323b83c760da9a3b Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Sat, 18 Jul 2026 21:02:07 +0000 Subject: [PATCH 1/4] feat(bulldozer-js): add declareBatchedPiledriverDatabase batching wrapper Co-Authored-By: Konstantin Wohlwend --- .../src/databases/piledriver/batched.test.ts | 170 +++++++++++ .../src/databases/piledriver/batched.ts | 274 ++++++++++++++++++ .../src/databases/piledriver/index.ts | 3 + .../src/payments/schema/performance.test.ts | 21 +- 4 files changed, 465 insertions(+), 3 deletions(-) create mode 100644 apps/bulldozer-js/src/databases/piledriver/batched.test.ts create mode 100644 apps/bulldozer-js/src/databases/piledriver/batched.ts diff --git a/apps/bulldozer-js/src/databases/piledriver/batched.test.ts b/apps/bulldozer-js/src/databases/piledriver/batched.test.ts new file mode 100644 index 000000000..393cbcace --- /dev/null +++ b/apps/bulldozer-js/src/databases/piledriver/batched.test.ts @@ -0,0 +1,170 @@ +import { encodeBase64 } from "@hexclave/shared/dist/utils/bytes"; +import { describe, expect, it } from "vitest"; +import { DatabaseSeq } from "../index.js"; +import { declareInMemoryLowLevelDatabase } from "../low-level/implementations/in-memory.js"; +import { declareBatchedPiledriverDatabase } from "./batched.js"; +import { declarePiledriverDatabase, PiledriverDatabase, PiledriverObject } from "./index.js"; + +const encoder = new TextEncoder(); +const keyOf = (value: string) => encoder.encode(value).buffer; + +type FakeBase = PiledriverDatabase & { + setCount: number, + deleteCount: number, + lastSetValueByKey: Map, + durableSeqs: DatabaseSeq[], + availableSeqs: DatabaseSeq[], +}; + +// A PiledriverDatabase that counts underlying calls, used to assert batching behavior. It delegates +// to a real in-memory piledriver base so all seqs are genuine DatabaseSeq values (no casts needed). +function createCountingBase(): FakeBase { + const inner = declarePiledriverDatabase(declareInMemoryLowLevelDatabase(crypto.randomUUID())); + const lastSetValueByKey = new Map(); + const durableSeqs: DatabaseSeq[] = []; + const availableSeqs: DatabaseSeq[] = []; + const keyBase64 = (key: ArrayBuffer) => encodeBase64(new Uint8Array(key)); + + return { + setCount: 0, + deleteCount: 0, + lastSetValueByKey, + durableSeqs, + availableSeqs, + getDebugInfo() { + return inner.getDebugInfo(); + }, + async getRootObject(key) { + return await inner.getRootObject(key); + }, + async setRootObject(key, value) { + this.setCount++; + lastSetValueByKey.set(keyBase64(key), value); + return await inner.setRootObject(key, value); + }, + async deleteRootObject(key) { + this.deleteCount++; + return await inner.deleteRootObject(key); + }, + async waitUntilAvailable(seq) { + availableSeqs.push(seq); + await inner.waitUntilAvailable(seq); + }, + async waitUntilDurable(seq) { + durableSeqs.push(seq); + await inner.waitUntilDurable(seq); + }, + async waitUntilReplicated(seq) { + durableSeqs.push(seq); + await inner.waitUntilReplicated(seq); + }, + combineSeqs(...seqs) { + return inner.combineSeqs(...seqs); + }, + initialSeq: inner.initialSeq, + }; +} + +// Resolves once the given promise settles or a microtask/timer boundary passes, whichever is first, +// so we can assert that a promise has NOT resolved yet without hanging the test. +async function isPending(promise: Promise): Promise { + const sentinel = Symbol("pending"); + const winner = await Promise.race([promise.then(() => "resolved"), new Promise(resolve => setTimeout(() => resolve(sentinel), 20))]); + return winner === sentinel; +} + +describe("declareBatchedPiledriverDatabase", () => { + it("serves the latest set value from memory before any flush", async () => { + const base = createCountingBase(); + const batched = declareBatchedPiledriverDatabase(base, { batchIntervalMs: 10_000 }); + const key = keyOf("root"); + + await batched.setRootObject(key, { value: 1 }); + const { object } = await batched.getRootObject(key); + + expect(object).toEqual({ value: 1 }); + expect(base.setCount).toBe(0); + + await batched.close(); + }); + + it("coalesces rapid writes to one key into a single underlying set with the last value", async () => { + const base = createCountingBase(); + const batched = declareBatchedPiledriverDatabase(base, { batchIntervalMs: 10_000 }); + const key = keyOf("root"); + + for (let i = 0; i < 25; i++) await batched.setRootObject(key, { value: i }); + expect(base.setCount).toBe(0); + + await batched.flushAll(); + + expect(base.setCount).toBe(1); + expect(base.deleteCount).toBe(0); + expect(base.lastSetValueByKey.get(encodeBase64(new Uint8Array(key)))).toEqual({ value: 24 }); + + await batched.close(); + }); + + it("coalesces a set followed by a delete into a single underlying delete and reports not-found", async () => { + const base = createCountingBase(); + const batched = declareBatchedPiledriverDatabase(base, { batchIntervalMs: 10_000 }); + const key = keyOf("root"); + + await batched.setRootObject(key, { value: 1 }); + await batched.deleteRootObject(key); + await expect(batched.getRootObject(key)).rejects.toThrow("Root object not found"); + + await batched.flushAll(); + + expect(base.setCount).toBe(0); + expect(base.deleteCount).toBe(1); + await expect(batched.getRootObject(key)).rejects.toThrow("Root object not found"); + + await batched.close(); + }); + + it("resolves waitUntilAvailable immediately but waitUntilDurable/Replicated only after the flush delegates to the base", async () => { + const base = createCountingBase(); + const batched = declareBatchedPiledriverDatabase(base, { batchIntervalMs: 10_000 }); + const key = keyOf("root"); + + const { seq } = await batched.setRootObject(key, { value: 1 }); + + // Available resolves without a flush having occurred. + await batched.waitUntilAvailable(seq); + expect(base.setCount).toBe(0); + + const durablePromise = batched.waitUntilDurable(seq); + const replicatedPromise = batched.waitUntilReplicated(seq); + expect(await isPending(durablePromise)).toBe(true); + expect(await isPending(replicatedPromise)).toBe(true); + expect(base.durableSeqs).toHaveLength(0); + + await batched.flushAll(); + await durablePromise; + await replicatedPromise; + + expect(base.setCount).toBe(1); + // Durable and replicated each delegated to the base once the flush produced a real base seq. + expect(base.durableSeqs).toHaveLength(2); + + await batched.close(); + }); + + it("flushes to a real in-memory piledriver base so a fresh client reads the persisted value", async () => { + const dbId = crypto.randomUUID(); + const base = declarePiledriverDatabase(declareInMemoryLowLevelDatabase(dbId)); + const batched = declareBatchedPiledriverDatabase(base, { batchIntervalMs: 10_000 }); + const key = keyOf("root"); + const value: PiledriverObject = { nested: { list: [1, 2, 3], flag: true } }; + + await batched.setRootObject(key, value); + await batched.flushAll(); + + const freshBase = declarePiledriverDatabase(declareInMemoryLowLevelDatabase(dbId)); + const { object } = await freshBase.getRootObject(key); + expect(object).toEqual(value); + + await batched.close(); + }); +}); diff --git a/apps/bulldozer-js/src/databases/piledriver/batched.ts b/apps/bulldozer-js/src/databases/piledriver/batched.ts new file mode 100644 index 000000000..44a382972 --- /dev/null +++ b/apps/bulldozer-js/src/databases/piledriver/batched.ts @@ -0,0 +1,274 @@ +import { encodeBase64 } from "@hexclave/shared/dist/utils/bytes"; +import { DatabaseSeq } from "../index.js"; +import { PiledriverDatabase, PiledriverObject } from "./index.js"; + +export type BatchedPiledriverDatabaseOptions = { + /** + * The maximum rate at which the wrapper flushes writes to the underlying database: there is at + * most one underlying `setRootObject`/`deleteRootObject` call per key per `batchIntervalMs`. + * Defaults to 200 ms. + */ + batchIntervalMs?: number, +}; + +/** + * A `PiledriverDatabase` that adds `flushAll`/`close` for draining pending batched writes and + * cancelling flush timers so none fire after teardown. + */ +export type BatchedPiledriverDatabase = PiledriverDatabase & { + /** + * Flushes all pending batched writes to the underlying database now and awaits their underlying + * (base) seqs. Re-throws the first error encountered by any underlying flush (fail loud). + */ + flushAll(): Promise, + /** + * Flushes all pending writes and cancels pending flush timers. + */ + close(): Promise, +}; + +type Deferred = { + promise: Promise, + resolve: (value: T) => void, + reject: (error: unknown) => void, +}; + +function createDeferred(): Deferred { + let resolve!: (value: T) => void; + let reject!: (error: unknown) => void; + const promise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + return { promise, resolve, reject }; +} + +type PendingWrite = + | { op: "set", value: PiledriverObject } + | { op: "delete" }; + +// The latest logical state of a key, used to serve reads instantly (whether it is a pending write +// or a value cached from a delegated read). `seq` is the seq getRootObject returns for this state. +type LatestState = + | { present: true, value: PiledriverObject, seq: DatabaseSeq } + | { present: false, seq: DatabaseSeq }; + +type KeyEntry = { + key: ArrayBuffer, + latest: LatestState | undefined, + // The un-flushed write for this key (null when everything has been flushed), plus the deferred + // that all synthetic seqs minted for the current batch window point to. It resolves to the base + // seq returned by the coalesced flush that includes those writes. + pending: PendingWrite | null, + pendingDeferred: Deferred | null, + flushTimer: ReturnType | null, + // The tail of this key's flush chain, so flushes for the same key never overlap or reorder + // (latest-write-wins would be violated if an older slow flush finished after a newer one). + inFlight: Promise, +}; + +/** + * Wraps a `PiledriverDatabase` and batches root writes: for each key there is at most one + * underlying `setRootObject`/`deleteRootObject` call per `batchIntervalMs`, coalescing to the + * latest value (latest-write-wins; a delete after a set flushes as a delete). + * + * Reads are served from an in-memory "latest" state, so `setRootObject`/`deleteRootObject` return + * immediately without awaiting the underlying write, and a subsequent `getRootObject` sees the + * pending value right away. This is safe because Piledriver objects are immutable, so holding and + * handing out references across reads without copying does not risk exposing mutations. + * + * ## Seq semantics + * Because reads are served from in-memory state, a write's seq is already "available" to this + * client before it is durable. We therefore mint synthetic seqs for batched writes and keep a + * registry mapping each synthetic seq to a promise that resolves to the real base seq once the + * coalesced flush that includes the write completes: + * - `waitUntilAvailable` resolves immediately for synthetic seqs (the value is already visible). + * - `waitUntilDurable`/`waitUntilReplicated` await the flush, then delegate to the base with the + * real base seq. + * + * Seqs returned by `getRootObject` when we delegate to the base are real base seqs, not synthetic. + * We distinguish the two by identity via the synthetic-seq registry: a seq present in the registry + * is synthetic (resolve it to a base seq before delegating), otherwise it is a base pass-through + * seq (delegate to the base directly). This keeps the common all-base `combineSeqs` path free of + * synthetic-seq allocation. + */ +export function declareBatchedPiledriverDatabase(basePiledriverDb: PiledriverDatabase, options: BatchedPiledriverDatabaseOptions = {}): BatchedPiledriverDatabase { + const batchIntervalMs = options.batchIntervalMs ?? 200; + if (!Number.isFinite(batchIntervalMs) || batchIntervalMs < 0) throw new Error("batchIntervalMs must be a non-negative finite number"); + + const entries = new Map(); + // Maps each synthetic seq (a fresh array object) to a promise resolving to the real base seq the + // coalesced flush produces. A WeakMap keyed by the seq object lets synthetic seqs be collected + // once no caller holds them, and doubles as the "is this seq synthetic?" test. + const syntheticSeqRegistry = new WeakMap>(); + // The first error thrown by any underlying flush, remembered so it is surfaced (fail loud) on the + // next flushAll/close rather than silently swallowed. + let flushError: unknown = undefined; + + const keyBase64Of = (key: ArrayBuffer) => encodeBase64(new Uint8Array(key)); + + const getOrCreateEntry = (keyBase64: string, key: ArrayBuffer): KeyEntry => { + let entry = entries.get(keyBase64); + if (entry === undefined) { + entry = { key, latest: undefined, pending: null, pendingDeferred: null, flushTimer: null, inFlight: Promise.resolve() }; + entries.set(keyBase64, entry); + } + return entry; + }; + + const mintSyntheticSeq = (baseSeqPromise: Promise): DatabaseSeq => { + // Branded seqs can only be constructed via an assertion; the low-level implementations mint + // their seqs the same way (see in-memory.ts / instant-availability.ts). + const seq = ["batched-piledriver-synthetic-seq", crypto.randomUUID()] as unknown as DatabaseSeq; + syntheticSeqRegistry.set(seq, baseSeqPromise); + return seq; + }; + + // Resolves a seq to a promise of a real base seq: synthetic seqs await their flush, base + // pass-through seqs resolve to themselves. Never awaits here so the fast (sync) path stays cheap. + const baseSeqPromiseOf = (seq: DatabaseSeq): Promise => syntheticSeqRegistry.get(seq) ?? Promise.resolve(seq); + + const performFlush = async (entry: KeyEntry, pending: PendingWrite, deferred: Deferred, previous: Promise): Promise => { + // Preserve per-key ordering; `previous` never rejects (this function records errors instead of + // throwing), but guard defensively so a future change can't turn it into an unhandled rejection. + await previous.catch(() => {}); + try { + const { seq } = pending.op === "set" + ? await basePiledriverDb.setRootObject(entry.key, pending.value) + : await basePiledriverDb.deleteRootObject(entry.key); + deferred.resolve(seq); + } catch (error) { + flushError ??= error; + deferred.reject(error); + } + }; + + // Moves the entry's currently-buffered write into an in-flight flush. Safe to call from the timer + // or from flushAll; a no-op when there is nothing pending. + const startFlush = (entry: KeyEntry) => { + if (entry.flushTimer !== null) { + clearTimeout(entry.flushTimer); + entry.flushTimer = null; + } + if (entry.pending === null || entry.pendingDeferred === null) return; + const pending = entry.pending; + const deferred = entry.pendingDeferred; + entry.pending = null; + entry.pendingDeferred = null; + entry.inFlight = performFlush(entry, pending, deferred, entry.inFlight); + }; + + const scheduleWrite = (key: ArrayBuffer, write: PendingWrite): DatabaseSeq => { + const keyBase64 = keyBase64Of(key); + const entry = getOrCreateEntry(keyBase64, key); + entry.pending = write; + if (entry.pendingDeferred === null) { + const deferred = createDeferred(); + // Guard against an unhandled rejection when no caller ever awaits this write's durability; the + // error is still remembered in flushError and surfaced by flushAll/close (fail loud). + deferred.promise.catch(() => {}); + entry.pendingDeferred = deferred; + } + const seq = mintSyntheticSeq(entry.pendingDeferred.promise); + entry.latest = write.op === "set" ? { present: true, value: write.value, seq } : { present: false, seq }; + // Fixed-window throttle: the first write in a window schedules the single flush; later writes in + // the same window only update `latest`/`pending`, guaranteeing at most one flush per interval. + if (entry.flushTimer === null) { + entry.flushTimer = setTimeout(() => startFlush(entry), batchIntervalMs); + } + return seq; + }; + + const flushAll = async (): Promise => { + const settled: Promise[] = []; + for (const entry of entries.values()) { + startFlush(entry); + settled.push(entry.inFlight.catch(() => {})); + } + await Promise.all(settled); + if (flushError !== undefined) throw flushError; + }; + + const result: BatchedPiledriverDatabase = { + getDebugInfo() { + return { + backend: "batched-piledriver", + constructorArguments: { basePiledriverDb, options }, + basePiledriverDb, + batchIntervalMs, + pendingKeys: [...entries.entries()].filter(([, entry]) => entry.pending !== null).map(([keyBase64]) => keyBase64), + hasFlushError: flushError !== undefined, + }; + }, + async getRootObject(key): Promise<{ object: PiledriverObject, seq: DatabaseSeq }> { + const keyBase64 = keyBase64Of(key); + const entry = entries.get(keyBase64); + if (entry?.latest !== undefined) { + if (entry.latest.present) return { object: entry.latest.value, seq: entry.latest.seq }; + // Mirror the base's not-found behavior exactly; bulldozer matches on this message. + throw new Error("Root object not found"); + } + const { object, seq } = await basePiledriverDb.getRootObject(key); + // Cache the read result so subsequent reads are instant. Only cache when nothing is pending or + // already cached, to avoid clobbering a newer in-memory state written concurrently. + const cacheEntry = getOrCreateEntry(keyBase64, key); + if (cacheEntry.latest === undefined && cacheEntry.pending === null) { + cacheEntry.latest = { present: true, value: object, seq }; + } + return { object, seq }; + }, + async setRootObject(key, value): Promise<{ seq: DatabaseSeq }> { + return { seq: scheduleWrite(key, { op: "set", value }) }; + }, + async deleteRootObject(key): Promise<{ seq: DatabaseSeq }> { + return { seq: scheduleWrite(key, { op: "delete" }) }; + }, + async waitUntilAvailable(seq): Promise { + // Synthetic writes are already visible to reads from this client; base pass-through seqs + // delegate (they come from getRootObject, which read them, so this resolves immediately). + if (syntheticSeqRegistry.has(seq)) return; + await basePiledriverDb.waitUntilAvailable(seq); + }, + async waitUntilDurable(seq): Promise { + await basePiledriverDb.waitUntilDurable(await baseSeqPromiseOf(seq)); + }, + async waitUntilReplicated(seq): Promise { + await basePiledriverDb.waitUntilReplicated(await baseSeqPromiseOf(seq)); + }, + combineSeqs(...seqs): DatabaseSeq { + // Fast path: when no seq is synthetic, delegate directly so we don't allocate a synthetic + // combined seq (the base already deduplicates and drops initial seqs). + if (!seqs.some(seq => syntheticSeqRegistry.has(seq))) return basePiledriverDb.combineSeqs(...seqs); + const componentPromises = seqs.map(baseSeqPromiseOf); + const combinedBaseSeqPromise = (async () => basePiledriverDb.combineSeqs(...await Promise.all(componentPromises)))(); + combinedBaseSeqPromise.catch(() => {}); + return mintSyntheticSeq(combinedBaseSeqPromise); + }, + async flushAll(): Promise { + await flushAll(); + }, + async close(): Promise { + // flushAll drains pending writes and clears every pending flush timer, so none can fire after + // teardown. The base PiledriverDatabase has no close() of its own to delegate to. + await flushAll(); + }, + initialSeq: basePiledriverDb.initialSeq, + }; + + if (basePiledriverDb.debugSnapshot !== undefined) { + const baseDebugSnapshot = basePiledriverDb.debugSnapshot.bind(basePiledriverDb); + result.debugSnapshot = async () => { + await flushAll(); + return await baseDebugSnapshot(); + }; + } + if (basePiledriverDb.debugLowLevelSnapshot !== undefined) { + const baseDebugLowLevelSnapshot = basePiledriverDb.debugLowLevelSnapshot.bind(basePiledriverDb); + result.debugLowLevelSnapshot = async () => { + await flushAll(); + return await baseDebugLowLevelSnapshot(); + }; + } + + return result; +} diff --git a/apps/bulldozer-js/src/databases/piledriver/index.ts b/apps/bulldozer-js/src/databases/piledriver/index.ts index 52e4670eb..9ed6ccbc4 100644 --- a/apps/bulldozer-js/src/databases/piledriver/index.ts +++ b/apps/bulldozer-js/src/databases/piledriver/index.ts @@ -3,6 +3,9 @@ import { traceSpan, traceSpanHot } from "../../otel.js"; import { Database, DatabaseSeq } from "../index.js"; import { LowLevelDatabase, LowLevelDatabaseDebugSnapshot } from "../low-level/index.js"; +export { declareBatchedPiledriverDatabase } from "./batched.js"; +export type { BatchedPiledriverDatabase, BatchedPiledriverDatabaseOptions } from "./batched.js"; + export const isPiledriverHeapObjectSymbol = Symbol.for("hexclave-piledriver-heap-object-symbol"); export type PiledriverHeapObject = { get(): Promise, diff --git a/apps/bulldozer-js/src/payments/schema/performance.test.ts b/apps/bulldozer-js/src/payments/schema/performance.test.ts index 689f837ce..e80cfbb7e 100644 --- a/apps/bulldozer-js/src/payments/schema/performance.test.ts +++ b/apps/bulldozer-js/src/payments/schema/performance.test.ts @@ -3,7 +3,7 @@ import { declareInMemoryLowLevelDatabase } from "../../databases/low-level/imple import { declareInstantAvailabilityLowLevelDatabase } from "../../databases/low-level/implementations/instant-availability.js"; import { declareLmdbLowLevelDatabase } from "../../databases/low-level/implementations/lmdb.js"; import { declareBulldozerDatabase } from "../../databases/bulldozer/index.js"; -import { declarePiledriverDatabase, PiledriverObject } from "../../databases/piledriver/index.js"; +import { declareBatchedPiledriverDatabase, declarePiledriverDatabase, PiledriverDatabase, PiledriverObject } from "../../databases/piledriver/index.js"; import { createPaymentsSchema } from "./index.js"; import type { ProductSnapshot, SubscriptionRow } from "./types.js"; import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; @@ -106,9 +106,22 @@ const newLowLevelDb = () => { } return declareInMemoryLowLevelDatabase(crypto.randomUUID()); }; +// Databases whose pending writes must be drained (and timers cancelled) before afterAll removes the +// temp dirs, so no batched flush fires against a deleted LMDB directory after the test finishes. +const databasesToClose: { close(): Promise }[] = []; +const newPiledriverDb = (): PiledriverDatabase => { + const base = declarePiledriverDatabase(newLowLevelDb()); + if (process.env.BULLDOZER_BATCHED === "1") { + const batchIntervalMs = Number(process.env.BULLDOZER_BATCH_INTERVAL_MS ?? 200); + const batched = declareBatchedPiledriverDatabase(base, { batchIntervalMs }); + databasesToClose.push(batched); + return batched; + } + return base; +}; const newPaymentsDb = async () => { const schema = createPaymentsSchema(); - const db = declareBulldozerDatabase(declarePiledriverDatabase(newLowLevelDb()), { migrations: schema.migrations }); + const db = declareBulldozerDatabase(newPiledriverDb(), { migrations: schema.migrations }); await db.applyRemainingMigrations(); return { db, schema }; }; @@ -239,6 +252,8 @@ describe("transactions listing performance", () => { }); }); -afterAll(() => { +afterAll(async () => { + // Drain batched writes and cancel flush timers before deleting the temp dirs they write into. + for (const db of databasesToClose) await db.close(); for (const path of tempPaths) rmSync(path, { recursive: true, force: true }); }); From 97e870a564ad777d606c609bf250b0af33c7d5fd Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Sat, 18 Jul 2026 21:22:44 +0000 Subject: [PATCH 2/4] fix(bulldozer-js): retain failed batched flush for retry; snapshot key defensively Co-Authored-By: Konstantin Wohlwend --- .../src/databases/piledriver/batched.test.ts | 44 +++++++++++++++++++ .../src/databases/piledriver/batched.ts | 27 +++++++++++- 2 files changed, 69 insertions(+), 2 deletions(-) diff --git a/apps/bulldozer-js/src/databases/piledriver/batched.test.ts b/apps/bulldozer-js/src/databases/piledriver/batched.test.ts index 393cbcace..85f65bdf0 100644 --- a/apps/bulldozer-js/src/databases/piledriver/batched.test.ts +++ b/apps/bulldozer-js/src/databases/piledriver/batched.test.ts @@ -167,4 +167,48 @@ describe("declareBatchedPiledriverDatabase", () => { await batched.close(); }); + + it("re-buffers a failed flush for retry instead of dropping the write", async () => { + const base = createCountingBase(); + let failNext = true; + const originalSet = base.setRootObject.bind(base); + base.setRootObject = async (key, value) => { + if (failNext) { + failNext = false; + throw new Error("boom"); + } + return await originalSet(key, value); + }; + const batched = declareBatchedPiledriverDatabase(base, { batchIntervalMs: 10_000 }); + const key = keyOf("root"); + + await batched.setRootObject(key, { value: 7 }); + // The first flush fails loud, but the write must not be dropped. + await expect(batched.flushAll()).rejects.toThrow("boom"); + expect((await batched.getRootObject(key)).object).toEqual({ value: 7 }); + + // The next flush retries the still-buffered write and persists it. + await batched.flushAll(); + expect(base.setCount).toBe(1); + expect((await base.getRootObject(key)).object).toEqual({ value: 7 }); + + await batched.close(); + }); + + it("snapshots the key so caller-side mutation before flush does not corrupt the write", async () => { + const base = createCountingBase(); + const batched = declareBatchedPiledriverDatabase(base, { batchIntervalMs: 10_000 }); + const key = keyOf("root"); + + await batched.setRootObject(key, { value: 1 }); + // Mutate the caller-owned buffer before the timer-driven flush fires. + new Uint8Array(key).fill(0); + await batched.flushAll(); + + // The write persisted under the ORIGINAL key bytes, not the mutated ones. + const { object } = await base.getRootObject(keyOf("root")); + expect(object).toEqual({ value: 1 }); + + await batched.close(); + }); }); diff --git a/apps/bulldozer-js/src/databases/piledriver/batched.ts b/apps/bulldozer-js/src/databases/piledriver/batched.ts index 44a382972..4d442eeb9 100644 --- a/apps/bulldozer-js/src/databases/piledriver/batched.ts +++ b/apps/bulldozer-js/src/databases/piledriver/batched.ts @@ -54,6 +54,9 @@ type LatestState = | { present: false, seq: DatabaseSeq }; type KeyEntry = { + // A defensive copy of the caller's key. Callers own the ArrayBuffer they pass in and may reuse or + // mutate it before our timer-driven flush fires, so we snapshot the bytes to keep the key we flush + // under consistent with the base64 key we index by. key: ArrayBuffer, latest: LatestState | undefined, // The un-flushed write for this key (null when everything has been flushed), plus the deferred @@ -110,7 +113,8 @@ export function declareBatchedPiledriverDatabase(basePiledriverDb: PiledriverDat const getOrCreateEntry = (keyBase64: string, key: ArrayBuffer): KeyEntry => { let entry = entries.get(keyBase64); if (entry === undefined) { - entry = { key, latest: undefined, pending: null, pendingDeferred: null, flushTimer: null, inFlight: Promise.resolve() }; + // slice(0) copies the bytes so a later caller-side mutation of `key` can't change what we flush. + entry = { key: key.slice(0), latest: undefined, pending: null, pendingDeferred: null, flushTimer: null, inFlight: Promise.resolve() }; entries.set(keyBase64, entry); } return entry; @@ -140,6 +144,18 @@ export function declareBatchedPiledriverDatabase(basePiledriverDb: PiledriverDat } catch (error) { flushError ??= error; deferred.reject(error); + // Re-buffer the failed write for retry unless a newer write has already superseded it, so a + // transient backend error doesn't silently drop a value that `latest` still serves. The next + // flushAll/close (or a subsequent write's timer) retries it. Callers that awaited this batch's + // durability still saw the rejection above (fail loud); the retry uses a fresh deferred. + if (entry.pending === null) { + entry.pending = pending; + if (entry.pendingDeferred === null) { + const retryDeferred = createDeferred(); + retryDeferred.promise.catch(() => {}); + entry.pendingDeferred = retryDeferred; + } + } } }; @@ -186,7 +202,14 @@ export function declareBatchedPiledriverDatabase(basePiledriverDb: PiledriverDat settled.push(entry.inFlight.catch(() => {})); } await Promise.all(settled); - if (flushError !== undefined) throw flushError; + // Surface (once) the first error any flush hit, then clear it so a later flushAll after a + // successful retry doesn't keep throwing a stale error. Individual waiters already saw their own + // rejection via the per-write deferred, so this is only a backstop for un-awaited writes. + if (flushError !== undefined) { + const error = flushError; + flushError = undefined; + throw error; + } }; const result: BatchedPiledriverDatabase = { From 365ff88968877c7f0a50114c7dad945ac243a8cc Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Sat, 18 Jul 2026 21:33:31 +0000 Subject: [PATCH 3/4] test(bulldozer-js): await waitUntilReplicated per phase; make perf workload env-scalable Co-Authored-By: Konstantin Wohlwend --- .../src/payments/schema/performance.test.ts | 50 +++++++++++++------ 1 file changed, 36 insertions(+), 14 deletions(-) diff --git a/apps/bulldozer-js/src/payments/schema/performance.test.ts b/apps/bulldozer-js/src/payments/schema/performance.test.ts index e80cfbb7e..5bced2175 100644 --- a/apps/bulldozer-js/src/payments/schema/performance.test.ts +++ b/apps/bulldozer-js/src/payments/schema/performance.test.ts @@ -4,6 +4,7 @@ import { declareInstantAvailabilityLowLevelDatabase } from "../../databases/low- import { declareLmdbLowLevelDatabase } from "../../databases/low-level/implementations/lmdb.js"; import { declareBulldozerDatabase } from "../../databases/bulldozer/index.js"; import { declareBatchedPiledriverDatabase, declarePiledriverDatabase, PiledriverDatabase, PiledriverObject } from "../../databases/piledriver/index.js"; +import { DatabaseSeq } from "../../databases/index.js"; import { createPaymentsSchema } from "./index.js"; import type { ProductSnapshot, SubscriptionRow } from "./types.js"; import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; @@ -13,9 +14,12 @@ import { join } from "node:path"; type Metric = { name: string, count: number, elapsedMs: number, opsPerSecond: number }; type Snapshot = Awaited["getSnapshot"]>>["snapshot"]; -const USER_COUNT = 6; -const ITEM_UPDATES_PER_USER = 10; -const PREFILL_USER_COUNT = 200; +// Iteration counts are env-overridable so the same workload can be scaled up for benchmarking (where +// each phase should take several seconds so per-op cost dominates over fixed batching latency). The +// defaults are kept small so the committed test stays fast in CI. +const USER_COUNT = Number(process.env.BULLDOZER_PERF_USER_COUNT ?? 6); +const ITEM_UPDATES_PER_USER = Number(process.env.BULLDOZER_PERF_ITEM_UPDATES_PER_USER ?? 10); +const PREFILL_USER_COUNT = Number(process.env.BULLDOZER_PERF_PREFILL_USER_COUNT ?? 200); const PREFILL_ITEM_UPDATES_PER_USER = 4; const PREFILL_SOURCE_FACT_COUNT = PREFILL_USER_COUNT * (2 + PREFILL_ITEM_UPDATES_PER_USER); const MONTH_MS = 2_592_000_000; @@ -121,9 +125,19 @@ const newPiledriverDb = (): PiledriverDatabase => { }; const newPaymentsDb = async () => { const schema = createPaymentsSchema(); - const db = declareBulldozerDatabase(newPiledriverDb(), { migrations: schema.migrations }); + const piledriver = newPiledriverDb(); + const db = declareBulldozerDatabase(piledriver, { migrations: schema.migrations }); await db.applyRemainingMigrations(); - return { db, schema }; + return { db, schema, piledriver }; +}; +// Waits until every write made during a phase is replicated (durable + on all replicas), the way a +// high-throughput caller that needs durability would. Batching only pays off if this end-of-phase +// wait is included in the timing: otherwise a batched write just defers its durability cost and +// looks artificially fast. All bulldozer writes go to the same piledriver root key, so combining the +// per-write seqs and waiting once reflects the amortized cost of the coalesced flushes. +const waitPhaseReplicated = async (piledriver: PiledriverDatabase, seqs: DatabaseSeq[]) => { + if (seqs.length === 0) return; + await piledriver.waitUntilReplicated(piledriver.combineSeqs(...seqs)); }; describe("payments schema performance", () => { @@ -132,44 +146,52 @@ describe("payments schema performance", () => { let initialized!: Awaited>; initialized = await measure(metrics, "initialize schema", 1, newPaymentsDb); - const { db, schema } = initialized; + const { db, schema, piledriver } = initialized; await measure(metrics, "prefill baseline rows", PREFILL_SOURCE_FACT_COUNT, async () => { + const seqs: DatabaseSeq[] = []; for (let i = 0; i < PREFILL_USER_COUNT; i++) { - await db.withSnapshot(async snapshot => await snapshot.setOrDeleteRow({ tableId: schema.subscriptions, rowIdentifier: `prefill-sub-${i}`, newRowData: subscription(i, "prefill-") as unknown as PiledriverObject })); - await db.withSnapshot(async snapshot => await snapshot.setOrDeleteRow({ tableId: schema.oneTimePurchases, rowIdentifier: `prefill-otp-${i}`, newRowData: oneTimePurchase(i, "prefill-") as unknown as PiledriverObject })); + seqs.push((await db.withSnapshot(async snapshot => await snapshot.setOrDeleteRow({ tableId: schema.subscriptions, rowIdentifier: `prefill-sub-${i}`, newRowData: subscription(i, "prefill-") as unknown as PiledriverObject }))).seq); + seqs.push((await db.withSnapshot(async snapshot => await snapshot.setOrDeleteRow({ tableId: schema.oneTimePurchases, rowIdentifier: `prefill-otp-${i}`, newRowData: oneTimePurchase(i, "prefill-") as unknown as PiledriverObject }))).seq); for (let updateIndex = 0; updateIndex < PREFILL_ITEM_UPDATES_PER_USER; updateIndex++) { - await db.withSnapshot(async snapshot => await snapshot.setOrDeleteRow({ + seqs.push((await db.withSnapshot(async snapshot => await snapshot.setOrDeleteRow({ tableId: schema.manualItemQuantityChanges, rowIdentifier: `prefill-miqc-${i}-${updateIndex}`, newRowData: manualItemQuantityChange(i, updateIndex, "prefill-"), - })); + }))).seq); } } + await waitPhaseReplicated(piledriver, seqs); }); await measure(metrics, "write subscriptions", USER_COUNT, async () => { + const seqs: DatabaseSeq[] = []; for (let i = 0; i < USER_COUNT; i++) { - await db.withSnapshot(async snapshot => await snapshot.setOrDeleteRow({ tableId: schema.subscriptions, rowIdentifier: `sub-${i}`, newRowData: subscription(i) as unknown as PiledriverObject })); + seqs.push((await db.withSnapshot(async snapshot => await snapshot.setOrDeleteRow({ tableId: schema.subscriptions, rowIdentifier: `sub-${i}`, newRowData: subscription(i) as unknown as PiledriverObject }))).seq); } + await waitPhaseReplicated(piledriver, seqs); }); await measure(metrics, "write one-time purchases", USER_COUNT, async () => { + const seqs: DatabaseSeq[] = []; for (let i = 0; i < USER_COUNT; i++) { - await db.withSnapshot(async snapshot => await snapshot.setOrDeleteRow({ tableId: schema.oneTimePurchases, rowIdentifier: `otp-${i}`, newRowData: oneTimePurchase(i) as unknown as PiledriverObject })); + seqs.push((await db.withSnapshot(async snapshot => await snapshot.setOrDeleteRow({ tableId: schema.oneTimePurchases, rowIdentifier: `otp-${i}`, newRowData: oneTimePurchase(i) as unknown as PiledriverObject }))).seq); } + await waitPhaseReplicated(piledriver, seqs); }); await measure(metrics, "write manual item quantity changes", USER_COUNT * ITEM_UPDATES_PER_USER, async () => { + const seqs: DatabaseSeq[] = []; for (let userIndex = 0; userIndex < USER_COUNT; userIndex++) { for (let updateIndex = 0; updateIndex < ITEM_UPDATES_PER_USER; updateIndex++) { - await db.withSnapshot(async snapshot => await snapshot.setOrDeleteRow({ + seqs.push((await db.withSnapshot(async snapshot => await snapshot.setOrDeleteRow({ tableId: schema.manualItemQuantityChanges, rowIdentifier: `miqc-${userIndex}-${updateIndex}`, newRowData: manualItemQuantityChange(userIndex, updateIndex), - })); + }))).seq); } } + await waitPhaseReplicated(piledriver, seqs); }); const { snapshot } = await db.getSnapshot(); From bbf148f51d9aa8ace4e2788f6f5085e86ea2ec58 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Sat, 18 Jul 2026 21:52:39 +0000 Subject: [PATCH 4/4] feat(bulldozer-js): batch root writes in the server (100ms window) with graceful-shutdown flush Co-Authored-By: Konstantin Wohlwend --- apps/bulldozer-js/src/index.ts | 58 ++++++++++++++++++++++++++++++---- 1 file changed, 52 insertions(+), 6 deletions(-) diff --git a/apps/bulldozer-js/src/index.ts b/apps/bulldozer-js/src/index.ts index d87b170df..efba9e140 100644 --- a/apps/bulldozer-js/src/index.ts +++ b/apps/bulldozer-js/src/index.ts @@ -16,6 +16,7 @@ import { declareInMemoryLowLevelDatabase } from "./databases/low-level/implement import { declareInstantAvailabilityLowLevelDatabase } from "./databases/low-level/implementations/instant-availability.js"; import { declareLmdbLowLevelDatabase } from "./databases/low-level/implementations/lmdb.js"; import type { LowLevelDatabase } from "./databases/low-level/index.js"; +import { declareBatchedPiledriverDatabase, type BatchedPiledriverDatabase } from "./databases/piledriver/batched.js"; import { declarePiledriverDatabase, type PiledriverObject } from "./databases/piledriver/index.js"; import "./load-env.js"; import { instrumentation, traceSpan } from "./otel.js"; @@ -35,6 +36,12 @@ const HEAP_GC_MAX_PASSES = 5; const HEAP_GC_FINALIZATION_DELAY_MS = 20; const HEAP_GC_RETRY_COOLDOWN_MS = 60_000; const TICK_LOOP_SLOW_MS = 500; +// Coalescing root writes into at most one underlying commit+replication per interval trades a small +// durability window (bounded by this interval, and drained on graceful shutdown) for much higher +// write throughput. A payments-workload sweep (lmdb-instant, median of 3) put 100 ms at the knee: +// it keeps ~all the throughput win of larger windows while cutting the worst-case in-memory-only +// window (and per-request latency for durability-awaiting callers) well below 200/400 ms. +const DEFAULT_PILEDRIVER_BATCH_INTERVAL_MS = 100; type BulldozerSnapshot = Awaited>["snapshot"]; type OwnedProductsRow = { ownedProducts: Record }; type SubscriptionMapRow = { subscriptions: Record }; @@ -70,12 +77,27 @@ function createLowLevelDatabase(): LowLevelDatabase { })); } -const bulldozerDb = declareBulldozerDatabase( - declarePiledriverDatabase(createLowLevelDatabase(), { - disableHeapReadCache: process.env.HEXCLAVE_BULLDOZER_JS_DISABLE_PILEDRIVER_HEAP_READ_CACHE === "1", - }), - { migrations: schema.migrations }, -); +// Returns the configured batch interval in ms, or null to disable batching (write straight through). +// `off` or `0` disables it; any other value must be a non-negative finite number. +function resolvePiledriverBatchIntervalMs(): number | null { + const raw = process.env.HEXCLAVE_BULLDOZER_JS_BATCH_INTERVAL_MS; + if (raw === undefined || raw.length === 0) return DEFAULT_PILEDRIVER_BATCH_INTERVAL_MS; + if (raw === "off") return null; + const parsed = Number(raw); + if (!Number.isFinite(parsed) || parsed < 0) throw new Error(`HEXCLAVE_BULLDOZER_JS_BATCH_INTERVAL_MS must be a non-negative finite number or "off", got ${JSON.stringify(raw)}`); + // 0 means "no batching" rather than "flush on a zero-delay timer" (which would still defer a tick). + if (parsed === 0) return null; + return parsed; +} + +const rootPiledriverDb = declarePiledriverDatabase(createLowLevelDatabase(), { + disableHeapReadCache: process.env.HEXCLAVE_BULLDOZER_JS_DISABLE_PILEDRIVER_HEAP_READ_CACHE === "1", +}); +const piledriverBatchIntervalMs = resolvePiledriverBatchIntervalMs(); +const batchedPiledriverDb: BatchedPiledriverDatabase | null = piledriverBatchIntervalMs === null + ? null + : declareBatchedPiledriverDatabase(rootPiledriverDb, { batchIntervalMs: piledriverBatchIntervalMs }); +const bulldozerDb = declareBulldozerDatabase(batchedPiledriverDb ?? rootPiledriverDb, { migrations: schema.migrations }); (globalThis as any).bulldozerDb = bulldozerDb; console.log(`Stored bulldozerDb in globalThis for pid ${process.pid}. Run \`kill -USR1 ${process.pid} && node inspect 127.0.0.1:9229\` and then \`exec("globalThis.bulldozerDb")\` to inspect it.`); await traceSpan("bulldozer-js.applyRemainingMigrations", async () => await bulldozerDb.applyRemainingMigrations()); @@ -1031,6 +1053,30 @@ const app = new Elysia({ adapter: node() }) .listen(port); console.log(`Bulldozer JS server listening on http://localhost:${app.server?.port ?? port}`); + +// Drain pending batched root writes (and cancel flush timers) before the process exits, so the +// in-memory durability window isn't lost on deploy/restart. Errors are captured (not swallowed), and +// we still exit afterward so the orchestrator's shutdown isn't blocked. +if (batchedPiledriverDb !== null) { + const db = batchedPiledriverDb; + let shuttingDown = false; + const flushAndExit = (signal: NodeJS.Signals) => { + if (shuttingDown) return; + shuttingDown = true; + runAsynchronously(async () => { + try { + await db.close(); + logBulldozerService("graceful-flush-complete", { signal }); + } catch (error) { + captureError("bulldozer-js:graceful-flush", error); + } finally { + process.exit(0); + } + }); + }; + process.once("SIGTERM", flushAndExit); + process.once("SIGINT", flushAndExit); +} const startupFields = { port: app.server?.port ?? port, pid: process.pid,