Merge bbf148f51d into f8c890e472

apps/bulldozer-js/src/databases/piledriver/batched.test.ts

@@ -0,0 +1,214 @@
+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<string, PiledriverObject>,
+  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<string, PiledriverObject>();
+  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<unknown>): Promise<boolean> {
+  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();
+  });
+
+  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();
+  });
+});

apps/bulldozer-js/src/databases/piledriver/batched.ts

@@ -0,0 +1,297 @@
+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<void>,
+  /**
+   * Flushes all pending writes and cancels pending flush timers.
+   */
+  close(): Promise<void>,
+};
+
+type Deferred<T> = {
+  promise: Promise<T>,
+  resolve: (value: T) => void,
+  reject: (error: unknown) => void,
+};
+
+function createDeferred<T>(): Deferred<T> {
+  let resolve!: (value: T) => void;
+  let reject!: (error: unknown) => void;
+  const promise = new Promise<T>((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 = {
+  // 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
+  // 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<DatabaseSeq> | null,
+  flushTimer: ReturnType<typeof setTimeout> | 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<void>,
+};
+
+/**
+ * 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<string, KeyEntry>();
+  // 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<object, Promise<DatabaseSeq>>();
+  // 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) {
+      // 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;
+  };
+
+  const mintSyntheticSeq = (baseSeqPromise: Promise<DatabaseSeq>): 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<DatabaseSeq> => syntheticSeqRegistry.get(seq) ?? Promise.resolve(seq);
+
+  const performFlush = async (entry: KeyEntry, pending: PendingWrite, deferred: Deferred<DatabaseSeq>, previous: Promise<void>): Promise<void> => {
+    // 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);
+      // 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<DatabaseSeq>();
+          retryDeferred.promise.catch(() => {});
+          entry.pendingDeferred = retryDeferred;
+        }
+      }
+    }
+  };
+
+  // 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<DatabaseSeq>();
+      // 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<void> => {
+    const settled: Promise<void>[] = [];
+    for (const entry of entries.values()) {
+      startFlush(entry);
+      settled.push(entry.inFlight.catch(() => {}));
+    }
+    await Promise.all(settled);
+    // 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 = {
+    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<void> {
+      // 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<void> {
+      await basePiledriverDb.waitUntilDurable(await baseSeqPromiseOf(seq));
+    },
+    async waitUntilReplicated(seq): Promise<void> {
+      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<void> {
+      await flushAll();
+    },
+    async close(): Promise<void> {
+      // 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;
+}

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<PiledriverObject>,

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<ReturnType<BulldozerDatabase["getSnapshot"]>>["snapshot"];
 type OwnedProductsRow = { ownedProducts: Record<string, unknown> };
 type SubscriptionMapRow = { subscriptions: Record<string, SubscriptionRow> };
@@ -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,

apps/bulldozer-js/src/payments/schema/performance.test.ts

@@ -3,7 +3,8 @@ 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 { 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<ReturnType<ReturnType<typeof declareBulldozerDatabase>["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;
@@ -106,11 +110,34 @@ 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<void> }[] = [];
+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 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", () => {
@@ -119,44 +146,52 @@ describe("payments schema performance", () => {
     let initialized!: Awaited<ReturnType<typeof newPaymentsDb>>;
 
     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();
@@ -239,6 +274,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 });
 });