From 0d70662eaffcfa2ae5d3415f863b0b1b869b8288 Mon Sep 17 00:00:00 2001 From: Konstantin Wohlwend Date: Fri, 3 Jul 2026 12:17:35 -0700 Subject: [PATCH 1/5] Vastly improved LMDB logging --- .../low-level/implementations/lmdb.ts | 104 +++++++++++++++--- 1 file changed, 87 insertions(+), 17 deletions(-) diff --git a/apps/bulldozer-js/src/databases/low-level/implementations/lmdb.ts b/apps/bulldozer-js/src/databases/low-level/implementations/lmdb.ts index 43a20e7dd..82f6fec39 100644 --- a/apps/bulldozer-js/src/databases/low-level/implementations/lmdb.ts +++ b/apps/bulldozer-js/src/databases/low-level/implementations/lmdb.ts @@ -53,7 +53,16 @@ function validateValue(name: string, value: ArrayBuffer) { type LmdbActivityStats = { puts: number, + putBytes: number, + putAwaitTotalMs: number, transactions: number, + transactionTotalMs: number, + transactionQueueWaitTotalMs: number, + transactionActionTotalMs: number, + metaPutTotalMs: number, + transactionCommitTailTotalMs: number, + requiredSeqWaits: number, + requiredSeqWaitTotalMs: number, waitUntilAvailableResolves: number, waitUntilDurableResolves: number, waitUntilAvailableResolveTotalMs: number, @@ -67,7 +76,16 @@ type LmdbActivityStats = { function emptyActivityStats(): LmdbActivityStats { return { puts: 0, + putBytes: 0, + putAwaitTotalMs: 0, transactions: 0, + transactionTotalMs: 0, + transactionQueueWaitTotalMs: 0, + transactionActionTotalMs: 0, + metaPutTotalMs: 0, + transactionCommitTailTotalMs: 0, + requiredSeqWaits: 0, + requiredSeqWaitTotalMs: 0, waitUntilAvailableResolves: 0, waitUntilDurableResolves: 0, waitUntilAvailableResolveTotalMs: 0, @@ -112,7 +130,16 @@ export function declareLmdbLowLevelDatabase(options: { path: string, dbId?: stri dbId, elapsedMs, putsPerSecond: activityStats.puts / elapsedSeconds, + averagePutBytes: activityStats.puts === 0 ? 0 : activityStats.putBytes / activityStats.puts, + averagePutAwaitMs: activityStats.puts === 0 ? 0 : activityStats.putAwaitTotalMs / activityStats.puts, transactionsPerSecond: activityStats.transactions / elapsedSeconds, + averageTransactionMs: activityStats.transactions === 0 ? 0 : activityStats.transactionTotalMs / activityStats.transactions, + averageTransactionQueueWaitMs: activityStats.transactions === 0 ? 0 : activityStats.transactionQueueWaitTotalMs / activityStats.transactions, + averageTransactionActionMs: activityStats.transactions === 0 ? 0 : activityStats.transactionActionTotalMs / activityStats.transactions, + averageMetaPutMs: activityStats.transactions === 0 ? 0 : activityStats.metaPutTotalMs / activityStats.transactions, + averageTransactionCommitTailMs: activityStats.transactions === 0 ? 0 : activityStats.transactionCommitTailTotalMs / activityStats.transactions, + requiredSeqWaitsPerSecond: activityStats.requiredSeqWaits / elapsedSeconds, + averageRequiredSeqWaitMs: activityStats.requiredSeqWaits === 0 ? 0 : activityStats.requiredSeqWaitTotalMs / activityStats.requiredSeqWaits, waitUntilAvailableResolvesPerSecond: activityStats.waitUntilAvailableResolves / elapsedSeconds, waitUntilDurableResolvesPerSecond: activityStats.waitUntilDurableResolves / elapsedSeconds, averageSeqToAvailabilityResolveMs: activityStats.waitUntilAvailableResolves === 0 ? 0 : activityStats.waitUntilAvailableResolveTotalMs / activityStats.waitUntilAvailableResolves, @@ -198,13 +225,32 @@ export function declareLmdbLowLevelDatabase(options: { path: string, dbId?: stri const commit = (requiresSeq: DatabaseSeq, action: (version: number) => Promise) => { const version = nextVersion(); const seqId = nextSeqId(); - const promise = traceSpan({ description: "bulldozer-js.low-level.lmdb.commit", attributes: { "bulldozer.low_level.backend": "lmdb" } }, async () => await waitUntilAvailable(requiresSeq).then(async () => await root.transaction(() => { - activityStats.transactions++; - return (async () => { - await action(version); - await meta.put("seq", version); - })(); - }))); + const promise = traceSpan({ description: "bulldozer-js.low-level.lmdb.commit", attributes: { "bulldozer.low_level.backend": "lmdb" } }, async () => { + const requiredSeqWaitStartedAt = performance.now(); + await waitUntilAvailable(requiresSeq); + activityStats.requiredSeqWaits++; + activityStats.requiredSeqWaitTotalMs += performance.now() - requiredSeqWaitStartedAt; + const transactionStartedAt = performance.now(); + let transactionCallbackFinishedAt: number | null = null; + return await root.transaction(() => { + activityStats.transactionQueueWaitTotalMs += performance.now() - transactionStartedAt; + activityStats.transactions++; + return (async () => { + const actionStartedAt = performance.now(); + await action(version); + activityStats.transactionActionTotalMs += performance.now() - actionStartedAt; + const metaPutStartedAt = performance.now(); + await meta.put("seq", version); + activityStats.metaPutTotalMs += performance.now() - metaPutStartedAt; + })().finally(() => { + transactionCallbackFinishedAt = performance.now(); + }); + }).finally(() => { + const transactionFinishedAt = performance.now(); + activityStats.transactionTotalMs += transactionFinishedAt - transactionStartedAt; + if (transactionCallbackFinishedAt !== null) activityStats.transactionCommitTailTotalMs += transactionFinishedAt - transactionCallbackFinishedAt; + }); + }); return trackCommit(seqId, promise); }; const commitIfVersion = async (db: BinaryDatabase, key: Buffer, version: number, action: (version: number) => Promise) => { @@ -231,7 +277,13 @@ export function declareLmdbLowLevelDatabase(options: { path: string, dbId?: stri }; const putWithVersion = async (db: BinaryDatabase, key: Buffer, value: Buffer, version: number) => { activityStats.puts++; - await db.put(key, value, version); + activityStats.putBytes += value.byteLength; + const startedAt = performance.now(); + try { + await db.put(key, value, version); + } finally { + activityStats.putAwaitTotalMs += performance.now() - startedAt; + } }; const waitUntilAvailable = async (seq: DatabaseSeq) => { await getAvailabilityPromise(getSeqId(seq)); @@ -299,15 +351,32 @@ export function declareLmdbLowLevelDatabase(options: { path: string, dbId?: stri const version = nextVersion(); const seqId = nextSeqId(); const keys = values.map((_, index) => dumpKeyForVersion(version, index)); - const promise = traceSpan({ description: "bulldozer-js.low-level.lmdb.insertAll.commit", attributes }, async () => await waitUntilAvailable(insertOptions?.requiresSeq ?? initialSeq).then(async () => await root.transaction(() => { - activityStats.transactions++; - return (async () => { - for (let i = 0; i < values.length; i++) { - await putWithVersion(db, bufferFromArrayBuffer(keys[i]), bufferFromArrayBuffer(values[i]), version); - } - await meta.put("seq", version); - })(); - }))); + const promise = traceSpan({ description: "bulldozer-js.low-level.lmdb.insertAll.commit", attributes }, async () => { + const requiredSeqWaitStartedAt = performance.now(); + await waitUntilAvailable(insertOptions?.requiresSeq ?? initialSeq); + activityStats.requiredSeqWaits++; + activityStats.requiredSeqWaitTotalMs += performance.now() - requiredSeqWaitStartedAt; + const transactionStartedAt = performance.now(); + let transactionCallbackFinishedAt: number | null = null; + return await root.transaction(() => { + activityStats.transactionQueueWaitTotalMs += performance.now() - transactionStartedAt; + activityStats.transactions++; + return (async () => { + const actionStartedAt = performance.now(); + await Promise.all(values.map(async (value, index) => await putWithVersion(db, bufferFromArrayBuffer(keys[index]), bufferFromArrayBuffer(value), version))); + activityStats.transactionActionTotalMs += performance.now() - actionStartedAt; + const metaPutStartedAt = performance.now(); + await meta.put("seq", version); + activityStats.metaPutTotalMs += performance.now() - metaPutStartedAt; + })().finally(() => { + transactionCallbackFinishedAt = performance.now(); + }); + }).finally(() => { + const transactionFinishedAt = performance.now(); + activityStats.transactionTotalMs += transactionFinishedAt - transactionStartedAt; + if (transactionCallbackFinishedAt !== null) activityStats.transactionCommitTailTotalMs += transactionFinishedAt - transactionCallbackFinishedAt; + }); + }); return { keys, seq: trackCommit(seqId, promise) }; }); }, @@ -384,6 +453,7 @@ export function declareLmdbLowLevelDatabase(options: { path: string, dbId?: stri }, combineSeqs(...seqs) { if (seqs.length === 0) return initialSeq; + if (seqs.length === 1) return seqs[0]; const seqId = nextSeqId(); rememberCombinedAvailability(seqId, Promise.all(seqs.map(seq => getAvailabilityPromise(getSeqId(seq))))); rememberCombinedDurability(seqId, Promise.all(seqs.map(seq => getDurabilityPromise(getSeqId(seq))))); From 0bdb102378d1658ac5c68f002bcd527fd6d0e308 Mon Sep 17 00:00:00 2001 From: Konstantin Wohlwend Date: Fri, 3 Jul 2026 14:23:55 -0700 Subject: [PATCH 2/5] Improve Bulldozer logging --- .../src/databases/bulldozer/index.ts | 48 ++- .../data-structures/augmented-tree-map.ts | 49 ++- .../src/databases/piledriver/index.ts | 407 +++++++++++++++++- 3 files changed, 463 insertions(+), 41 deletions(-) diff --git a/apps/bulldozer-js/src/databases/bulldozer/index.ts b/apps/bulldozer-js/src/databases/bulldozer/index.ts index 022e75152..2deebc806 100644 --- a/apps/bulldozer-js/src/databases/bulldozer/index.ts +++ b/apps/bulldozer-js/src/databases/bulldozer/index.ts @@ -887,15 +887,59 @@ export function declareBulldozerDatabase(piledriverDatabase: PiledriverDatabase, options: { replicated: boolean }, ) => { return await traceSpan({ description: "bulldozer-js.bulldozer.withSnapshot", attributes: { "bulldozer.replicated": options.replicated } }, async () => { + const startedAt = performance.now(); + let writeLockWaitMs = 0; + let getSnapshotMs = 0; + let updateSnapshotMs = 0; + let toPiledriverObjectMs = 0; + let setRootMs = 0; + let waitUntilAvailableMs = 0; + let waitUntilReplicatedMs = 0; + let mutationDebugInfo: BulldozerSnapshotMutationDebugInfo | undefined; + const writeLockWaitStartedAt = performance.now(); const result = await withWriteLock(async () => { + writeLockWaitMs = performance.now() - writeLockWaitStartedAt; + const getSnapshotStartedAt = performance.now(); const { snapshot } = await getSnapshot(); + getSnapshotMs = performance.now() - getSnapshotStartedAt; + const updateSnapshotStartedAt = performance.now(); const updateResult = await updateSnapshot(snapshot); + updateSnapshotMs = performance.now() - updateSnapshotStartedAt; + mutationDebugInfo = updateResult instanceof BulldozerDatabaseSnapshot ? undefined : updateResult.debugInfo; const newSnapshot = updateResult instanceof BulldozerDatabaseSnapshot ? updateResult : updateResult.newSnapshot; - const { seq } = await setRoot({ snapshot: newSnapshot.toPiledriverObject() }); + const toPiledriverObjectStartedAt = performance.now(); + const newSnapshotPiledriverObject = newSnapshot.toPiledriverObject(); + toPiledriverObjectMs = performance.now() - toPiledriverObjectStartedAt; + const setRootStartedAt = performance.now(); + const { seq } = await setRoot({ snapshot: newSnapshotPiledriverObject }); + setRootMs = performance.now() - setRootStartedAt; + const waitUntilAvailableStartedAt = performance.now(); await piledriverDatabase.waitUntilAvailable(seq); + waitUntilAvailableMs = performance.now() - waitUntilAvailableStartedAt; return { snapshot: newSnapshot, seq }; }); - if (options.replicated) await piledriverDatabase.waitUntilReplicated(result.seq); + if (options.replicated) { + const waitUntilReplicatedStartedAt = performance.now(); + await piledriverDatabase.waitUntilReplicated(result.seq); + waitUntilReplicatedMs = performance.now() - waitUntilReplicatedStartedAt; + } + console.debug("bulldozer-js withSnapshot timing", inspect({ + replicated: options.replicated, + elapsedMs: performance.now() - startedAt, + writeLockWaitMs, + getSnapshotMs, + updateSnapshotMs, + toPiledriverObjectMs, + setRootMs, + waitUntilAvailableMs, + waitUntilReplicatedMs, + mutation: mutationDebugInfo === undefined ? undefined : { + operation: mutationDebugInfo.operation, + sourceTableId: mutationDebugInfo.sourceTableId, + rowsSetOrDeleted: mutationDebugInfo.rowsSetOrDeleted, + durationMs: mutationDebugInfo.durationMs, + }, + }, { depth: null, maxArrayLength: null })); return result; }); }; diff --git a/apps/bulldozer-js/src/databases/piledriver/data-structures/augmented-tree-map.ts b/apps/bulldozer-js/src/databases/piledriver/data-structures/augmented-tree-map.ts index 0fa07cfb9..dd37a2659 100644 --- a/apps/bulldozer-js/src/databases/piledriver/data-structures/augmented-tree-map.ts +++ b/apps/bulldozer-js/src/databases/piledriver/data-structures/augmented-tree-map.ts @@ -26,10 +26,11 @@ const noAugmentation = Symbol("no-augmentation"); type NoAugmentation = typeof noAugmentation; type Entry = [K, V]; type Child = { ref: PiledriverHeapObject, augmentation: A, size: number, entryCount: number, minKey: K, maxKey: K }; -type Split = { entry: Entry, right: Child }; +type StoredValue = V | PiledriverHeapObject; +type Split = { entry: Entry>, right: Child }; // Persisted B-tree node. Children are heap objects, so unchanged subtrees are shared across versions. -type Node = { - entries: Entry[], +type Node = { + entries: Entry>[], children: Child[], augmentation: A, size: number, @@ -176,7 +177,7 @@ export class AugmentedTreeMultiMap): Promise { for (let node = await this.node(this.root?.ref ?? null); node;) { const { index, found } = this.search(node.entries, key); - if (found) return node.entries[index][1]; + if (found) return await this.loadValue(node.entries[index][1]); node = await this.node(node.children[index]?.ref ?? null); } } @@ -206,8 +207,8 @@ export class AugmentedTreeMultiMap= options.limit) || !tree.overlaps(child, options)) return; const node = (await tree.node(child.ref))!; - const visitEntry = function* (entry: Entry, Value>): Iterable, Value>> { - if (isInRange(entry[0]) && (options.limit === undefined || yielded++ < options.limit)) yield entry; + const visitEntry = async function* (entry: Entry, StoredValue>): AsyncIterable, Value>> { + if (isInRange(entry[0]) && (options.limit === undefined || yielded++ < options.limit)) yield [entry[0], await tree.loadValue(entry[1])]; }; if (options.reverse) { @@ -246,6 +247,14 @@ export class AugmentedTreeMultiMap, Value, Augmentation> : null; } + private storeValue(value: Value): StoredValue { + return value !== null && typeof value === "object" ? asHeapObject(value) : value; + } + + private async loadValue(value: StoredValue): Promise { + return value !== null && typeof value === "object" && isPiledriverHeapObjectSymbol in value ? await value.get() as Value : value; + } + private async empty() { return this.options.initialAugmentation !== undefined ? this.options.initialAugmentation : await this.options.mergeAugmentations(); } @@ -268,7 +277,7 @@ export class AugmentedTreeMultiMap, Value>[], children: Child, Augmentation>[] = []) { + private async make(entries: Entry, StoredValue>[], children: Child, Augmentation>[] = []) { if (!entries.length && children.length !== 1) throw new Error("Invalid empty B-tree node"); if (!entries.length) { const [onlyChild] = children; @@ -292,7 +301,7 @@ export class AugmentedTreeMultiMap, Value>[], key: MultiKey) { + private search(entries: Entry, StoredValue>[], key: MultiKey) { let low = 0, high = entries.length; while (low < high) { const mid = (low + high) >> 1; @@ -337,21 +346,21 @@ export class AugmentedTreeMultiMap, Value>[], children: Child, Augmentation>[]) { + private async split(entries: Entry, StoredValue>[], children: Child, Augmentation>[]) { const middle = entries.length >> 1; const left = await this.make(entries.slice(0, middle), children.length ? children.slice(0, middle + 1) : []); const right = await this.make(entries.slice(middle + 1), children.length ? children.slice(middle + 1) : []); return { root: left, split: { entry: entries[middle], right } }; } - private async done(entries: Entry, Value>[], children: Child, Augmentation>[], added: boolean) { + private async done(entries: Entry, StoredValue>[], children: Child, Augmentation>[], added: boolean) { const result = entries.length > this.maxEntries() ? await this.split(entries, children) : { root: await this.make(entries, children) }; return { ...result, added }; } private async upsert(heapObject: PiledriverHeapObject | null, key: MultiKey, value: Value, replace: boolean): Promise<{ root: Child, Augmentation>, split?: Split, Value, Augmentation>, added: boolean }> { const node = await this.node(heapObject); - if (!node) return { root: await this.make([[key, value]]), added: true }; + if (!node) return { root: await this.make([[key, this.storeValue(value)]]), added: true }; const { index, found } = this.search(node.entries, key); const entries = [...node.entries]; @@ -359,12 +368,12 @@ export class AugmentedTreeMultiMap, Augmentation>): Promise<{ root: Child, Augmentation> | null, entry: Entry, Value> }> { + private async deleteMin(child: Child, Augmentation>): Promise<{ root: Child, Augmentation> | null, entry: Entry, StoredValue> }> { const node = (await this.node(child.ref))!; const entries = [...node.entries]; const children = [...node.children] as (Child, Augmentation> | null)[]; @@ -432,7 +441,7 @@ export class AugmentedTreeMultiMap, Augmentation>): Promise<{ root: Child, Augmentation> | null, entry: Entry, Value> }> { + private async deleteMax(child: Child, Augmentation>): Promise<{ root: Child, Augmentation> | null, entry: Entry, StoredValue> }> { const node = (await this.node(child.ref))!; const entries = [...node.entries]; const children = [...node.children] as (Child, Augmentation> | null)[]; @@ -448,13 +457,13 @@ export class AugmentedTreeMultiMap, Value>[], children: (Child, Augmentation> | null)[], isRoot: boolean, deleted: boolean) { + private async afterDelete(entries: Entry, StoredValue>[], children: (Child, Augmentation> | null)[], isRoot: boolean, deleted: boolean) { const liveChildren = children.filter(child => child !== null); if (!entries.length) return { root: isRoot ? liveChildren[0] ?? null : liveChildren[0] ? await this.make([], [liveChildren[0]]) : null, deleted }; return { root: await this.make(entries, liveChildren), deleted }; } - private async fixChild(entries: Entry, Value>[], children: (Child, Augmentation> | null)[], index: number, isRoot: boolean): Promise<{ root: Child, Augmentation> | null, deleted: boolean }> { + private async fixChild(entries: Entry, StoredValue>[], children: (Child, Augmentation> | null)[], index: number, isRoot: boolean): Promise<{ root: Child, Augmentation> | null, deleted: boolean }> { const child = children[index]; if (child && child.entryCount >= this.minEntries()) return await this.afterDelete(entries, children, isRoot, true); @@ -491,7 +500,7 @@ export class AugmentedTreeMultiMap, Augmentation> | null, entry: Entry, Value>, right: Child, Augmentation> | null) { + private async mergeChildren(left: Child, Augmentation> | null, entry: Entry, StoredValue>, right: Child, Augmentation> | null) { const leftNode = left ? (await this.node(left.ref))! : { entries: [], children: [] }; const rightNode = right ? (await this.node(right.ref))! : { entries: [], children: [] }; return await this.make( @@ -518,7 +527,7 @@ export class AugmentedTreeMultiMap, - heapObjects: ReadonlySet, + objects: Set, + heapObjects: Set, }; -const emptySerializationPath: SerializationPath = { objects: new Set(), heapObjects: new Set() }; +const emptySerializationPath = (): SerializationPath => ({ objects: new Set(), heapObjects: new Set() }); +type PendingHeapInsert = { + buffer: ArrayBuffer, + requiresSeq: DatabaseSeq, + serializationTimingStats: PiledriverSerializationTimingStats | undefined, + resolve: (value: { key: ArrayBuffer, seq: DatabaseSeq }) => void, + reject: (error: unknown) => void, +}; +type PiledriverSerializationTimingStats = { + primitiveNodes: number, + arrayNodes: number, + arrayItems: number, + objectNodes: number, + objectEntries: number, + heapReferenceNodes: number, + serializeToJsonableTotalMs: number, + jsonStringifyTotalMs: number, + textEncodeTotalMs: number, + heapObjectCacheHits: number, + heapObjectCacheMisses: number, + heapObjectCacheHitAwaitTotalMs: number, + heapObjectCacheMissAwaitTotalMs: number, + heapObjectGetTotalMs: number, + heapObjectSerializeTotalMs: number, + heapObjectInsertAwaitTotalMs: number, + heapInsertFlushes: number, + heapInsertValues: number, + heapInsertAllTotalMs: number, + branchStats: Map, + heapObjectCacheMissesByShape: Map, + heapObjectCacheMissInlineNodeCountsByShape: Map, +}; +type PiledriverSerializationBranchStats = { + primitiveNodes: number, + arrayNodes: number, + objectNodes: number, + objectEntries: number, + heapReferenceNodes: number, + heapObjectCacheHits: number, + heapObjectCacheMisses: number, + heapObjectCacheMissesByShape: Map, + heapObjectCacheMissInlineNodeCountsByShape: Map, +}; +type PiledriverInlineNodeCounts = { + primitiveNodes: number, + arrayNodes: number, + objectNodes: number, + heapReferenceNodes: number, +}; + +function emptyPiledriverSerializationBranchStats(): PiledriverSerializationBranchStats { + return { + primitiveNodes: 0, + arrayNodes: 0, + objectNodes: 0, + objectEntries: 0, + heapReferenceNodes: 0, + heapObjectCacheHits: 0, + heapObjectCacheMisses: 0, + heapObjectCacheMissesByShape: new Map(), + heapObjectCacheMissInlineNodeCountsByShape: new Map(), + }; +} + +function emptyPiledriverSerializationTimingStats(): PiledriverSerializationTimingStats { + return { + primitiveNodes: 0, + arrayNodes: 0, + arrayItems: 0, + objectNodes: 0, + objectEntries: 0, + heapReferenceNodes: 0, + serializeToJsonableTotalMs: 0, + jsonStringifyTotalMs: 0, + textEncodeTotalMs: 0, + heapObjectCacheHits: 0, + heapObjectCacheMisses: 0, + heapObjectCacheHitAwaitTotalMs: 0, + heapObjectCacheMissAwaitTotalMs: 0, + heapObjectGetTotalMs: 0, + heapObjectSerializeTotalMs: 0, + heapObjectInsertAwaitTotalMs: 0, + heapInsertFlushes: 0, + heapInsertValues: 0, + heapInsertAllTotalMs: 0, + branchStats: new Map(), + heapObjectCacheMissesByShape: new Map(), + heapObjectCacheMissInlineNodeCountsByShape: new Map(), + }; +} + +function incrementMapCount(map: Map, key: string) { + map.set(key, (map.get(key) ?? 0) + 1); +} + +function addMapCount(map: Map, key: string, value: number) { + map.set(key, (map.get(key) ?? 0) + value); +} + +function emptyPiledriverInlineNodeCounts(): PiledriverInlineNodeCounts { + return { + primitiveNodes: 0, + arrayNodes: 0, + objectNodes: 0, + heapReferenceNodes: 0, + }; +} + +function addPiledriverInlineNodeCounts(target: PiledriverInlineNodeCounts, source: PiledriverInlineNodeCounts) { + target.primitiveNodes += source.primitiveNodes; + target.arrayNodes += source.arrayNodes; + target.objectNodes += source.objectNodes; + target.heapReferenceNodes += source.heapReferenceNodes; +} + +function totalInlineNodes(counts: PiledriverInlineNodeCounts) { + return counts.primitiveNodes + counts.arrayNodes + counts.objectNodes + counts.heapReferenceNodes; +} + +function countPiledriverInlineNodes(obj: PiledriverObject, path: Set = new Set()): PiledriverInlineNodeCounts { + switch (typeof obj) { + case "number": + case "string": + case "boolean": { + return { ...emptyPiledriverInlineNodeCounts(), primitiveNodes: 1 }; + } + case "object": { + if (obj === null) return { ...emptyPiledriverInlineNodeCounts(), primitiveNodes: 1 }; + if (isPiledriverHeapObjectSymbol in obj) return { ...emptyPiledriverInlineNodeCounts(), heapReferenceNodes: 1 }; + if (path.has(obj)) throw new Error("Piledriver objects must not contain cycles"); + const childPath = new Set(path).add(obj); + const counts = emptyPiledriverInlineNodeCounts(); + if (Array.isArray(obj)) { + counts.arrayNodes++; + for (const item of obj) addPiledriverInlineNodeCounts(counts, countPiledriverInlineNodes(item, childPath)); + } else { + counts.objectNodes++; + for (const value of Object.values(obj)) addPiledriverInlineNodeCounts(counts, countPiledriverInlineNodes(value, childPath)); + } + return counts; + } + default: { + throw new Error("Assertion error: Unknown type of Piledriver object " + typeof obj); + } + } +} + +function classifyHeapObjectPayload(obj: PiledriverObject): string { + if (obj === null) return "null"; + if (Array.isArray(obj)) return `array:${obj.length}`; + if (typeof obj !== "object") return typeof obj; + const keys = Object.keys(obj).sort(); + if (keys.includes("entries") && keys.includes("children") && keys.includes("augmentation") && keys.includes("size") && keys.includes("minKey") && keys.includes("maxKey")) { + const entries = Reflect.get(obj, "entries"); + const children = Reflect.get(obj, "children"); + return `btree-node:entries=${Array.isArray(entries) ? entries.length : "?"}:children=${Array.isArray(children) ? children.length : "?"}`; + } + if (keys.includes("key") && keys.includes("id") && keys.length === 2) return "multi-key"; + if (keys.includes("groupKey") && keys.includes("rows") && keys.length === 2) return "group-with-rows"; + if (keys.includes("inputRowData") && keys.includes("outputRowData") && keys.includes("stateAfter")) return "left-fold-row"; + if (keys.includes("state") && keys.includes("nextTriggerTimeMs") && keys.includes("emittedRows")) return "time-fold-row"; + if (keys.includes("outputRows") && keys.length === 1) return "time-fold-group"; + if (keys.includes("rowData") && keys.includes("rowIdentifier") && keys.includes("rowSortKey") && keys.includes("groupKey")) return "row-object"; + return `object:${keys.slice(0, 6).join(",")}${keys.length > 6 ? ",..." : ""}`; +} + +function serializationBranchKey(path: readonly string[]): string { + if (path[0] === "snapshot" && path[1] === "serializedTables") return `table:${path[2]}`; + return path[0] ?? ""; +} + +function serializationBranchStats(stats: PiledriverSerializationTimingStats | undefined, path: readonly string[]) { + if (stats === undefined) return undefined; + const key = serializationBranchKey(path); + let result = stats.branchStats.get(key); + if (result === undefined) { + result = emptyPiledriverSerializationBranchStats(); + stats.branchStats.set(key, result); + } + return result; +} export function declarePiledriverDatabase(lowLevelDb: LowLevelDatabase, options: PiledriverDatabaseOptions = {}): PiledriverDatabase { // TODO actually support cycles both for heap and non-heap objects (right now they are detected and rejected) @@ -85,6 +266,8 @@ export function declarePiledriverDatabase(lowLevelDb: LowLevelDatabase, options: const heapObjectsByHeapKeyBase64 = new Map, seq: DatabaseSeq }>(); const heapObjectsByHeapKeyFinalizer = new FinalizationRegistry(([keyBase64, refIdentity]: [string, string]) => heapObjectsByHeapKeyBase64.get(keyBase64)?.refIdentity === refIdentity && heapObjectsByHeapKeyBase64.delete(keyBase64)); const heapKeysAndSeqByHeapObjects = new WeakMap>(); + let pendingHeapInserts: PendingHeapInsert[] = []; + let heapInsertFlushScheduled = false; const cacheHeapObjectByKey = (key: ArrayBuffer, heapObj: PiledriverHeapObject, seq: DatabaseSeq) => { const keyBase64 = encodeBase64(new Uint8Array(key)); @@ -92,36 +275,112 @@ export function declarePiledriverDatabase(lowLevelDb: LowLevelDatabase, options: heapObjectsByHeapKeyBase64.set(keyBase64, { refIdentity, object: new WeakRef(heapObj), seq }); heapObjectsByHeapKeyFinalizer.register(heapObj, [keyBase64, refIdentity]); }; + const flushPendingHeapInserts = async () => { + const batch = pendingHeapInserts; + pendingHeapInserts = []; + heapInsertFlushScheduled = false; + if (batch.length === 0) return; - const getHeapKeyAndSeq = async (heapObj: PiledriverHeapObject, path: SerializationPath = emptySerializationPath): Promise<{ key: ArrayBuffer, seq: DatabaseSeq }> => { + try { + const entriesBySerializationTimingStats = new Map(); + for (const entry of batch) { + if (entry.serializationTimingStats !== undefined) { + entriesBySerializationTimingStats.set(entry.serializationTimingStats, (entriesBySerializationTimingStats.get(entry.serializationTimingStats) ?? 0) + 1); + } + } + const heapInsertAllStartedAt = performance.now(); + const inserted = await heapDump.insertAll(batch.map(entry => entry.buffer)); + const heapInsertAllMs = performance.now() - heapInsertAllStartedAt; + for (const [stats, entryCount] of entriesBySerializationTimingStats) { + stats.heapInsertFlushes++; + stats.heapInsertValues += entryCount; + stats.heapInsertAllTotalMs += heapInsertAllMs * entryCount / batch.length; + } + for (let i = 0; i < batch.length; i++) { + batch[i].resolve({ key: inserted.keys[i], seq: lowLevelDb.combineSeqs(batch[i].requiresSeq, inserted.seq) }); + } + } catch (error) { + for (const entry of batch) entry.reject(error); + } + }; + const insertHeapObjectBatched = async (buffer: ArrayBuffer, requiresSeq: DatabaseSeq, serializationTimingStats: PiledriverSerializationTimingStats | undefined): Promise<{ key: ArrayBuffer, seq: DatabaseSeq }> => { + return await new Promise((resolve, reject) => { + pendingHeapInserts.push({ buffer, requiresSeq, serializationTimingStats, resolve, reject }); + if (!heapInsertFlushScheduled) { + heapInsertFlushScheduled = true; + const timeout = setTimeout(() => { + runAsynchronously(async () => await flushPendingHeapInserts()); + }, 0); + timeout.unref(); + } + }); + }; + + const getHeapKeyAndSeq = async (heapObj: PiledriverHeapObject, path: SerializationPath = emptySerializationPath(), serializationTimingStats?: PiledriverSerializationTimingStats, logicalPath: readonly string[] = []): Promise<{ key: ArrayBuffer, seq: DatabaseSeq }> => { // Must be checked before the memo lookup: awaiting the memoized promise of an ancestor // that is still being serialized would deadlock. if (path.heapObjects.has(heapObj)) throw new Error("Piledriver objects must not contain cycles (found a cycle of heap objects)"); const existing = heapKeysAndSeqByHeapObjects.get(heapObj); - if (existing) return await existing; + if (existing) { + if (serializationTimingStats !== undefined) { + serializationTimingStats.heapObjectCacheHits++; + const branch = serializationBranchStats(serializationTimingStats, logicalPath); + if (branch !== undefined) branch.heapObjectCacheHits++; + } + const cacheHitAwaitStartedAt = performance.now(); + try { + return await existing; + } finally { + if (serializationTimingStats !== undefined) serializationTimingStats.heapObjectCacheHitAwaitTotalMs += performance.now() - cacheHitAwaitStartedAt; + } + } + if (serializationTimingStats !== undefined) { + serializationTimingStats.heapObjectCacheMisses++; + const branch = serializationBranchStats(serializationTimingStats, logicalPath); + if (branch !== undefined) branch.heapObjectCacheMisses++; + } const promise = (async () => { // A heap object starts a fresh plain-object path; plain object cycles can't span heap // boundaries without also forming a heap cycle, which is tracked separately. const childPath: SerializationPath = { objects: new Set(), heapObjects: new Set(path.heapObjects).add(heapObj) }; return await traceSpan("bulldozer-js.piledriver.heap.serializeAndInsert", async () => { - const serialized = await serializePiledriverObject(await heapObj.get(), childPath); - const inserted = await heapDump.insertAll([serialized.buffer]); - return { - key: inserted.keys[0], - seq: lowLevelDb.combineSeqs(serialized.seq, inserted.seq), - }; + const heapObjectGetStartedAt = performance.now(); + const heapObject = await heapObj.get(); + let shape: string | undefined; + if (serializationTimingStats !== undefined) { + shape = classifyHeapObjectPayload(heapObject); + const inlineNodeCount = totalInlineNodes(countPiledriverInlineNodes(heapObject)); + incrementMapCount(serializationTimingStats.heapObjectCacheMissesByShape, shape); + addMapCount(serializationTimingStats.heapObjectCacheMissInlineNodeCountsByShape, shape, inlineNodeCount); + const branch = serializationBranchStats(serializationTimingStats, logicalPath); + if (branch !== undefined) { + incrementMapCount(branch.heapObjectCacheMissesByShape, shape); + addMapCount(branch.heapObjectCacheMissInlineNodeCountsByShape, shape, inlineNodeCount); + } + } + if (serializationTimingStats !== undefined) serializationTimingStats.heapObjectGetTotalMs += performance.now() - heapObjectGetStartedAt; + const heapObjectSerializeStartedAt = performance.now(); + const serialized = await serializePiledriverObject(heapObject, childPath, serializationTimingStats, logicalPath); + if (serializationTimingStats !== undefined) serializationTimingStats.heapObjectSerializeTotalMs += performance.now() - heapObjectSerializeStartedAt; + const heapObjectInsertStartedAt = performance.now(); + const result = await insertHeapObjectBatched(serialized.buffer, serialized.seq, serializationTimingStats); + if (serializationTimingStats !== undefined) serializationTimingStats.heapObjectInsertAwaitTotalMs += performance.now() - heapObjectInsertStartedAt; + return result; }); })(); heapKeysAndSeqByHeapObjects.set(heapObj, promise); let result; + const cacheMissAwaitStartedAt = performance.now(); try { result = await promise; } catch (error) { // Don't leave a poisoned rejected promise in the cache; a later retry may succeed. if (heapKeysAndSeqByHeapObjects.get(heapObj) === promise) heapKeysAndSeqByHeapObjects.delete(heapObj); throw error; + } finally { + if (serializationTimingStats !== undefined) serializationTimingStats.heapObjectCacheMissAwaitTotalMs += performance.now() - cacheMissAwaitStartedAt; } cacheHeapObjectByKey(result.key, heapObj, result.seq); return result; @@ -167,30 +426,56 @@ export function declarePiledriverDatabase(lowLevelDb: LowLevelDatabase, options: return { object: heapObj, seq }; }; - const serializePiledriverObjectToJsonableObject = async (obj: PiledriverObject, path: SerializationPath): Promise<{ jsonableObject: unknown, seq: DatabaseSeq }> => { + const serializePiledriverObjectToJsonableObject = async (obj: PiledriverObject, path: SerializationPath, serializationTimingStats?: PiledriverSerializationTimingStats, logicalPath: readonly string[] = []): Promise<{ jsonableObject: unknown, seq: DatabaseSeq }> => { switch (typeof obj) { case "number": { + if (serializationTimingStats !== undefined) { + serializationTimingStats.primitiveNodes++; + const branch = serializationBranchStats(serializationTimingStats, logicalPath); + if (branch !== undefined) branch.primitiveNodes++; + } if (!Number.isFinite(obj)) return { jsonableObject: [obj.toString()], seq: lowLevelDb.initialSeq }; if (Object.is(obj, -0)) return { jsonableObject: ["-0"], seq: lowLevelDb.initialSeq }; // intentionally fall through to the primitive case below } case "string": case "boolean": { + if (serializationTimingStats !== undefined && typeof obj !== "number") { + serializationTimingStats.primitiveNodes++; + const branch = serializationBranchStats(serializationTimingStats, logicalPath); + if (branch !== undefined) branch.primitiveNodes++; + } return { jsonableObject: obj, seq: lowLevelDb.initialSeq }; } case "object": { if (obj === null) { + if (serializationTimingStats !== undefined) { + serializationTimingStats.primitiveNodes++; + const branch = serializationBranchStats(serializationTimingStats, logicalPath); + if (branch !== undefined) branch.primitiveNodes++; + } return { jsonableObject: obj, seq: lowLevelDb.initialSeq }; } else if (Array.isArray(obj)) { + if (serializationTimingStats !== undefined) { + serializationTimingStats.arrayNodes++; + serializationTimingStats.arrayItems += obj.length; + const branch = serializationBranchStats(serializationTimingStats, logicalPath); + if (branch !== undefined) branch.arrayNodes++; + } if (path.objects.has(obj)) throw new Error("Piledriver objects must not contain cycles"); const childPath: SerializationPath = { ...path, objects: new Set(path.objects).add(obj) }; - const itemsSerializeResults = await Promise.all(obj.map(async o => await serializePiledriverObjectToJsonableObject(o, childPath))); + const itemsSerializeResults = await Promise.all(obj.map(async o => await serializePiledriverObjectToJsonableObject(o, childPath, serializationTimingStats, logicalPath))); return { jsonableObject: ["array", itemsSerializeResults.map(r => r.jsonableObject)], seq: lowLevelDb.combineSeqs(...itemsSerializeResults.map(r => r.seq)), }; } else if (isPiledriverHeapObjectSymbol in obj) { - const heapKeyAndSeq = await getHeapKeyAndSeq(obj, path); + if (serializationTimingStats !== undefined) { + serializationTimingStats.heapReferenceNodes++; + const branch = serializationBranchStats(serializationTimingStats, logicalPath); + if (branch !== undefined) branch.heapReferenceNodes++; + } + const heapKeyAndSeq = await getHeapKeyAndSeq(obj, path, serializationTimingStats, logicalPath); return { jsonableObject: ["heap-reference", encodeBase64(new Uint8Array(heapKeyAndSeq.key))], seq: heapKeyAndSeq.seq, @@ -201,7 +486,17 @@ export function declarePiledriverDatabase(lowLevelDb: LowLevelDatabase, options: if (path.objects.has(obj)) throw new Error("Piledriver objects must not contain cycles"); const childPath: SerializationPath = { ...path, objects: new Set(path.objects).add(obj) }; - const entriesSerializeResults = await Promise.all(Object.entries(obj).map(async ([k, v]) => [k, await serializePiledriverObjectToJsonableObject(v, childPath)] as const)); + const entries = Object.entries(obj); + if (serializationTimingStats !== undefined) { + serializationTimingStats.objectNodes++; + serializationTimingStats.objectEntries += entries.length; + const branch = serializationBranchStats(serializationTimingStats, logicalPath); + if (branch !== undefined) { + branch.objectNodes++; + branch.objectEntries += entries.length; + } + } + const entriesSerializeResults = await Promise.all(entries.map(async ([k, v]) => [k, await serializePiledriverObjectToJsonableObject(v, childPath, serializationTimingStats, [...logicalPath, k])] as const)); return { jsonableObject: Object.fromEntries(entriesSerializeResults.map(([k, v]) => [k, v.jsonableObject] as const)), seq: lowLevelDb.combineSeqs(...entriesSerializeResults.map(([_, v]) => v.seq)), @@ -214,10 +509,18 @@ export function declarePiledriverDatabase(lowLevelDb: LowLevelDatabase, options: } }; - const serializePiledriverObject = async (obj: PiledriverObject, path: SerializationPath = emptySerializationPath): Promise<{ buffer: ArrayBuffer, seq: DatabaseSeq }> => { - const toJsonableResponse = await serializePiledriverObjectToJsonableObject(obj, path); + const serializePiledriverObject = async (obj: PiledriverObject, path: SerializationPath = emptySerializationPath(), serializationTimingStats?: PiledriverSerializationTimingStats, logicalPath: readonly string[] = []): Promise<{ buffer: ArrayBuffer, seq: DatabaseSeq }> => { + const toJsonableStartedAt = performance.now(); + const toJsonableResponse = await serializePiledriverObjectToJsonableObject(obj, path, serializationTimingStats, logicalPath); + if (serializationTimingStats !== undefined) serializationTimingStats.serializeToJsonableTotalMs += performance.now() - toJsonableStartedAt; + const jsonStringifyStartedAt = performance.now(); + const json = JSON.stringify(toJsonableResponse.jsonableObject); + if (serializationTimingStats !== undefined) serializationTimingStats.jsonStringifyTotalMs += performance.now() - jsonStringifyStartedAt; + const textEncodeStartedAt = performance.now(); + const buffer = new TextEncoder().encode(json).buffer; + if (serializationTimingStats !== undefined) serializationTimingStats.textEncodeTotalMs += performance.now() - textEncodeStartedAt; return { - buffer: new TextEncoder().encode(JSON.stringify(toJsonableResponse.jsonableObject)).buffer, + buffer, seq: toJsonableResponse.seq, }; }; @@ -311,8 +614,74 @@ export function declarePiledriverDatabase(lowLevelDb: LowLevelDatabase, options: }, async setRootObject(key, value): Promise<{ seq: DatabaseSeq }> { return await traceSpan("bulldozer-js.piledriver.setRootObject", async () => { - const { buffer, seq } = await serializePiledriverObject(value); + const timingStats = emptyPiledriverSerializationTimingStats(); + const startedAt = performance.now(); + const serializeStartedAt = performance.now(); + const serializeCpuStartedAt = process.cpuUsage(); + const { buffer, seq } = await serializePiledriverObject(value, emptySerializationPath(), timingStats); + const serializeCpuUsage = process.cpuUsage(serializeCpuStartedAt); + const serializePiledriverObjectMs = performance.now() - serializeStartedAt; + const serializeCpuMs = (serializeCpuUsage.user + serializeCpuUsage.system) / 1000; + const rootStoreSetAllStartedAt = performance.now(); const { seq: rootSeq } = await rootStore.setAll([{ key, value: buffer }], { requiresSeq: seq }); + const rootStoreSetAllMs = performance.now() - rootStoreSetAllStartedAt; + const topSerializationBranches = [...timingStats.branchStats.entries()] + .map(([branch, stats]) => ({ + branch, + totalNodes: stats.primitiveNodes + stats.arrayNodes + stats.objectNodes + stats.heapReferenceNodes, + primitiveNodes: stats.primitiveNodes, + arrayNodes: stats.arrayNodes, + objectNodes: stats.objectNodes, + objectEntries: stats.objectEntries, + heapReferenceNodes: stats.heapReferenceNodes, + heapObjectCacheHits: stats.heapObjectCacheHits, + heapObjectCacheMisses: stats.heapObjectCacheMisses, + topHeapObjectCacheMissShapes: [...stats.heapObjectCacheMissesByShape.entries()] + .map(([shape, count]) => { + const totalInlineNodeCount = stats.heapObjectCacheMissInlineNodeCountsByShape.get(shape) ?? 0; + return { shape, count, totalInlineNodeCount, averageInlineNodeCount: count === 0 ? 0 : totalInlineNodeCount / count }; + }) + .sort((a, b) => b.count - a.count) + .slice(0, 5), + })) + .sort((a, b) => b.totalNodes - a.totalNodes) + .slice(0, 5); + const topHeapObjectCacheMissShapes = [...timingStats.heapObjectCacheMissesByShape.entries()] + .map(([shape, count]) => { + const totalInlineNodeCount = timingStats.heapObjectCacheMissInlineNodeCountsByShape.get(shape) ?? 0; + return { shape, count, totalInlineNodeCount, averageInlineNodeCount: count === 0 ? 0 : totalInlineNodeCount / count }; + }) + .sort((a, b) => b.count - a.count) + .slice(0, 10); + console.debug("bulldozer-js piledriver setRootObject timing", { + elapsedMs: performance.now() - startedAt, + serializePiledriverObjectMs, + serializeCpuMs, + serializeCpuToWallRatio: serializePiledriverObjectMs === 0 ? 0 : serializeCpuMs / serializePiledriverObjectMs, + rootStoreSetAllMs, + rootValueBytes: buffer.byteLength, + primitiveNodes: timingStats.primitiveNodes, + arrayNodes: timingStats.arrayNodes, + arrayItems: timingStats.arrayItems, + objectNodes: timingStats.objectNodes, + objectEntries: timingStats.objectEntries, + heapReferenceNodes: timingStats.heapReferenceNodes, + serializeToJsonableTotalMs: timingStats.serializeToJsonableTotalMs, + jsonStringifyTotalMs: timingStats.jsonStringifyTotalMs, + textEncodeTotalMs: timingStats.textEncodeTotalMs, + heapObjectCacheHits: timingStats.heapObjectCacheHits, + heapObjectCacheMisses: timingStats.heapObjectCacheMisses, + heapObjectCacheHitAwaitTotalMs: timingStats.heapObjectCacheHitAwaitTotalMs, + heapObjectCacheMissAwaitTotalMs: timingStats.heapObjectCacheMissAwaitTotalMs, + heapObjectGetTotalMs: timingStats.heapObjectGetTotalMs, + heapObjectSerializeTotalMs: timingStats.heapObjectSerializeTotalMs, + heapObjectInsertAwaitTotalMs: timingStats.heapObjectInsertAwaitTotalMs, + heapInsertFlushes: timingStats.heapInsertFlushes, + heapInsertValues: timingStats.heapInsertValues, + heapInsertAllTotalMs: timingStats.heapInsertAllTotalMs, + topHeapObjectCacheMissShapes, + topSerializationBranches, + }); return { seq: rootSeq }; }); }, From eb29c711f4ed740664c6527394c86bbd45eef4a0 Mon Sep 17 00:00:00 2001 From: Konstantin Wohlwend Date: Fri, 3 Jul 2026 14:40:58 -0700 Subject: [PATCH 3/5] Fabulous performance improvements --- .../src/databases/bulldozer/index.ts | 15 +- .../low-level/implementations/in-memory.ts | 22 +- .../implementations/instant-availability.ts | 30 +- .../low-level/implementations/lmdb.ts | 34 +- .../src/databases/piledriver/index.ts | 336 +++++++++++------- apps/bulldozer-js/src/otel.ts | 15 + 6 files changed, 271 insertions(+), 181 deletions(-) diff --git a/apps/bulldozer-js/src/databases/bulldozer/index.ts b/apps/bulldozer-js/src/databases/bulldozer/index.ts index 2deebc806..fb26338f5 100644 --- a/apps/bulldozer-js/src/databases/bulldozer/index.ts +++ b/apps/bulldozer-js/src/databases/bulldozer/index.ts @@ -23,12 +23,23 @@ async function fromAsync(iterable: AsyncIterable): Promise { * piledriverObjectEquals) is needed on top of a compareGroupKeys ordering: two group keys map * to the same string iff they are piledriverObjectEquals-equal. Object keys are sorted so that * key insertion order does not matter. + * + * Memoized by object identity: this function is called from B-tree comparators (twice per key + * comparison, O(log n) comparisons per tree operation), and CPU profiling showed it at ~5% of + * process CPU during backfills. Piledriver objects are immutable, so caching by identity is + * safe; the WeakMap lets group key objects be collected normally. */ +const canonicalGroupKeyStringCache = new WeakMap(); function canonicalGroupKeyString(groupKey: PiledriverObject): string { if (groupKey !== null && typeof groupKey === "object") { + const cached = canonicalGroupKeyStringCache.get(groupKey); + if (cached !== undefined) return cached; if (isPiledriverHeapObjectSymbol in groupKey) throw new Error("Group keys must not contain heap objects"); - if (Array.isArray(groupKey)) return "[" + groupKey.map(canonicalGroupKeyString).join(",") + "]"; - return "{" + Object.entries(groupKey).sort(([a], [b]) => compareStrings(a, b)).map(([k, v]) => JSON.stringify(k) + ":" + canonicalGroupKeyString(v)).join(",") + "}"; + const result = Array.isArray(groupKey) + ? "[" + groupKey.map(canonicalGroupKeyString).join(",") + "]" + : "{" + Object.entries(groupKey).sort(([a], [b]) => compareStrings(a, b)).map(([k, v]) => JSON.stringify(k) + ":" + canonicalGroupKeyString(v)).join(",") + "}"; + canonicalGroupKeyStringCache.set(groupKey, result); + return result; } return JSON.stringify(groupKey); } diff --git a/apps/bulldozer-js/src/databases/low-level/implementations/in-memory.ts b/apps/bulldozer-js/src/databases/low-level/implementations/in-memory.ts index 4a889fffc..6ff04ae12 100644 --- a/apps/bulldozer-js/src/databases/low-level/implementations/in-memory.ts +++ b/apps/bulldozer-js/src/databases/low-level/implementations/in-memory.ts @@ -1,6 +1,6 @@ import { encodeBase64 } from "@hexclave/shared/dist/utils/bytes"; import { stringCompare } from "@hexclave/shared/dist/utils/strings"; -import { traceSpan } from "../../../otel.js"; +import { traceSpanHot } from "../../../otel.js"; import { Database, DatabaseSeq } from "../../index.js"; import { LowLevelDatabase, LowLevelDatabaseDebugEntry, LowLevelKvDump, LowLevelKvStore } from "../index.js"; @@ -47,7 +47,7 @@ export function declareInMemoryLowLevelDatabase(dbId: string): LowLevelDatabase const seqSentinel: DatabaseSeq = [] as unknown as DatabaseSeq; const result: LowLevelKvStore & LowLevelKvDump = { async get(key: ArrayBuffer) { - return await traceSpan({ description: "bulldozer-js.low-level.in-memory.get", attributes }, async () => { + return await traceSpanHot({ description: "bulldozer-js.low-level.in-memory.get", attributes }, async () => { if (key.byteLength > 64) throw new Error("KV store key must be <= 64 bytes"); return { buffer: base64KeyToValue.get(encodeBase64(new Uint8Array(key)))?.slice(0) ?? null, @@ -56,7 +56,7 @@ export function declareInMemoryLowLevelDatabase(dbId: string): LowLevelDatabase }); }, async setAll(entries: Array<{ key: ArrayBuffer, value: ArrayBuffer }>) { - return await traceSpan({ description: "bulldozer-js.low-level.in-memory.setAll", attributes: { ...attributes, "bulldozer.low_level.entry_count": entries.length } }, async () => { + return await traceSpanHot({ description: "bulldozer-js.low-level.in-memory.setAll", attributes: { ...attributes, "bulldozer.low_level.entry_count": entries.length } }, async () => { for (const { key, value } of entries) { if (key.byteLength > 64) throw new Error("KV store key must be <= 64 bytes"); if (value.byteLength > 2_000_000_000) throw new Error("KV store value must be <= 2GB"); @@ -68,7 +68,7 @@ export function declareInMemoryLowLevelDatabase(dbId: string): LowLevelDatabase }); }, async deleteAll(keys: ArrayBuffer[]) { - return await traceSpan({ description: "bulldozer-js.low-level.in-memory.deleteAll", attributes: { ...attributes, "bulldozer.low_level.key_count": keys.length } }, async () => { + return await traceSpanHot({ description: "bulldozer-js.low-level.in-memory.deleteAll", attributes: { ...attributes, "bulldozer.low_level.key_count": keys.length } }, async () => { for (const key of keys) { if (key.byteLength > 64) throw new Error("KV store key must be <= 64 bytes"); base64KeyToValue.delete(encodeBase64(new Uint8Array(key))); @@ -79,7 +79,7 @@ export function declareInMemoryLowLevelDatabase(dbId: string): LowLevelDatabase }); }, async insertAll(values: ArrayBuffer[], options: { requiresSeq: DatabaseSeq }) { - return await traceSpan({ description: "bulldozer-js.low-level.in-memory.insertAll", attributes: { ...attributes, "bulldozer.low_level.value_count": values.length } }, async () => { + return await traceSpanHot({ description: "bulldozer-js.low-level.in-memory.insertAll", attributes: { ...attributes, "bulldozer.low_level.value_count": values.length } }, async () => { for (const value of values) { if (value.byteLength > 2_000_000_000) throw new Error("KV store value must be <= 2GB"); } @@ -91,7 +91,7 @@ export function declareInMemoryLowLevelDatabase(dbId: string): LowLevelDatabase }); }, async compareAndSet(key: ArrayBuffer, compare: ArrayBuffer, value: ArrayBuffer, options: { requiresSeq: DatabaseSeq }) { - return await traceSpan({ description: "bulldozer-js.low-level.in-memory.compareAndSet", attributes }, async () => { + return await traceSpanHot({ description: "bulldozer-js.low-level.in-memory.compareAndSet", attributes }, async () => { if (key.byteLength > 64) throw new Error("KV store key must be <= 64 bytes"); if (compare.byteLength > 2_000_000_000) throw new Error("KV store compare must be <= 2GB"); if (value.byteLength > 2_000_000_000) throw new Error("KV store value must be <= 2GB"); @@ -107,7 +107,7 @@ export function declareInMemoryLowLevelDatabase(dbId: string): LowLevelDatabase }); }, async debugEntries() { - return await traceSpan({ description: "bulldozer-js.low-level.in-memory.debugEntries", attributes }, async () => [...base64KeyToValue.entries()] + return await traceSpanHot({ description: "bulldozer-js.low-level.in-memory.debugEntries", attributes }, async () => [...base64KeyToValue.entries()] .sort(([a], [b]) => stringCompare(a, b)) .map(([keyBase64, value]) => { const keyBytes = Buffer.from(keyBase64, "base64"); @@ -143,19 +143,19 @@ export function declareInMemoryLowLevelDatabase(dbId: string): LowLevelDatabase return declareInMemoryLowLevelKvStoreOrDump("store", JSON.stringify([dbId, storeId])); }, async waitUntilAvailable() { - return await traceSpan({ description: "bulldozer-js.low-level.in-memory.waitUntilAvailable", attributes: { "bulldozer.low_level.backend": "in-memory" } }, async () => {}); + return await traceSpanHot({ description: "bulldozer-js.low-level.in-memory.waitUntilAvailable", attributes: { "bulldozer.low_level.backend": "in-memory" } }, async () => {}); }, async waitUntilDurable() { - return await traceSpan({ description: "bulldozer-js.low-level.in-memory.waitUntilDurable", attributes: { "bulldozer.low_level.backend": "in-memory" } }, async () => {}); + return await traceSpanHot({ description: "bulldozer-js.low-level.in-memory.waitUntilDurable", attributes: { "bulldozer.low_level.backend": "in-memory" } }, async () => {}); }, async waitUntilReplicated() { - return await traceSpan({ description: "bulldozer-js.low-level.in-memory.waitUntilReplicated", attributes: { "bulldozer.low_level.backend": "in-memory" } }, async () => {}); + return await traceSpanHot({ description: "bulldozer-js.low-level.in-memory.waitUntilReplicated", attributes: { "bulldozer.low_level.backend": "in-memory" } }, async () => {}); }, combineSeqs(...seqs) { return this.initialSeq; }, async debugSnapshot() { - return await traceSpan({ description: "bulldozer-js.low-level.in-memory.debugSnapshot", attributes: { "bulldozer.low_level.backend": "in-memory" } }, async () => { + return await traceSpanHot({ description: "bulldozer-js.low-level.in-memory.debugSnapshot", attributes: { "bulldozer.low_level.backend": "in-memory" } }, async () => { const stores: Record = {}; const dumps: Record = {}; for (const [storeId, entries] of debugEntriesByStoreId.entries()) { diff --git a/apps/bulldozer-js/src/databases/low-level/implementations/instant-availability.ts b/apps/bulldozer-js/src/databases/low-level/implementations/instant-availability.ts index bafc6a592..937fcce8a 100644 --- a/apps/bulldozer-js/src/databases/low-level/implementations/instant-availability.ts +++ b/apps/bulldozer-js/src/databases/low-level/implementations/instant-availability.ts @@ -1,6 +1,6 @@ import { encodeBase64 } from "@hexclave/shared/dist/utils/bytes"; import { captureError } from "@hexclave/shared/dist/utils/errors"; -import { traceSpan } from "../../../otel.js"; +import { traceSpanHot } from "../../../otel.js"; import { DatabaseSeq } from "../../index.js"; import { LowLevelDatabase, LowLevelKvDump, LowLevelKvStore } from "../index.js"; @@ -84,14 +84,14 @@ export function declareInstantAvailabilityLowLevelDatabase(wrapped: LowLevelData }); }; const waitForPendingSeqRecordBudget = async () => { - await traceSpan({ description: "bulldozer-js.low-level.instant.waitForPendingSeqRecordBudget", attributes: { "bulldozer.low_level.backend": "instant-availability" } }, async () => { + await traceSpanHot({ description: "bulldozer-js.low-level.instant.waitForPendingSeqRecordBudget", attributes: { "bulldozer.low_level.backend": "instant-availability" } }, async () => { while (pendingSeqRecords.size >= maxPendingSeqRecords) { await pendingSeqRecordsChanged; } }); }; const withWriteGate = async (operation: () => Promise): Promise => { - return await traceSpan({ description: "bulldozer-js.low-level.instant.withWriteGate", attributes: { "bulldozer.low_level.backend": "instant-availability" } }, async () => { + return await traceSpanHot({ description: "bulldozer-js.low-level.instant.withWriteGate", attributes: { "bulldozer.low_level.backend": "instant-availability" } }, async () => { const previousOperation = currentWriteGateOperation; let releaseCurrentOperation: () => void; currentWriteGateOperation = new Promise(resolve => { @@ -109,7 +109,7 @@ export function declareInstantAvailabilityLowLevelDatabase(wrapped: LowLevelData const createSeq = (underlyingSeq: Promise) => { const seq = toSeq(crypto.randomUUID()); underlyingSeq.catch(() => {}); - const underlyingAvailable = traceSpan({ description: "bulldozer-js.low-level.instant.underlyingAvailable", attributes: { "bulldozer.low_level.backend": "instant-availability" } }, async () => { + const underlyingAvailable = traceSpanHot({ description: "bulldozer-js.low-level.instant.underlyingAvailable", attributes: { "bulldozer.low_level.backend": "instant-availability" } }, async () => { const resolvedUnderlyingSeq = await underlyingSeq; await wrapped.waitUntilAvailable(resolvedUnderlyingSeq); }); @@ -169,7 +169,7 @@ export function declareInstantAvailabilityLowLevelDatabase(wrapped: LowLevelData const result: LowLevelKvStore & LowLevelKvDump = { async get(key) { - return await traceSpan({ description: "bulldozer-js.low-level.instant.get", attributes }, async (span) => { + return await traceSpanHot({ description: "bulldozer-js.low-level.instant.get", attributes }, async (span) => { const cached = cachedValues.get(cacheKey(key)); span.setAttribute("bulldozer.low_level.instant.cache_hit", cached !== undefined); if (cached !== undefined) { @@ -185,13 +185,13 @@ export function declareInstantAvailabilityLowLevelDatabase(wrapped: LowLevelData }); }, async setAll(entries, setOptions) { - return await traceSpan({ description: "bulldozer-js.low-level.instant.setAll", attributes: { ...attributes, "bulldozer.low_level.entry_count": entries.length } }, async () => { + return await traceSpanHot({ description: "bulldozer-js.low-level.instant.setAll", attributes: { ...attributes, "bulldozer.low_level.entry_count": entries.length } }, async () => { if (entries.length === 0) return { seq: setOptions?.requiresSeq ?? initialSeq }; return await withWriteGate(async () => setAllLocked(entries, setOptions)); }); }, async deleteAll(keys) { - return await traceSpan({ description: "bulldozer-js.low-level.instant.deleteAll", attributes: { ...attributes, "bulldozer.low_level.key_count": keys.length } }, async () => { + return await traceSpanHot({ description: "bulldozer-js.low-level.instant.deleteAll", attributes: { ...attributes, "bulldozer.low_level.key_count": keys.length } }, async () => { if (keys.length === 0) return { seq: initialSeq }; return await withWriteGate(async () => { const keysForWrapped = keys.map(cloneArrayBuffer); @@ -207,7 +207,7 @@ export function declareInstantAvailabilityLowLevelDatabase(wrapped: LowLevelData }); }, async insertAll(values, insertOptions) { - return await traceSpan({ description: "bulldozer-js.low-level.instant.insertAll", attributes: { ...attributes, "bulldozer.low_level.value_count": values.length } }, async () => { + return await traceSpanHot({ description: "bulldozer-js.low-level.instant.insertAll", attributes: { ...attributes, "bulldozer.low_level.value_count": values.length } }, async () => { if (values.length === 0) return { keys: [], seq: insertOptions?.requiresSeq ?? initialSeq }; return await withWriteGate(async () => { const valuesForWrapped = values.map(cloneArrayBuffer); @@ -221,7 +221,7 @@ export function declareInstantAvailabilityLowLevelDatabase(wrapped: LowLevelData }); }, async compareAndSet(key, compare, value, compareAndSetOptions) { - return await traceSpan({ description: "bulldozer-js.low-level.instant.compareAndSet", attributes }, async () => { + return await traceSpanHot({ description: "bulldozer-js.low-level.instant.compareAndSet", attributes }, async () => { // The read+compare must happen inside the SAME write-gate critical section as the // subsequent write; otherwise two concurrent calls could both read the same value, // both pass the comparison, and both write — each returning wasSet: true, defeating @@ -235,7 +235,7 @@ export function declareInstantAvailabilityLowLevelDatabase(wrapped: LowLevelData }); }, async debugEntries() { - return await traceSpan({ description: "bulldozer-js.low-level.instant.debugEntries", attributes }, async () => await wrappedStore.debugEntries?.() ?? []); + return await traceSpanHot({ description: "bulldozer-js.low-level.instant.debugEntries", attributes }, async () => await wrappedStore.debugEntries?.() ?? []); }, }; @@ -267,13 +267,13 @@ export function declareInstantAvailabilityLowLevelDatabase(wrapped: LowLevelData return declareStoreOrDump(wrapped.declareKvDump(id) as LowLevelKvStore & LowLevelKvDump); }, async waitUntilAvailable() { - return await traceSpan({ description: "bulldozer-js.low-level.instant.waitUntilAvailable", attributes: { "bulldozer.low_level.backend": "instant-availability" } }, async () => {}); + return await traceSpanHot({ description: "bulldozer-js.low-level.instant.waitUntilAvailable", attributes: { "bulldozer.low_level.backend": "instant-availability" } }, async () => {}); }, async waitUntilDurable(seq) { - await traceSpan({ description: "bulldozer-js.low-level.instant.waitUntilDurable", attributes: { "bulldozer.low_level.backend": "instant-availability" } }, async () => await wrapped.waitUntilDurable(await getUnderlyingSeq(seq))); + await traceSpanHot({ description: "bulldozer-js.low-level.instant.waitUntilDurable", attributes: { "bulldozer.low_level.backend": "instant-availability" } }, async () => await wrapped.waitUntilDurable(await getUnderlyingSeq(seq))); }, async waitUntilReplicated(seq) { - await traceSpan({ description: "bulldozer-js.low-level.instant.waitUntilReplicated", attributes: { "bulldozer.low_level.backend": "instant-availability" } }, async () => await wrapped.waitUntilReplicated(await getUnderlyingSeq(seq))); + await traceSpanHot({ description: "bulldozer-js.low-level.instant.waitUntilReplicated", attributes: { "bulldozer.low_level.backend": "instant-availability" } }, async () => await wrapped.waitUntilReplicated(await getUnderlyingSeq(seq))); }, combineSeqs(...seqs) { if (seqs.length === 0) return initialSeq; @@ -283,7 +283,7 @@ export function declareInstantAvailabilityLowLevelDatabase(wrapped: LowLevelData return createSeq((async () => wrapped.combineSeqs(...await Promise.all(seqs.map(async seq => await getUnderlyingSeq(seq)))))()); }, async debugSnapshot() { - return await traceSpan({ description: "bulldozer-js.low-level.instant.debugSnapshot", attributes: { "bulldozer.low_level.backend": "instant-availability" } }, async () => await wrapped.debugSnapshot?.() ?? { stores: {}, dumps: {} }); + return await traceSpanHot({ description: "bulldozer-js.low-level.instant.debugSnapshot", attributes: { "bulldozer.low_level.backend": "instant-availability" } }, async () => await wrapped.debugSnapshot?.() ?? { stores: {}, dumps: {} }); }, debugInstantAvailability() { return { @@ -294,7 +294,7 @@ export function declareInstantAvailabilityLowLevelDatabase(wrapped: LowLevelData }; }, async waitUntilUnderlyingAvailable(seq) { - await traceSpan({ description: "bulldozer-js.low-level.instant.waitUntilUnderlyingAvailable", attributes: { "bulldozer.low_level.backend": "instant-availability" } }, async () => await getSeqRecord(seq).underlyingAvailable); + await traceSpanHot({ description: "bulldozer-js.low-level.instant.waitUntilUnderlyingAvailable", attributes: { "bulldozer.low_level.backend": "instant-availability" } }, async () => await getSeqRecord(seq).underlyingAvailable); }, initialSeq, }; diff --git a/apps/bulldozer-js/src/databases/low-level/implementations/lmdb.ts b/apps/bulldozer-js/src/databases/low-level/implementations/lmdb.ts index 82f6fec39..8c9b9b31e 100644 --- a/apps/bulldozer-js/src/databases/low-level/implementations/lmdb.ts +++ b/apps/bulldozer-js/src/databases/low-level/implementations/lmdb.ts @@ -1,7 +1,7 @@ import { encodeBase64 } from "@hexclave/shared/dist/utils/bytes"; import { wait } from "@hexclave/shared/dist/utils/promises"; import * as lmdb from "lmdb"; -import { traceSpan } from "../../../otel.js"; +import { traceSpanHot } from "../../../otel.js"; import { DatabaseSeq } from "../../index.js"; import { LowLevelDatabase, LowLevelDatabaseDebugEntry, LowLevelKvDump, LowLevelKvStore } from "../index.js"; @@ -179,7 +179,7 @@ export function declareLmdbLowLevelDatabase(options: { path: string, dbId?: stri }; const rememberAvailability = (seqId: string, promise: Promise) => { const insertedAt = performance.now(); - const availability = traceSpan({ description: "bulldozer-js.low-level.lmdb.availability", attributes: { "bulldozer.low_level.backend": "lmdb" } }, async () => await promise.then(() => { + const availability = traceSpanHot({ description: "bulldozer-js.low-level.lmdb.availability", attributes: { "bulldozer.low_level.backend": "lmdb" } }, async () => await promise.then(() => { activityStats.waitUntilAvailableResolveTotalMs += performance.now() - insertedAt; activityStats.waitUntilAvailableResolves++; seqToAvailability.delete(seqId); @@ -189,7 +189,7 @@ export function declareLmdbLowLevelDatabase(options: { path: string, dbId?: stri }; const rememberDurability = (seqId: string, promise: Promise) => { const insertedAt = performance.now(); - const durability = traceSpan({ description: "bulldozer-js.low-level.lmdb.durability", attributes: { "bulldozer.low_level.backend": "lmdb" } }, async () => await promise.then(async () => await root.flushed).then(() => { + const durability = traceSpanHot({ description: "bulldozer-js.low-level.lmdb.durability", attributes: { "bulldozer.low_level.backend": "lmdb" } }, async () => await promise.then(async () => await root.flushed).then(() => { activityStats.waitUntilDurableResolveTotalMs += performance.now() - insertedAt; activityStats.waitUntilDurableResolves++; seqToDurability.delete(seqId); @@ -199,7 +199,7 @@ export function declareLmdbLowLevelDatabase(options: { path: string, dbId?: stri }; const rememberCombinedAvailability = (seqId: string, promise: Promise) => { const insertedAt = performance.now(); - const availability = traceSpan({ description: "bulldozer-js.low-level.lmdb.combinedAvailability", attributes: { "bulldozer.low_level.backend": "lmdb" } }, async () => await promise.then(() => { + const availability = traceSpanHot({ description: "bulldozer-js.low-level.lmdb.combinedAvailability", attributes: { "bulldozer.low_level.backend": "lmdb" } }, async () => await promise.then(() => { activityStats.combinedSeqAvailabilityResolveTotalMs += performance.now() - insertedAt; activityStats.combinedSeqAvailabilityResolves++; combinedSeqToAvailability.delete(seqId); @@ -209,7 +209,7 @@ export function declareLmdbLowLevelDatabase(options: { path: string, dbId?: stri }; const rememberCombinedDurability = (seqId: string, promise: Promise) => { const insertedAt = performance.now(); - const durability = traceSpan({ description: "bulldozer-js.low-level.lmdb.combinedDurability", attributes: { "bulldozer.low_level.backend": "lmdb" } }, async () => await promise.then(() => { + const durability = traceSpanHot({ description: "bulldozer-js.low-level.lmdb.combinedDurability", attributes: { "bulldozer.low_level.backend": "lmdb" } }, async () => await promise.then(() => { activityStats.combinedSeqDurabilityResolveTotalMs += performance.now() - insertedAt; activityStats.combinedSeqDurabilityResolves++; combinedSeqToDurability.delete(seqId); @@ -225,7 +225,7 @@ export function declareLmdbLowLevelDatabase(options: { path: string, dbId?: stri const commit = (requiresSeq: DatabaseSeq, action: (version: number) => Promise) => { const version = nextVersion(); const seqId = nextSeqId(); - const promise = traceSpan({ description: "bulldozer-js.low-level.lmdb.commit", attributes: { "bulldozer.low_level.backend": "lmdb" } }, async () => { + const promise = traceSpanHot({ description: "bulldozer-js.low-level.lmdb.commit", attributes: { "bulldozer.low_level.backend": "lmdb" } }, async () => { const requiredSeqWaitStartedAt = performance.now(); await waitUntilAvailable(requiresSeq); activityStats.requiredSeqWaits++; @@ -307,7 +307,7 @@ export function declareLmdbLowLevelDatabase(options: { path: string, dbId?: stri const result: LowLevelKvStore & LowLevelKvDump = { async get(key) { - return await traceSpan({ description: "bulldozer-js.low-level.lmdb.get", attributes }, async () => { + return await traceSpanHot({ description: "bulldozer-js.low-level.lmdb.get", attributes }, async () => { validateKey(key); if (simulateReadMissDelayMs > 0) await wait(simulateReadMissDelayMs); const [buffer] = await db.getMany([bufferFromArrayBuffer(key)]); @@ -318,7 +318,7 @@ export function declareLmdbLowLevelDatabase(options: { path: string, dbId?: stri }); }, async setAll(entries, setOptions) { - return await traceSpan({ description: "bulldozer-js.low-level.lmdb.setAll", attributes: { ...attributes, "bulldozer.low_level.entry_count": entries.length } }, async () => { + return await traceSpanHot({ description: "bulldozer-js.low-level.lmdb.setAll", attributes: { ...attributes, "bulldozer.low_level.entry_count": entries.length } }, async () => { for (const { key, value } of entries) { validateKey(key); validateValue("value", value); @@ -334,7 +334,7 @@ export function declareLmdbLowLevelDatabase(options: { path: string, dbId?: stri }); }, async deleteAll(keys) { - return await traceSpan({ description: "bulldozer-js.low-level.lmdb.deleteAll", attributes: { ...attributes, "bulldozer.low_level.key_count": keys.length } }, async () => { + return await traceSpanHot({ description: "bulldozer-js.low-level.lmdb.deleteAll", attributes: { ...attributes, "bulldozer.low_level.key_count": keys.length } }, async () => { for (const key of keys) validateKey(key); if (keys.length === 0) return { seq: initialSeq }; return { @@ -345,13 +345,13 @@ export function declareLmdbLowLevelDatabase(options: { path: string, dbId?: stri }); }, async insertAll(values, insertOptions) { - return await traceSpan({ description: "bulldozer-js.low-level.lmdb.insertAll", attributes: { ...attributes, "bulldozer.low_level.value_count": values.length } }, async () => { + return await traceSpanHot({ description: "bulldozer-js.low-level.lmdb.insertAll", attributes: { ...attributes, "bulldozer.low_level.value_count": values.length } }, async () => { for (const value of values) validateValue("value", value); if (values.length === 0) return { keys: [], seq: insertOptions?.requiresSeq ?? initialSeq }; const version = nextVersion(); const seqId = nextSeqId(); const keys = values.map((_, index) => dumpKeyForVersion(version, index)); - const promise = traceSpan({ description: "bulldozer-js.low-level.lmdb.insertAll.commit", attributes }, async () => { + const promise = traceSpanHot({ description: "bulldozer-js.low-level.lmdb.insertAll.commit", attributes }, async () => { const requiredSeqWaitStartedAt = performance.now(); await waitUntilAvailable(insertOptions?.requiresSeq ?? initialSeq); activityStats.requiredSeqWaits++; @@ -381,7 +381,7 @@ export function declareLmdbLowLevelDatabase(options: { path: string, dbId?: stri }); }, async compareAndSet(key, compare, value, casOptions) { - return await traceSpan({ description: "bulldozer-js.low-level.lmdb.compareAndSet", attributes }, async () => { + return await traceSpanHot({ description: "bulldozer-js.low-level.lmdb.compareAndSet", attributes }, async () => { validateKey(key); validateValue("compare", compare); validateValue("value", value); @@ -397,7 +397,7 @@ export function declareLmdbLowLevelDatabase(options: { path: string, dbId?: stri }); }, async debugEntries() { - return await traceSpan({ description: "bulldozer-js.low-level.lmdb.debugEntries", attributes }, async () => await (db.getRange() as lmdb.RangeIterable<{ key: Uint8Array, value: Buffer }>).map(({ key, value }) => { + return await traceSpanHot({ description: "bulldozer-js.low-level.lmdb.debugEntries", attributes }, async () => await (db.getRange() as lmdb.RangeIterable<{ key: Uint8Array, value: Buffer }>).map(({ key, value }) => { const keyBuffer = Buffer.from(key); const valueBuffer = Buffer.from(value); return { @@ -440,13 +440,13 @@ export function declareLmdbLowLevelDatabase(options: { path: string, dbId?: stri return declareLmdbLowLevelKvStoreOrDump("store", storeId); }, async waitUntilAvailable(seq) { - await traceSpan({ description: "bulldozer-js.low-level.lmdb.waitUntilAvailable", attributes: { "bulldozer.low_level.backend": "lmdb" } }, async () => await waitUntilAvailable(seq)); + await traceSpanHot({ description: "bulldozer-js.low-level.lmdb.waitUntilAvailable", attributes: { "bulldozer.low_level.backend": "lmdb" } }, async () => await waitUntilAvailable(seq)); }, async waitUntilDurable(seq) { - await traceSpan({ description: "bulldozer-js.low-level.lmdb.waitUntilDurable", attributes: { "bulldozer.low_level.backend": "lmdb" } }, async () => await waitUntilDurable(seq)); + await traceSpanHot({ description: "bulldozer-js.low-level.lmdb.waitUntilDurable", attributes: { "bulldozer.low_level.backend": "lmdb" } }, async () => await waitUntilDurable(seq)); }, async waitUntilReplicated(seq) { - await traceSpan({ description: "bulldozer-js.low-level.lmdb.waitUntilReplicated", attributes: { "bulldozer.low_level.backend": "lmdb" } }, async () => { + await traceSpanHot({ description: "bulldozer-js.low-level.lmdb.waitUntilReplicated", attributes: { "bulldozer.low_level.backend": "lmdb" } }, async () => { await this.waitUntilAvailable(seq); await this.waitUntilDurable(seq); }); @@ -460,7 +460,7 @@ export function declareLmdbLowLevelDatabase(options: { path: string, dbId?: stri return toSeq(seqId); }, async debugSnapshot() { - return await traceSpan({ description: "bulldozer-js.low-level.lmdb.debugSnapshot", attributes: { "bulldozer.low_level.backend": "lmdb" } }, async () => { + return await traceSpanHot({ description: "bulldozer-js.low-level.lmdb.debugSnapshot", attributes: { "bulldozer.low_level.backend": "lmdb" } }, async () => { const stores: Record = {}; const dumps: Record = {}; for (const [storeId, entries] of debugEntriesByStoreId.entries()) { diff --git a/apps/bulldozer-js/src/databases/piledriver/index.ts b/apps/bulldozer-js/src/databases/piledriver/index.ts index 4980da937..4ff7deb1d 100644 --- a/apps/bulldozer-js/src/databases/piledriver/index.ts +++ b/apps/bulldozer-js/src/databases/piledriver/index.ts @@ -1,6 +1,6 @@ import { decodeBase64, encodeBase64 } from "@hexclave/shared/dist/utils/bytes"; import { runAsynchronously } from "@hexclave/shared/dist/utils/promises"; -import { traceSpan } from "../../otel.js"; +import { traceSpan, traceSpanHot } from "../../otel.js"; import { Database, DatabaseSeq } from "../index.js"; import { LowLevelDatabase, LowLevelDatabaseDebugSnapshot } from "../low-level/index.js"; @@ -10,6 +10,9 @@ export type PiledriverHeapObject = { [isPiledriverHeapObjectSymbol]: true, }; +const textEncoder = new TextEncoder(); +const textDecoder = new TextDecoder(); + const heapObjectsMapNullSentinel = { __heapObjectsMapNullSentinel: true }; const heapObjectsByObject = new WeakMap(); /** @@ -67,15 +70,11 @@ export type PiledriverDatabaseOptions = { disableHeapReadCache?: boolean, }; -// Tracks the chain of objects currently being serialized, so cycles fail fast with a clear -// error instead of hanging (a heap cycle would deadlock on its own memoized promise, and a -// plain object cycle would recurse forever). Sibling/DAG sharing is fine: only true ancestors -// are in the path. -type SerializationPath = { - objects: Set, - heapObjects: Set, -}; -const emptySerializationPath = (): SerializationPath => ({ objects: new Set(), heapObjects: new Set() }); +// Tracks the chain of *heap objects* currently being serialized, so heap cycles fail fast with +// a clear error instead of deadlocking on their own memoized promise. Sibling/DAG sharing is +// fine: only true ancestors are in the path. Plain-object cycles are detected separately with a +// push/pop set during the synchronous payload walk. +type HeapSerializationPath = ReadonlySet; type PendingHeapInsert = { buffer: ArrayBuffer, requiresSeq: DatabaseSeq, @@ -246,13 +245,12 @@ function serializationBranchKey(path: readonly string[]): string { return path[0] ?? ""; } -function serializationBranchStats(stats: PiledriverSerializationTimingStats | undefined, path: readonly string[]) { +function serializationBranchStatsByKey(stats: PiledriverSerializationTimingStats | undefined, branchKey: string) { if (stats === undefined) return undefined; - const key = serializationBranchKey(path); - let result = stats.branchStats.get(key); + let result = stats.branchStats.get(branchKey); if (result === undefined) { result = emptyPiledriverSerializationBranchStats(); - stats.branchStats.set(key, result); + stats.branchStats.set(branchKey, result); } return result; } @@ -297,7 +295,9 @@ export function declarePiledriverDatabase(lowLevelDb: LowLevelDatabase, options: stats.heapInsertAllTotalMs += heapInsertAllMs * entryCount / batch.length; } for (let i = 0; i < batch.length; i++) { - batch[i].resolve({ key: inserted.keys[i], seq: lowLevelDb.combineSeqs(batch[i].requiresSeq, inserted.seq) }); + // Most heap payloads (e.g. leaf row data) have no heap children, so requiresSeq is the + // singleton initial seq — skip the combined-seq allocation entirely for those. + batch[i].resolve({ key: inserted.keys[i], seq: batch[i].requiresSeq === lowLevelDb.initialSeq ? inserted.seq : lowLevelDb.combineSeqs(batch[i].requiresSeq, inserted.seq) }); } } catch (error) { for (const entry of batch) entry.reject(error); @@ -316,16 +316,27 @@ export function declarePiledriverDatabase(lowLevelDb: LowLevelDatabase, options: }); }; - const getHeapKeyAndSeq = async (heapObj: PiledriverHeapObject, path: SerializationPath = emptySerializationPath(), serializationTimingStats?: PiledriverSerializationTimingStats, logicalPath: readonly string[] = []): Promise<{ key: ArrayBuffer, seq: DatabaseSeq }> => { + // Deduplicates and drops initial seqs before delegating to the low-level combineSeqs. This is + // important for performance: each low-level combined seq allocates promises/map entries, so we + // only want to create one when there are actually ≥2 distinct non-initial seqs to combine. + // (Identity comparison against initialSeq is safe because initialSeq is a singleton object.) + const combineSeqsDeduped = (seqs: Iterable): DatabaseSeq => { + const unique = [...new Set(seqs)].filter(seq => seq !== lowLevelDb.initialSeq); + if (unique.length === 0) return lowLevelDb.initialSeq; + if (unique.length === 1) return unique[0]; + return lowLevelDb.combineSeqs(...unique); + }; + + const getHeapKeyAndSeq = async (heapObj: PiledriverHeapObject, heapPath: HeapSerializationPath, serializationTimingStats: PiledriverSerializationTimingStats | undefined, branchKey: string | undefined): Promise<{ key: ArrayBuffer, seq: DatabaseSeq }> => { // Must be checked before the memo lookup: awaiting the memoized promise of an ancestor // that is still being serialized would deadlock. - if (path.heapObjects.has(heapObj)) throw new Error("Piledriver objects must not contain cycles (found a cycle of heap objects)"); + if (heapPath.has(heapObj)) throw new Error("Piledriver objects must not contain cycles (found a cycle of heap objects)"); const existing = heapKeysAndSeqByHeapObjects.get(heapObj); if (existing) { if (serializationTimingStats !== undefined) { serializationTimingStats.heapObjectCacheHits++; - const branch = serializationBranchStats(serializationTimingStats, logicalPath); + const branch = branchKey === undefined ? undefined : serializationBranchStatsByKey(serializationTimingStats, branchKey); if (branch !== undefined) branch.heapObjectCacheHits++; } const cacheHitAwaitStartedAt = performance.now(); @@ -337,32 +348,29 @@ export function declarePiledriverDatabase(lowLevelDb: LowLevelDatabase, options: } if (serializationTimingStats !== undefined) { serializationTimingStats.heapObjectCacheMisses++; - const branch = serializationBranchStats(serializationTimingStats, logicalPath); + const branch = branchKey === undefined ? undefined : serializationBranchStatsByKey(serializationTimingStats, branchKey); if (branch !== undefined) branch.heapObjectCacheMisses++; } const promise = (async () => { - // A heap object starts a fresh plain-object path; plain object cycles can't span heap - // boundaries without also forming a heap cycle, which is tracked separately. - const childPath: SerializationPath = { objects: new Set(), heapObjects: new Set(path.heapObjects).add(heapObj) }; - return await traceSpan("bulldozer-js.piledriver.heap.serializeAndInsert", async () => { + const childHeapPath = new Set(heapPath).add(heapObj); + return await traceSpanHot("bulldozer-js.piledriver.heap.serializeAndInsert", async () => { const heapObjectGetStartedAt = performance.now(); const heapObject = await heapObj.get(); - let shape: string | undefined; if (serializationTimingStats !== undefined) { - shape = classifyHeapObjectPayload(heapObject); + const shape = classifyHeapObjectPayload(heapObject); const inlineNodeCount = totalInlineNodes(countPiledriverInlineNodes(heapObject)); incrementMapCount(serializationTimingStats.heapObjectCacheMissesByShape, shape); addMapCount(serializationTimingStats.heapObjectCacheMissInlineNodeCountsByShape, shape, inlineNodeCount); - const branch = serializationBranchStats(serializationTimingStats, logicalPath); + const branch = branchKey === undefined ? undefined : serializationBranchStatsByKey(serializationTimingStats, branchKey); if (branch !== undefined) { incrementMapCount(branch.heapObjectCacheMissesByShape, shape); addMapCount(branch.heapObjectCacheMissInlineNodeCountsByShape, shape, inlineNodeCount); } + serializationTimingStats.heapObjectGetTotalMs += performance.now() - heapObjectGetStartedAt; } - if (serializationTimingStats !== undefined) serializationTimingStats.heapObjectGetTotalMs += performance.now() - heapObjectGetStartedAt; const heapObjectSerializeStartedAt = performance.now(); - const serialized = await serializePiledriverObject(heapObject, childPath, serializationTimingStats, logicalPath); + const serialized = await serializePiledriverObject(heapObject, childHeapPath, serializationTimingStats, branchKey); if (serializationTimingStats !== undefined) serializationTimingStats.heapObjectSerializeTotalMs += performance.now() - heapObjectSerializeStartedAt; const heapObjectInsertStartedAt = performance.now(); const result = await insertHeapObjectBatched(serialized.buffer, serialized.seq, serializationTimingStats); @@ -404,7 +412,7 @@ export function declarePiledriverDatabase(lowLevelDb: LowLevelDatabase, options: let loadPromise: Promise | undefined; const heapObj: PiledriverHeapObject = { async get() { - loadPromise ??= traceSpan({ description: "bulldozer-js.piledriver.heap.get", attributes: { "bulldozer.piledriver.heap_read_cache_disabled": options.disableHeapReadCache === true } }, async () => { + loadPromise ??= traceSpanHot({ description: "bulldozer-js.piledriver.heap.get", attributes: { "bulldozer.piledriver.heap_read_cache_disabled": options.disableHeapReadCache === true } }, async () => { const { buffer, seq: heapSeq } = await heapDump.get(key); if (buffer === null) throw new Error(`Assertion error: Heap object with base64 key "${keyBase64}" not found`); const deserialized = await deserializePiledriverObject(buffer, heapSeq); @@ -426,148 +434,202 @@ export function declarePiledriverDatabase(lowLevelDb: LowLevelDatabase, options: return { object: heapObj, seq }; }; - const serializePiledriverObjectToJsonableObject = async (obj: PiledriverObject, path: SerializationPath, serializationTimingStats?: PiledriverSerializationTimingStats, logicalPath: readonly string[] = []): Promise<{ jsonableObject: unknown, seq: DatabaseSeq }> => { - switch (typeof obj) { - case "number": { - if (serializationTimingStats !== undefined) { - serializationTimingStats.primitiveNodes++; - const branch = serializationBranchStats(serializationTimingStats, logicalPath); - if (branch !== undefined) branch.primitiveNodes++; - } - if (!Number.isFinite(obj)) return { jsonableObject: [obj.toString()], seq: lowLevelDb.initialSeq }; - if (Object.is(obj, -0)) return { jsonableObject: ["-0"], seq: lowLevelDb.initialSeq }; - // intentionally fall through to the primitive case below - } - case "string": - case "boolean": { - if (serializationTimingStats !== undefined && typeof obj !== "number") { - serializationTimingStats.primitiveNodes++; - const branch = serializationBranchStats(serializationTimingStats, logicalPath); - if (branch !== undefined) branch.primitiveNodes++; - } - return { jsonableObject: obj, seq: lowLevelDb.initialSeq }; - } - case "object": { - if (obj === null) { - if (serializationTimingStats !== undefined) { - serializationTimingStats.primitiveNodes++; - const branch = serializationBranchStats(serializationTimingStats, logicalPath); + // A heap reference discovered during the synchronous payload walk. The jsonable slot is + // created immediately (as ["heap-reference", null]) and patched with the base64 key once the + // referenced heap object has been resolved/inserted. + type PendingHeapReferenceSlot = { + slot: [string, string | null], + heapObj: PiledriverHeapObject, + branchKey: string | undefined, + }; + + // Serializes a Piledriver object in two phases: + // 1. A fully SYNCHRONOUS walk that builds the jsonable structure, detects plain-object + // cycles (push/pop set — safe because nothing interleaves during a sync walk), counts + // stats, and records one slot per heap reference. + // 2. Resolution of the (deduplicated) referenced heap objects in parallel, then patching + // their base64 keys into the recorded slots. + // This replaces a previous fully-async recursive serializer that allocated promises for every + // node and called lowLevelDb.combineSeqs once per array/object node (each combined seq + // allocates a UUID, tracking promises, and map entries). Profiling showed that overhead — not + // LMDB — dominated CPU during backfills, so seqs are now combined exactly once per heap + // object/root, and only heap references involve async work at all. + const serializePiledriverObject = async (obj: PiledriverObject, heapPath: HeapSerializationPath, serializationTimingStats: PiledriverSerializationTimingStats | undefined, inheritedBranchKey?: string): Promise<{ buffer: ArrayBuffer, seq: DatabaseSeq }> => { + const stats = serializationTimingStats; + const pendingSlots: PendingHeapReferenceSlot[] = []; + const objectPath = new Set(); + // Branch keys only depend on the first 3 path segments (see serializationBranchKey), so we + // stop tracking the key path once the branch is determined. Nested heap objects inherit the + // branch key of their reference site, which is equivalent to the old logicalPath threading + // because heap references only occur at depth >= 3 in the root snapshot. + const keyPath: string[] = []; + + const build = (node: PiledriverObject, branch: PiledriverSerializationBranchStats | undefined, branchKey: string | undefined, branchDetermined: boolean): unknown => { + switch (typeof node) { + case "number": { + if (stats !== undefined) { + stats.primitiveNodes++; if (branch !== undefined) branch.primitiveNodes++; } - return { jsonableObject: obj, seq: lowLevelDb.initialSeq }; - } else if (Array.isArray(obj)) { - if (serializationTimingStats !== undefined) { - serializationTimingStats.arrayNodes++; - serializationTimingStats.arrayItems += obj.length; - const branch = serializationBranchStats(serializationTimingStats, logicalPath); - if (branch !== undefined) branch.arrayNodes++; + if (!Number.isFinite(node)) return [node.toString()]; + if (Object.is(node, -0)) return ["-0"]; + return node; + } + case "string": + case "boolean": { + if (stats !== undefined) { + stats.primitiveNodes++; + if (branch !== undefined) branch.primitiveNodes++; } - if (path.objects.has(obj)) throw new Error("Piledriver objects must not contain cycles"); - const childPath: SerializationPath = { ...path, objects: new Set(path.objects).add(obj) }; - const itemsSerializeResults = await Promise.all(obj.map(async o => await serializePiledriverObjectToJsonableObject(o, childPath, serializationTimingStats, logicalPath))); - return { - jsonableObject: ["array", itemsSerializeResults.map(r => r.jsonableObject)], - seq: lowLevelDb.combineSeqs(...itemsSerializeResults.map(r => r.seq)), - }; - } else if (isPiledriverHeapObjectSymbol in obj) { - if (serializationTimingStats !== undefined) { - serializationTimingStats.heapReferenceNodes++; - const branch = serializationBranchStats(serializationTimingStats, logicalPath); - if (branch !== undefined) branch.heapReferenceNodes++; - } - const heapKeyAndSeq = await getHeapKeyAndSeq(obj, path, serializationTimingStats, logicalPath); - return { - jsonableObject: ["heap-reference", encodeBase64(new Uint8Array(heapKeyAndSeq.key))], - seq: heapKeyAndSeq.seq, - }; - } else { - // "normal" object - // TODO: assert this is a POJO - - if (path.objects.has(obj)) throw new Error("Piledriver objects must not contain cycles"); - const childPath: SerializationPath = { ...path, objects: new Set(path.objects).add(obj) }; - const entries = Object.entries(obj); - if (serializationTimingStats !== undefined) { - serializationTimingStats.objectNodes++; - serializationTimingStats.objectEntries += entries.length; - const branch = serializationBranchStats(serializationTimingStats, logicalPath); - if (branch !== undefined) { - branch.objectNodes++; - branch.objectEntries += entries.length; + return node; + } + case "object": { + if (node === null) { + if (stats !== undefined) { + stats.primitiveNodes++; + if (branch !== undefined) branch.primitiveNodes++; } + return node; + } else if (Array.isArray(node)) { + if (stats !== undefined) { + stats.arrayNodes++; + stats.arrayItems += node.length; + if (branch !== undefined) branch.arrayNodes++; + } + if (objectPath.has(node)) throw new Error("Piledriver objects must not contain cycles"); + objectPath.add(node); + const items = node.map(item => build(item, branch, branchKey, branchDetermined)); + objectPath.delete(node); + return ["array", items]; + } else if (isPiledriverHeapObjectSymbol in node) { + if (stats !== undefined) { + stats.heapReferenceNodes++; + if (branch !== undefined) branch.heapReferenceNodes++; + } + // Fail fast on heap cycles at walk time (resolution would deadlock on the ancestor's + // own memoized promise otherwise). + if (heapPath.has(node)) throw new Error("Piledriver objects must not contain cycles (found a cycle of heap objects)"); + const slot: [string, string | null] = ["heap-reference", null]; + pendingSlots.push({ slot, heapObj: node, branchKey }); + return slot; + } else { + // "normal" object + // TODO: assert this is a POJO + + if (objectPath.has(node)) throw new Error("Piledriver objects must not contain cycles"); + objectPath.add(node); + const entries = Object.entries(node); + if (stats !== undefined) { + stats.objectNodes++; + stats.objectEntries += entries.length; + if (branch !== undefined) { + branch.objectNodes++; + branch.objectEntries += entries.length; + } + } + const result: Record = {}; + for (const [k, v] of entries) { + let childBranch = branch; + let childBranchKey = branchKey; + let childBranchDetermined = branchDetermined; + if (!branchDetermined) { + keyPath.push(k); + childBranchKey = serializationBranchKey(keyPath); + childBranch = stats === undefined ? undefined : serializationBranchStatsByKey(stats, childBranchKey); + childBranchDetermined = keyPath.length >= 3; + } + result[k] = build(v, childBranch, childBranchKey, childBranchDetermined); + if (!branchDetermined) keyPath.pop(); + } + objectPath.delete(node); + return result; } - const entriesSerializeResults = await Promise.all(entries.map(async ([k, v]) => [k, await serializePiledriverObjectToJsonableObject(v, childPath, serializationTimingStats, [...logicalPath, k])] as const)); - return { - jsonableObject: Object.fromEntries(entriesSerializeResults.map(([k, v]) => [k, v.jsonableObject] as const)), - seq: lowLevelDb.combineSeqs(...entriesSerializeResults.map(([_, v]) => v.seq)), - }; + } + default: { + throw new Error("Assertion error: Unknown type of Piledriver object " + typeof node); } } - default: { - throw new Error("Assertion error: Unknown type of Piledriver object " + typeof obj); - } - } - }; - - const serializePiledriverObject = async (obj: PiledriverObject, path: SerializationPath = emptySerializationPath(), serializationTimingStats?: PiledriverSerializationTimingStats, logicalPath: readonly string[] = []): Promise<{ buffer: ArrayBuffer, seq: DatabaseSeq }> => { - const toJsonableStartedAt = performance.now(); - const toJsonableResponse = await serializePiledriverObjectToJsonableObject(obj, path, serializationTimingStats, logicalPath); - if (serializationTimingStats !== undefined) serializationTimingStats.serializeToJsonableTotalMs += performance.now() - toJsonableStartedAt; - const jsonStringifyStartedAt = performance.now(); - const json = JSON.stringify(toJsonableResponse.jsonableObject); - if (serializationTimingStats !== undefined) serializationTimingStats.jsonStringifyTotalMs += performance.now() - jsonStringifyStartedAt; - const textEncodeStartedAt = performance.now(); - const buffer = new TextEncoder().encode(json).buffer; - if (serializationTimingStats !== undefined) serializationTimingStats.textEncodeTotalMs += performance.now() - textEncodeStartedAt; - return { - buffer, - seq: toJsonableResponse.seq, }; + + const toJsonableStartedAt = performance.now(); + const inheritedBranch = inheritedBranchKey === undefined ? undefined : serializationBranchStatsByKey(stats, inheritedBranchKey); + const jsonableObject = build(obj, inheritedBranch, inheritedBranchKey, inheritedBranchKey !== undefined); + if (stats !== undefined) stats.serializeToJsonableTotalMs += performance.now() - toJsonableStartedAt; + + let seq = lowLevelDb.initialSeq; + if (pendingSlots.length > 0) { + // Resolve each distinct heap object once, even if it is referenced from multiple slots. + const slotsByHeapObj = new Map(); + for (const pendingSlot of pendingSlots) { + const existing = slotsByHeapObj.get(pendingSlot.heapObj); + if (existing === undefined) slotsByHeapObj.set(pendingSlot.heapObj, [pendingSlot]); + else existing.push(pendingSlot); + } + const resolvedSeqs = await Promise.all([...slotsByHeapObj.entries()].map(async ([heapObj, slots]) => { + const heapKeyAndSeq = await getHeapKeyAndSeq(heapObj, heapPath, stats, slots[0].branchKey); + const keyBase64 = encodeBase64(new Uint8Array(heapKeyAndSeq.key)); + for (const { slot } of slots) slot[1] = keyBase64; + return heapKeyAndSeq.seq; + })); + seq = combineSeqsDeduped(resolvedSeqs); + } + + const jsonStringifyStartedAt = performance.now(); + const json = JSON.stringify(jsonableObject); + if (stats !== undefined) stats.jsonStringifyTotalMs += performance.now() - jsonStringifyStartedAt; + const textEncodeStartedAt = performance.now(); + const buffer = textEncoder.encode(json).buffer; + if (stats !== undefined) stats.textEncodeTotalMs += performance.now() - textEncodeStartedAt; + return { buffer, seq }; }; - const deserializePiledriverObjectFromJsonableObject = async (jsonableObject: unknown, enclosingSeq: DatabaseSeq): Promise<{ object: PiledriverObject, seq: DatabaseSeq }> => { + // Fully synchronous (getHeapObjectByKey is sync; heap payloads are only fetched lazily on + // .get()). Seqs are collected into one array and combined once per buffer instead of once per + // node — see serializePiledriverObject for why this matters. + const deserializePiledriverObjectFromJsonableObject = (jsonableObject: unknown, enclosingSeq: DatabaseSeq, seqs: DatabaseSeq[]): PiledriverObject => { switch (typeof jsonableObject) { case "string": case "number": case "boolean": { - return { object: jsonableObject, seq: lowLevelDb.initialSeq }; + return jsonableObject; } case "object": { if (jsonableObject === null) { - return { object: jsonableObject, seq: lowLevelDb.initialSeq }; + return jsonableObject; } else if (Array.isArray(jsonableObject)) { switch (jsonableObject[0]) { case "array": { - const itemsDeserializeResults = await Promise.all(jsonableObject[1].map(async (o: any) => await deserializePiledriverObjectFromJsonableObject(o, enclosingSeq))); - return { object: itemsDeserializeResults.map(r => r.object), seq: lowLevelDb.combineSeqs(...itemsDeserializeResults.map(r => r.seq)) }; + // any: JSON.parse output is structurally validated by the surrounding switch; a malformed + // payload would throw in the recursive call rather than silently passing through. + return jsonableObject[1].map((o: any) => deserializePiledriverObjectFromJsonableObject(o, enclosingSeq, seqs)); } case "heap-reference": { const heapObjAndSeq = getHeapObjectByKey(decodeBase64(jsonableObject[1]).buffer, enclosingSeq); - return { object: heapObjAndSeq.object, seq: heapObjAndSeq.seq }; + seqs.push(heapObjAndSeq.seq); + return heapObjAndSeq.object; } case "NaN": { - return { object: NaN, seq: lowLevelDb.initialSeq }; + return NaN; } case "Infinity": { - return { object: Infinity, seq: lowLevelDb.initialSeq }; + return Infinity; } case "-Infinity": { - return { object: -Infinity, seq: lowLevelDb.initialSeq }; + return -Infinity; } case "-0": { - return { object: -0, seq: lowLevelDb.initialSeq }; + return -0; } default: { throw new Error("Assertion error: Serialized Piledriver JSONable object array has unknown type " + jsonableObject[0]); } } } else { - const entries = Object.entries(jsonableObject); - const entriesDeserializeResults = await Promise.all(entries.map(async ([k, v]) => [k, await deserializePiledriverObjectFromJsonableObject(v, enclosingSeq)] as const)); - return { - object: Object.fromEntries(entriesDeserializeResults.map(([k, v]) => [k, v.object] as const)), - seq: lowLevelDb.combineSeqs(...entriesDeserializeResults.map(([_, v]) => v.seq)), - }; + const result: Record = {}; + for (const [k, v] of Object.entries(jsonableObject)) { + result[k] = deserializePiledriverObjectFromJsonableObject(v, enclosingSeq, seqs); + } + return result; } } default: { @@ -577,7 +639,9 @@ export function declarePiledriverDatabase(lowLevelDb: LowLevelDatabase, options: }; const deserializePiledriverObject = async (buffer: ArrayBuffer, enclosingSeq: DatabaseSeq): Promise<{ object: PiledriverObject, seq: DatabaseSeq }> => { - return await deserializePiledriverObjectFromJsonableObject(JSON.parse(new TextDecoder().decode(buffer)), enclosingSeq); + const seqs: DatabaseSeq[] = []; + const object = deserializePiledriverObjectFromJsonableObject(JSON.parse(textDecoder.decode(buffer)), enclosingSeq, seqs); + return { object, seq: combineSeqsDeduped(seqs) }; }; const parseDebugEntryValue = (valueUtf8: string | null) => { @@ -618,7 +682,7 @@ export function declarePiledriverDatabase(lowLevelDb: LowLevelDatabase, options: const startedAt = performance.now(); const serializeStartedAt = performance.now(); const serializeCpuStartedAt = process.cpuUsage(); - const { buffer, seq } = await serializePiledriverObject(value, emptySerializationPath(), timingStats); + const { buffer, seq } = await serializePiledriverObject(value, new Set(), timingStats); const serializeCpuUsage = process.cpuUsage(serializeCpuStartedAt); const serializePiledriverObjectMs = performance.now() - serializeStartedAt; const serializeCpuMs = (serializeCpuUsage.user + serializeCpuUsage.system) / 1000; diff --git a/apps/bulldozer-js/src/otel.ts b/apps/bulldozer-js/src/otel.ts index 4a983dc4e..35d8a8e72 100644 --- a/apps/bulldozer-js/src/otel.ts +++ b/apps/bulldozer-js/src/otel.ts @@ -28,3 +28,18 @@ export async function traceSpan(optionsOrDescription: string | { description: return await fn(span); }); } + +const hotPathTracingEnabled = getEnvVariable("HEXCLAVE_BULLDOZER_HOT_PATH_TRACING", "") === "true"; +const noopTraceSpan: TraceSpan = { setAttribute: () => {} }; + +/** + * Like traceSpan, but for operations that run at very high frequency (per KV put, per seq, per + * heap object, ...). CPU profiling showed span creation alone was >20% of process CPU during + * backfills (plus a large share of GC), so these spans are disabled unless + * HEXCLAVE_BULLDOZER_HOT_PATH_TRACING=true is set. Coarse operations (setRootObject, snapshot + * mutations, HTTP handlers) keep using traceSpan unconditionally. + */ +export async function traceSpanHot(optionsOrDescription: string | { description: string, attributes?: Record }, fn: (span: TraceSpan) => Promise): Promise { + if (!hotPathTracingEnabled) return await fn(noopTraceSpan); + return await traceSpan(optionsOrDescription, fn); +} From f5f440757c25c58387db075ee4513528f9a7274b Mon Sep 17 00:00:00 2001 From: Konstantin Wohlwend Date: Fri, 3 Jul 2026 15:46:14 -0700 Subject: [PATCH 4/5] Left join table no longer has quadratic behavior --- .../src/databases/bulldozer/index.ts | 8 +- .../augmented-tree-map.test.ts | 76 +++++++++++++++++++ .../data-structures/augmented-tree-map.ts | 43 +++++++++-- .../bulldozer-js/src/payments/schema/index.ts | 2 +- 4 files changed, 120 insertions(+), 9 deletions(-) diff --git a/apps/bulldozer-js/src/databases/bulldozer/index.ts b/apps/bulldozer-js/src/databases/bulldozer/index.ts index fb26338f5..776caeca9 100644 --- a/apps/bulldozer-js/src/databases/bulldozer/index.ts +++ b/apps/bulldozer-js/src/databases/bulldozer/index.ts @@ -2368,7 +2368,11 @@ export function declareLeftJoinTable(options: { const addRight = async (row: InputRow) => { const right: StoredRow = { ...row, joinKey: await options.rightJoinKeyExtractor(row) }; const lefts = await leftMatches(right.joinKey); - if (!(await state.rightByJoinKey.getAll(right.joinKey)).length) for (const left of lefts) await deleteOutput(left, null); + // hasAny (O(depth)) instead of getAll().length: many rows can share one join key (e.g. + // a null-ish key on most rows), and materializing all of them per inserted row makes + // bulk ingestion O(n^2). The check is only needed when there are matching lefts whose + // left-with-null output row must be retracted. + if (lefts.length > 0 && !(await state.rightByJoinKey.hasAny(right.joinKey))) for (const left of lefts) await deleteOutput(left, null); state = { ...state, rightRows: await state.rightRows.set(right.rowIdentifier, right), rightByJoinKey: await state.rightByJoinKey.add(right.joinKey, right.rowIdentifier, right.rowIdentifier) }; for (const left of lefts) await addOutput(left, right); }; @@ -2378,7 +2382,7 @@ export function declareLeftJoinTable(options: { const lefts = await leftMatches(right.joinKey); for (const left of lefts) await deleteOutput(left, right); state = { ...state, rightRows: await state.rightRows.delete(right.rowIdentifier), rightByJoinKey: await state.rightByJoinKey.delete(right.joinKey, right.rowIdentifier) }; - if (!(await state.rightByJoinKey.getAll(right.joinKey)).length) for (const left of lefts) await addOutput(left, null); + if (lefts.length > 0 && !(await state.rightByJoinKey.hasAny(right.joinKey))) for (const left of lefts) await addOutput(left, null); }; for (const row of changedRowsFromTableChanges(changes.left)) { diff --git a/apps/bulldozer-js/src/databases/piledriver/data-structures/augmented-tree-map.test.ts b/apps/bulldozer-js/src/databases/piledriver/data-structures/augmented-tree-map.test.ts index b2f20e747..1ffcc3a8e 100644 --- a/apps/bulldozer-js/src/databases/piledriver/data-structures/augmented-tree-map.test.ts +++ b/apps/bulldozer-js/src/databases/piledriver/data-structures/augmented-tree-map.test.ts @@ -162,6 +162,34 @@ function withHeapCounters(tree: AugmentedTreeMap, arity }; } +function withMultiMapHeapCounters(tree: AugmentedTreeMultiMap, multiMapOptions: ConstructorParameters>[0]) { + let gets = 0; + const seen = new WeakMap(); + + const wrapRef = (ref: PiledriverHeapObject): PiledriverHeapObject => { + const cached = seen.get(ref); + if (cached) return cached; + const wrapped = { + async get() { + gets++; + const node: any = await ref.get(); + return isNode(node) ? { ...node, children: node.children.map(wrapChild) } : node; + }, + [isPiledriverHeapObjectSymbol]: true as const, + } as PiledriverHeapObject; + seen.set(ref, wrapped); + return wrapped; + }; + const wrapChild = (child: any) => child ? { ...child, ref: wrapRef(child.ref) } : child; + const serialized = tree.toPiledriverObject() as { type: string, tree: { type: string, root: any } }; + + return { + tree: AugmentedTreeMultiMap.fromPiledriverObject({ ...serialized, tree: { ...serialized.tree, root: wrapChild(serialized.tree.root) } }, multiMapOptions), + reset: () => { gets = 0; }, + gets: () => gets, + }; +} + async function collectRefs(tree: AugmentedTreeMap) { const refs = new Set(); const visit = async (child: Serialized["root"]) => { @@ -295,6 +323,54 @@ describe("AugmentedTreeMap", () => { expect(await arrayFrom(tree.entries())).toEqual([[10, "b", 2], [11, "a", 4]]); }); + it("answers hasAny without materializing all entries for a key", async () => { + const multiMapOptions = { + ...options(8), + entryIdComparator: stringCompare, + }; + let tree = new AugmentedTreeMultiMap(multiMapOptions); + + expect(await tree.hasAny(10)).toBe(false); + for (let i = 0; i < 1000; i++) tree = await tree.add(10, `id-${i}`, i); + tree = await tree.add(11, "a", 1); + + expect(await tree.hasAny(10)).toBe(true); + expect(await tree.hasAny(11)).toBe(true); + expect(await tree.hasAny(12)).toBe(false); + + // hasAny must stay O(depth) even when many entries share the key; getAll here would load + // all 1000 entries (hundreds of node reads). + const counted = withMultiMapHeapCounters(tree, multiMapOptions); + counted.reset(); + await counted.tree.hasAny(10); + expect(counted.gets()).toBeLessThanOrEqual(8); + }); + + it("uses binary search for bounded range scans inside large nodes", async () => { + let comparisons = 0; + const countedOptions = { + arity: 2048, + comparator: (a: number, b: number) => { + comparisons++; + return a - b; + }, + initialAugmentation: 0, + extractAugmentation: (value: number) => value, + mergeAugmentations: (...values: number[]) => values.reduce((sum, value) => sum + value, 0), + entryIdComparator: stringCompare, + }; + let tree = new AugmentedTreeMultiMap(countedOptions); + for (let i = 0; i < 1000; i++) tree = await tree.add(i, "", i); + + comparisons = 0; + expect(await arrayFrom(tree.entries({ gte: 990, limit: 1 }))).toEqual([[990, "", 990]]); + expect(comparisons).toBeLessThan(80); + + comparisons = 0; + expect(await arrayFrom(tree.entries({ lte: 10, reverse: true, limit: 1 }))).toEqual([[10, "", 10]]); + expect(comparisons).toBeLessThan(80); + }); + it("keeps heap reads logarithmic for point operations", async () => { for (const size of [100, 1_000, 10_000]) { const counted = withHeapCounters(await build(size)); diff --git a/apps/bulldozer-js/src/databases/piledriver/data-structures/augmented-tree-map.ts b/apps/bulldozer-js/src/databases/piledriver/data-structures/augmented-tree-map.ts index dd37a2659..6177fb0da 100644 --- a/apps/bulldozer-js/src/databases/piledriver/data-structures/augmented-tree-map.ts +++ b/apps/bulldozer-js/src/databases/piledriver/data-structures/augmented-tree-map.ts @@ -105,6 +105,15 @@ export class AugmentedTreeMultiMap { + for await (const _entry of this.entries({ gte: key, lte: key, limit: 1 })) return true; + return false; + } + async add(key: Key, entryId: EntryId, value: Value) { return await this.insertRaw({ key, id: entryId }, value); } @@ -197,6 +206,8 @@ export class AugmentedTreeMultiMap> = {}): AsyncIterable<[MultiKey, Value]> { let yielded = 0; + const lowerBound = this.tighterLowerBound(options.gte, options.gt); + const upperBound = this.tighterUpperBound(options.lte, options.lt); const isInRange = (key: MultiKey) => (options.gte === undefined || this.compareKeys(key, options.gte) >= 0) && (options.gt === undefined || this.compareKeys(key, options.gt) > 0) @@ -211,18 +222,21 @@ export class AugmentedTreeMultiMap= 0; i--) { - yield* walk(tree, node.children[i + 1] ?? null); + yield* walk(tree, node.children[endIndex] ?? null); + for (let i = endIndex - 1; i >= startIndex; i--) { yield* visitEntry(node.entries[i]); + yield* walk(tree, node.children[i] ?? null); } - yield* walk(tree, node.children[0] ?? null); } else { - for (let i = 0; i < node.entries.length; i++) { + for (let i = startIndex; i < endIndex; i++) { yield* walk(tree, node.children[i] ?? null); yield* visitEntry(node.entries[i]); } - yield* walk(tree, node.children[node.entries.length] ?? null); + yield* walk(tree, node.children[endIndex] ?? null); } }; @@ -333,13 +347,30 @@ export class AugmentedTreeMultiMap, StoredValue>[], key: MultiKey) { + const low = this.lowerBoundIndex(entries, key); + return { index: low, found: low < entries.length && this.compareKeys(entries[low][0], key) === 0 }; + } + + private lowerBoundIndex(entries: Entry, StoredValue>[], key: MultiKey) { let low = 0, high = entries.length; while (low < high) { const mid = (low + high) >> 1; if (this.compareKeys(entries[mid][0], key) < 0) low = mid + 1; else high = mid; } - return { index: low, found: low < entries.length && this.compareKeys(entries[low][0], key) === 0 }; + return low; + } + + private tighterLowerBound(a: MultiKey | undefined, b: MultiKey | undefined) { + if (a === undefined) return b; + if (b === undefined) return a; + return this.compareKeys(a, b) >= 0 ? a : b; + } + + private tighterUpperBound(a: MultiKey | undefined, b: MultiKey | undefined) { + if (a === undefined) return b; + if (b === undefined) return a; + return this.compareKeys(a, b) <= 0 ? a : b; } private async finishUpsert(result: { root: Child, Augmentation>, split?: Split, Value, Augmentation> }) { diff --git a/apps/bulldozer-js/src/payments/schema/index.ts b/apps/bulldozer-js/src/payments/schema/index.ts index 8b99c6313..87231cac0 100644 --- a/apps/bulldozer-js/src/payments/schema/index.ts +++ b/apps/bulldozer-js/src/payments/schema/index.ts @@ -668,7 +668,7 @@ export function createPaymentsSchema() { const subscription = rowObject(row.rowData); return { tenancyId: subscription.tenancyId, stripeSubscriptionId: subscription.stripeSubscriptionId }; }, - joinKeyComparator: compareJson, + joinKeyComparator: (a: any, b: any) => stringCompare(a.tenancyId, b.tenancyId) || stringCompare(a.stripeSubscriptionId ?? "", b.stripeSubscriptionId ?? "") || (a.stripeSubscriptionId === b.stripeSubscriptionId ? 0 : a.stripeSubscriptionId === null ? -1 : 1), joiner: async (left, right) => ({ leftRowData: left.rowData, rightRowData: right?.rowData ?? null }), }), { left: "payments-subscription-invoices", right: "payments-subscriptions" }), table("payments-renewal-invoice-rows", defineFilterTable(row => { From b6a5fb2ac61c1b9f84fd0aaaf3b9953e1b75baf7 Mon Sep 17 00:00:00 2001 From: nams1570 Date: Fri, 3 Jul 2026 16:28:25 -0700 Subject: [PATCH 5/5] fix: fail loud if batch size not passed correctly --- .../scripts/bulldozer-payments-init.ts | 26 ++++++++++++++++++- 1 file changed, 25 insertions(+), 1 deletion(-) diff --git a/apps/backend/scripts/bulldozer-payments-init.ts b/apps/backend/scripts/bulldozer-payments-init.ts index 0247e0e93..5d64a7569 100644 --- a/apps/backend/scripts/bulldozer-payments-init.ts +++ b/apps/backend/scripts/bulldozer-payments-init.ts @@ -466,7 +466,13 @@ export function parseBackfillResumeOptions(args: string[]): BackfillResumeOption const continueOnError = args.includes("--continue-on-error"); - const batchSizeArg = readArg("batch-size"); + // Accept both `--batch-size=500` and the space form `--batch-size 500`. The + // space form arrives as two separate argv tokens, so `readArg` (which only + // matches the `name=` prefix) would miss it and we'd silently fall back to the + // default — a nasty footgun. Read the following token in that case, and fail + // loudly if `--batch-size` is present but has no usable value. + const bareBatchSizeIndex = args.indexOf("--batch-size"); + const batchSizeArg = readArg("batch-size") ?? (bareBatchSizeIndex === -1 ? undefined : args[bareBatchSizeIndex + 1]); let batchSize: number | undefined = undefined; if (batchSizeArg !== undefined) { const parsed = Number(batchSizeArg); @@ -474,6 +480,8 @@ export function parseBackfillResumeOptions(args: string[]): BackfillResumeOption throw new Error(`--batch-size must be a positive integer (got "${batchSizeArg}")`); } batchSize = parsed; + } else if (bareBatchSizeIndex !== -1) { + throw new Error("--batch-size requires a value, e.g. --batch-size=500 or --batch-size 500"); } // Common options that apply regardless of whether a resume cursor was passed. const base: BackfillResumeOptions = { continueOnError, ...(batchSize !== undefined ? { batchSize } : {}) }; @@ -648,6 +656,22 @@ import.meta.vitest?.describe("parseBackfillResumeOptions", (test) => { .toEqual({ continueOnError: false, batchSize: 1000 }); }); + test("parses the space form --batch-size 100 (two argv tokens)", ({ expect }) => { + expect(parseBackfillResumeOptions(["--batch-size", "100"])) + .toEqual({ continueOnError: false, batchSize: 100 }); + // ...including after the command token, as it arrives via the CLI. + expect(parseBackfillResumeOptions(["backfill-bulldozer-from-prisma", "--batch-size", "100"])) + .toEqual({ continueOnError: false, batchSize: 100 }); + }); + + test("throws (never silently defaults) when --batch-size has no value", ({ expect }) => { + expect(() => parseBackfillResumeOptions(["--batch-size"])) + .toThrow("--batch-size requires a value"); + // A following flag is not a value → still a positive-integer failure, loud. + expect(() => parseBackfillResumeOptions(["--batch-size", "--continue-on-error"])) + .toThrow("--batch-size must be a positive integer"); + }); + test("parses --batch-size alongside resume flags", ({ expect }) => { expect(parseBackfillResumeOptions(["--resume-table=Subscription", "--resume-cursor=ten-1,sub-9", "--batch-size=250"])) .toEqual({