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); +}