CodeCow/stack
1
0
Fork 0
mirror of https://github.com/stack-auth/stack.git synced 2026-07-20 21:29:36 +08:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

Merge branch 'dev' into devin/1783108675-session-replay-perf

Browse Source
This commit is contained in:
Armaan Jain 2026-07-03 16:42:04 -07:00 committed by GitHub
commit 4abc80c5f7
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
10 changed files with 899 additions and 182 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
apps/backend/scripts/bulldozer-payments-init.ts
View File

@ -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({

71
apps/bulldozer-js/src/databases/bulldozer/index.ts
View File

@ -23,12 +23,23 @@ async function fromAsync<T>(iterable: AsyncIterable<T>): Promise<T[]> {
* 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<object, string>();
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);
}
@ -887,15 +898,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;
});
};
@ -2313,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);
};
@ -2323,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)) {

22
apps/bulldozer-js/src/databases/low-level/implementations/in-memory.ts
View File

@ -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<string, LowLevelDatabaseDebugEntry[]> = {};
const dumps: Record<string, LowLevelDatabaseDebugEntry[]> = {};
for (const [storeId, entries] of debugEntriesByStoreId.entries()) {

30
apps/bulldozer-js/src/databases/low-level/implementations/instant-availability.ts
View File

@ -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 <T>(operation: () => Promise<T>): Promise<T> => {
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<void>(resolve => {
@ -109,7 +109,7 @@ export function declareInstantAvailabilityLowLevelDatabase(wrapped: LowLevelData
const createSeq = (underlyingSeq: Promise<DatabaseSeq>) => {
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,
};

134
apps/bulldozer-js/src/databases/low-level/implementations/lmdb.ts
View File

@ -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";
@ -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,
@ -152,7 +179,7 @@ export function declareLmdbLowLevelDatabase(options: { path: string, dbId?: stri
};
const rememberAvailability = (seqId: string, promise: Promise<unknown>) => {
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);
@ -162,7 +189,7 @@ export function declareLmdbLowLevelDatabase(options: { path: string, dbId?: stri
};
const rememberDurability = (seqId: string, promise: Promise<unknown>) => {
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);
@ -172,7 +199,7 @@ export function declareLmdbLowLevelDatabase(options: { path: string, dbId?: stri
};
const rememberCombinedAvailability = (seqId: string, promise: Promise<unknown>) => {
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);
@ -182,7 +209,7 @@ export function declareLmdbLowLevelDatabase(options: { path: string, dbId?: stri
};
const rememberCombinedDurability = (seqId: string, promise: Promise<unknown>) => {
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);
@ -198,13 +225,32 @@ export function declareLmdbLowLevelDatabase(options: { path: string, dbId?: stri
const commit = (requiresSeq: DatabaseSeq, action: (version: number) => Promise<void>) => {
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 = traceSpanHot({ 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<void>) => {
@ -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));
@ -255,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)]);
@ -266,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);
@ -282,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 {
@ -293,26 +345,43 @@ 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 () => 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 = traceSpanHot({ 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) };
});
},
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);
@ -328,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 {
@ -371,26 +440,27 @@ 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);
});
},
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)))));
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<string, LowLevelDatabaseDebugEntry[]> = {};
const dumps: Record<string, LowLevelDatabaseDebugEntry[]> = {};
for (const [storeId, entries] of debugEntriesByStoreId.entries()) {

76
apps/bulldozer-js/src/databases/piledriver/data-structures/augmented-tree-map.test.ts
View File

@ -162,6 +162,34 @@ function withHeapCounters(tree: AugmentedTreeMap<number, number, number>, arity
};
}
function withMultiMapHeapCounters(tree: AugmentedTreeMultiMap<number, number, number, string>, multiMapOptions: ConstructorParameters<typeof AugmentedTreeMultiMap<number, number, number, string>>[0]) {
let gets = 0;
const seen = new WeakMap<PiledriverHeapObject, PiledriverHeapObject>();
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<number, number, number, string>({ ...serialized, tree: { ...serialized.tree, root: wrapChild(serialized.tree.root) } }, multiMapOptions),
reset: () => { gets = 0; },
gets: () => gets,
};
}
async function collectRefs(tree: AugmentedTreeMap<number, number, number>) {
const refs = new Set<PiledriverHeapObject>();
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<number, number, number, string>(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));

92
apps/bulldozer-js/src/databases/piledriver/data-structures/augmented-tree-map.ts
View File

@ -26,10 +26,11 @@ const noAugmentation = Symbol("no-augmentation");
type NoAugmentation = typeof noAugmentation;
type Entry<K, V> = [K, V];
type Child<K, A> = { ref: PiledriverHeapObject, augmentation: A, size: number, entryCount: number, minKey: K, maxKey: K };
type Split<K, V, A> = { entry: Entry<K, V>, right: Child<K, A> };
type StoredValue<V extends PiledriverObject> = V | PiledriverHeapObject;
type Split<K, V extends PiledriverObject, A> = { entry: Entry<K, StoredValue<V>>, right: Child<K, A> };
// Persisted B-tree node. Children are heap objects, so unchanged subtrees are shared across versions.
type Node<K, V, A> = {
entries: Entry<K, V>[],
type Node<K, V extends PiledriverObject, A> = {
entries: Entry<K, StoredValue<V>>[],
children: Child<K, A>[],
augmentation: A,
size: number,
@ -104,6 +105,15 @@ export class AugmentedTreeMultiMap<Key extends PiledriverObject, Value extends P
return result;
}
// O(depth) existence check. Callers that only need to know whether *any* entry exists for a
// key must use this instead of `getAll(key).length`: getAll materializes every entry, which
// is O(n) when many entries share one key (e.g. null-ish join keys) and turns bulk ingestion
// into O(n^2).
async hasAny(key: Key): Promise<boolean> {
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);
}
@ -176,7 +186,7 @@ export class AugmentedTreeMultiMap<Key extends PiledriverObject, Value extends P
private async getRaw(key: MultiKey<Key, EntryId>): Promise<Value | undefined> {
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);
}
}
@ -196,6 +206,8 @@ export class AugmentedTreeMultiMap<Key extends PiledriverObject, Value extends P
private async *rawEntries(options: EntryOptions<MultiKey<Key, EntryId>> = {}): AsyncIterable<[MultiKey<Key, EntryId>, Value]> {
let yielded = 0;
const lowerBound = this.tighterLowerBound(options.gte, options.gt);
const upperBound = this.tighterUpperBound(options.lte, options.lt);
const isInRange = (key: MultiKey<Key, EntryId>) =>
(options.gte === undefined || this.compareKeys(key, options.gte) >= 0)
&& (options.gt === undefined || this.compareKeys(key, options.gt) > 0)
@ -206,22 +218,25 @@ export class AugmentedTreeMultiMap<Key extends PiledriverObject, Value extends P
if (!child || (options.limit !== undefined && yielded >= options.limit) || !tree.overlaps(child, options)) return;
const node = (await tree.node(child.ref))!;
const visitEntry = function* (entry: Entry<MultiKey<Key, EntryId>, Value>): Iterable<Entry<MultiKey<Key, EntryId>, Value>> {
if (isInRange(entry[0]) && (options.limit === undefined || yielded++ < options.limit)) yield entry;
const visitEntry = async function* (entry: Entry<MultiKey<Key, EntryId>, StoredValue<Value>>): AsyncIterable<Entry<MultiKey<Key, EntryId>, Value>> {
if (isInRange(entry[0]) && (options.limit === undefined || yielded++ < options.limit)) yield [entry[0], await tree.loadValue(entry[1])];
};
const startIndex = lowerBound === undefined ? 0 : tree.lowerBoundIndex(node.entries, lowerBound);
const endIndex = upperBound === undefined ? node.entries.length : tree.lowerBoundIndex(node.entries, upperBound);
if (options.reverse) {
for (let i = node.entries.length - 1; i >= 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);
}
};
@ -246,6 +261,14 @@ export class AugmentedTreeMultiMap<Key extends PiledriverObject, Value extends P
return heapObject ? await heapObject.get() as Node<MultiKey<Key, EntryId>, Value, Augmentation> : null;
}
private storeValue(value: Value): StoredValue<Value> {
return value !== null && typeof value === "object" ? asHeapObject(value) : value;
}
private async loadValue(value: StoredValue<Value>): Promise<Value> {
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 +291,7 @@ export class AugmentedTreeMultiMap<Key extends PiledriverObject, Value extends P
}
// Recomputes all cached metadata for a freshly path-copied node.
private async make(entries: Entry<MultiKey<Key, EntryId>, Value>[], children: Child<MultiKey<Key, EntryId>, Augmentation>[] = []) {
private async make(entries: Entry<MultiKey<Key, EntryId>, StoredValue<Value>>[], children: Child<MultiKey<Key, EntryId>, Augmentation>[] = []) {
if (!entries.length && children.length !== 1) throw new Error("Invalid empty B-tree node");
if (!entries.length) {
const [onlyChild] = children;
@ -292,7 +315,7 @@ export class AugmentedTreeMultiMap<Key extends PiledriverObject, Value extends P
augmentations.push(child.augmentation);
size += child.size;
}
augmentations.push(await this.options.extractAugmentation(entries[i][1], entries[i][0].key, entries[i][0].id));
augmentations.push(await this.options.extractAugmentation(await this.loadValue(entries[i][1]), entries[i][0].key, entries[i][0].id));
}
const lastChild = children.at(entries.length);
if (lastChild) {
@ -323,35 +346,52 @@ export class AugmentedTreeMultiMap<Key extends PiledriverObject, Value extends P
return Math.floor(this.maxEntries() / 2);
}
private search(entries: Entry<MultiKey<Key, EntryId>, Value>[], key: MultiKey<Key, EntryId>) {
private search(entries: Entry<MultiKey<Key, EntryId>, StoredValue<Value>>[], key: MultiKey<Key, EntryId>) {
const low = this.lowerBoundIndex(entries, key);
return { index: low, found: low < entries.length && this.compareKeys(entries[low][0], key) === 0 };
}
private lowerBoundIndex(entries: Entry<MultiKey<Key, EntryId>, StoredValue<Value>>[], key: MultiKey<Key, EntryId>) {
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<Key, EntryId> | undefined, b: MultiKey<Key, EntryId> | undefined) {
if (a === undefined) return b;
if (b === undefined) return a;
return this.compareKeys(a, b) >= 0 ? a : b;
}
private tighterUpperBound(a: MultiKey<Key, EntryId> | undefined, b: MultiKey<Key, EntryId> | 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<MultiKey<Key, EntryId>, Augmentation>, split?: Split<MultiKey<Key, EntryId>, Value, Augmentation> }) {
return result.split ? await this.make([result.split.entry], [result.root, result.split.right]) : result.root;
}
private async split(entries: Entry<MultiKey<Key, EntryId>, Value>[], children: Child<MultiKey<Key, EntryId>, Augmentation>[]) {
private async split(entries: Entry<MultiKey<Key, EntryId>, StoredValue<Value>>[], children: Child<MultiKey<Key, EntryId>, 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<MultiKey<Key, EntryId>, Value>[], children: Child<MultiKey<Key, EntryId>, Augmentation>[], added: boolean) {
private async done(entries: Entry<MultiKey<Key, EntryId>, StoredValue<Value>>[], children: Child<MultiKey<Key, EntryId>, 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<Key, EntryId>, value: Value, replace: boolean): Promise<{ root: Child<MultiKey<Key, EntryId>, Augmentation>, split?: Split<MultiKey<Key, EntryId>, 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 +399,12 @@ export class AugmentedTreeMultiMap<Key extends PiledriverObject, Value extends P
if (found) {
if (!replace) throw new Error("Key already exists");
entries[index] = [key, value];
entries[index] = [key, this.storeValue(value)];
return await this.done(entries, children, false);
}
if (!children.length) {
entries.splice(index, 0, [key, value]);
entries.splice(index, 0, [key, this.storeValue(value)]);
return await this.done(entries, children, true);
}
@ -417,7 +457,7 @@ export class AugmentedTreeMultiMap<Key extends PiledriverObject, Value extends P
return await this.fixChild(entries, children, index, isRoot);
}
private async deleteMin(child: Child<MultiKey<Key, EntryId>, Augmentation>): Promise<{ root: Child<MultiKey<Key, EntryId>, Augmentation> | null, entry: Entry<MultiKey<Key, EntryId>, Value> }> {
private async deleteMin(child: Child<MultiKey<Key, EntryId>, Augmentation>): Promise<{ root: Child<MultiKey<Key, EntryId>, Augmentation> | null, entry: Entry<MultiKey<Key, EntryId>, StoredValue<Value>> }> {
const node = (await this.node(child.ref))!;
const entries = [...node.entries];
const children = [...node.children] as (Child<MultiKey<Key, EntryId>, Augmentation> | null)[];
@ -432,7 +472,7 @@ export class AugmentedTreeMultiMap<Key extends PiledriverObject, Value extends P
return { root: (await this.fixChild(entries, children, 0, false)).root, entry: result.entry };
}
private async deleteMax(child: Child<MultiKey<Key, EntryId>, Augmentation>): Promise<{ root: Child<MultiKey<Key, EntryId>, Augmentation> | null, entry: Entry<MultiKey<Key, EntryId>, Value> }> {
private async deleteMax(child: Child<MultiKey<Key, EntryId>, Augmentation>): Promise<{ root: Child<MultiKey<Key, EntryId>, Augmentation> | null, entry: Entry<MultiKey<Key, EntryId>, StoredValue<Value>> }> {
const node = (await this.node(child.ref))!;
const entries = [...node.entries];
const children = [...node.children] as (Child<MultiKey<Key, EntryId>, Augmentation> | null)[];
@ -448,13 +488,13 @@ export class AugmentedTreeMultiMap<Key extends PiledriverObject, Value extends P
return { root: (await this.fixChild(entries, children, index, false)).root, entry: result.entry };
}
private async afterDelete(entries: Entry<MultiKey<Key, EntryId>, Value>[], children: (Child<MultiKey<Key, EntryId>, Augmentation> | null)[], isRoot: boolean, deleted: boolean) {
private async afterDelete(entries: Entry<MultiKey<Key, EntryId>, StoredValue<Value>>[], children: (Child<MultiKey<Key, EntryId>, 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<MultiKey<Key, EntryId>, Value>[], children: (Child<MultiKey<Key, EntryId>, Augmentation> | null)[], index: number, isRoot: boolean): Promise<{ root: Child<MultiKey<Key, EntryId>, Augmentation> | null, deleted: boolean }> {
private async fixChild(entries: Entry<MultiKey<Key, EntryId>, StoredValue<Value>>[], children: (Child<MultiKey<Key, EntryId>, Augmentation> | null)[], index: number, isRoot: boolean): Promise<{ root: Child<MultiKey<Key, EntryId>, Augmentation> | null, deleted: boolean }> {
const child = children[index];
if (child && child.entryCount >= this.minEntries()) return await this.afterDelete(entries, children, isRoot, true);
@ -491,7 +531,7 @@ export class AugmentedTreeMultiMap<Key extends PiledriverObject, Value extends P
return await this.afterDelete(entries, children, isRoot, true);
}
private async mergeChildren(left: Child<MultiKey<Key, EntryId>, Augmentation> | null, entry: Entry<MultiKey<Key, EntryId>, Value>, right: Child<MultiKey<Key, EntryId>, Augmentation> | null) {
private async mergeChildren(left: Child<MultiKey<Key, EntryId>, Augmentation> | null, entry: Entry<MultiKey<Key, EntryId>, StoredValue<Value>>, right: Child<MultiKey<Key, EntryId>, 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 +558,7 @@ export class AugmentedTreeMultiMap<Key extends PiledriverObject, Value extends P
for (let i = 0; i < node.entries.length; i++) {
result = await this.merge(result, await this.augmentation(node.children[i] ?? null, range));
if (aboveLowerBound(node.entries[i][0]) && belowUpperBound(node.entries[i][0])) {
result = await this.merge(result, await this.options.extractAugmentation(node.entries[i][1], node.entries[i][0].key, node.entries[i][0].id));
result = await this.merge(result, await this.options.extractAugmentation(await this.loadValue(node.entries[i][1]), node.entries[i][0].key, node.entries[i][0].id));
}
}
return await this.merge(result, await this.augmentation(node.children[node.entries.length] ?? null, range));

613
apps/bulldozer-js/src/databases/piledriver/index.ts
View File

@ -1,5 +1,6 @@
import { decodeBase64, encodeBase64 } from "@hexclave/shared/dist/utils/bytes";
import { traceSpan } from "../../otel.js";
import { runAsynchronously } from "@hexclave/shared/dist/utils/promises";
import { traceSpan, traceSpanHot } from "../../otel.js";
import { Database, DatabaseSeq } from "../index.js";
import { LowLevelDatabase, LowLevelDatabaseDebugSnapshot } from "../low-level/index.js";
@ -9,6 +10,9 @@ export type PiledriverHeapObject = {
[isPiledriverHeapObjectSymbol]: true,
};
const textEncoder = new TextEncoder();
const textDecoder = new TextDecoder();
const heapObjectsMapNullSentinel = { __heapObjectsMapNullSentinel: true };
const heapObjectsByObject = new WeakMap<PiledriverObject & object, PiledriverHeapObject>();
/**
@ -66,15 +70,190 @@ 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: ReadonlySet<object>,
heapObjects: ReadonlySet<PiledriverHeapObject>,
// 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<PiledriverHeapObject>;
type PendingHeapInsert = {
buffer: ArrayBuffer,
requiresSeq: DatabaseSeq,
serializationTimingStats: PiledriverSerializationTimingStats | undefined,
resolve: (value: { key: ArrayBuffer, seq: DatabaseSeq }) => void,
reject: (error: unknown) => void,
};
const emptySerializationPath: SerializationPath = { objects: new Set(), heapObjects: new Set() };
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<string, PiledriverSerializationBranchStats>,
heapObjectCacheMissesByShape: Map<string, number>,
heapObjectCacheMissInlineNodeCountsByShape: Map<string, number>,
};
type PiledriverSerializationBranchStats = {
primitiveNodes: number,
arrayNodes: number,
objectNodes: number,
objectEntries: number,
heapReferenceNodes: number,
heapObjectCacheHits: number,
heapObjectCacheMisses: number,
heapObjectCacheMissesByShape: Map<string, number>,
heapObjectCacheMissInlineNodeCountsByShape: Map<string, number>,
};
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<string, number>, key: string) {
map.set(key, (map.get(key) ?? 0) + 1);
}
function addMapCount(map: Map<string, number>, 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<object> = 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] ?? "<root>";
}
function serializationBranchStatsByKey(stats: PiledriverSerializationTimingStats | undefined, branchKey: string) {
if (stats === undefined) return undefined;
let result = stats.branchStats.get(branchKey);
if (result === undefined) {
result = emptyPiledriverSerializationBranchStats();
stats.branchStats.set(branchKey, 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 +264,8 @@ export function declarePiledriverDatabase(lowLevelDb: LowLevelDatabase, options:
const heapObjectsByHeapKeyBase64 = new Map<string, { refIdentity: string, object: WeakRef<PiledriverHeapObject>, seq: DatabaseSeq }>();
const heapObjectsByHeapKeyFinalizer = new FinalizationRegistry(([keyBase64, refIdentity]: [string, string]) => heapObjectsByHeapKeyBase64.get(keyBase64)?.refIdentity === refIdentity && heapObjectsByHeapKeyBase64.delete(keyBase64));
const heapKeysAndSeqByHeapObjects = new WeakMap<PiledriverHeapObject, Promise<{ key: ArrayBuffer, seq: DatabaseSeq }>>();
let pendingHeapInserts: PendingHeapInsert[] = [];
let heapInsertFlushScheduled = false;
const cacheHeapObjectByKey = (key: ArrayBuffer, heapObj: PiledriverHeapObject, seq: DatabaseSeq) => {
const keyBase64 = encodeBase64(new Uint8Array(key));
@ -92,36 +273,122 @@ 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<PiledriverSerializationTimingStats, number>();
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++) {
// 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);
}
};
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();
}
});
};
// 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>): 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) return await existing;
if (existing) {
if (serializationTimingStats !== undefined) {
serializationTimingStats.heapObjectCacheHits++;
const branch = branchKey === undefined ? undefined : serializationBranchStatsByKey(serializationTimingStats, branchKey);
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 = 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 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 childHeapPath = new Set(heapPath).add(heapObj);
return await traceSpanHot("bulldozer-js.piledriver.heap.serializeAndInsert", async () => {
const heapObjectGetStartedAt = performance.now();
const heapObject = await heapObj.get();
if (serializationTimingStats !== undefined) {
const shape = classifyHeapObjectPayload(heapObject);
const inlineNodeCount = totalInlineNodes(countPiledriverInlineNodes(heapObject));
incrementMapCount(serializationTimingStats.heapObjectCacheMissesByShape, shape);
addMapCount(serializationTimingStats.heapObjectCacheMissInlineNodeCountsByShape, shape, inlineNodeCount);
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;
}
const heapObjectSerializeStartedAt = performance.now();
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);
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;
@ -145,7 +412,7 @@ export function declarePiledriverDatabase(lowLevelDb: LowLevelDatabase, options:
let loadPromise: Promise<PiledriverObject> | 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);
@ -167,104 +434,202 @@ export function declarePiledriverDatabase(lowLevelDb: LowLevelDatabase, options:
return { object: heapObj, seq };
};
const serializePiledriverObjectToJsonableObject = async (obj: PiledriverObject, path: SerializationPath): Promise<{ jsonableObject: unknown, seq: DatabaseSeq }> => {
switch (typeof obj) {
case "number": {
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": {
return { jsonableObject: obj, seq: lowLevelDb.initialSeq };
}
case "object": {
if (obj === null) {
return { jsonableObject: obj, seq: lowLevelDb.initialSeq };
} else if (Array.isArray(obj)) {
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)));
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);
return {
jsonableObject: ["heap-reference", encodeBase64(new Uint8Array(heapKeyAndSeq.key))],
seq: heapKeyAndSeq.seq,
};
} else {
// "normal" object
// TODO: assert this is a POJO
// 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,
};
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));
return {
jsonableObject: Object.fromEntries(entriesSerializeResults.map(([k, v]) => [k, v.jsonableObject] as const)),
seq: lowLevelDb.combineSeqs(...entriesSerializeResults.map(([_, v]) => v.seq)),
};
// 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<object>();
// 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++;
}
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++;
}
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<string, unknown> = {};
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;
}
}
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): Promise<{ buffer: ArrayBuffer, seq: DatabaseSeq }> => {
const toJsonableResponse = await serializePiledriverObjectToJsonableObject(obj, path);
return {
buffer: new TextEncoder().encode(JSON.stringify(toJsonableResponse.jsonableObject)).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<PiledriverHeapObject, PendingHeapReferenceSlot[]>();
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<string, PiledriverObject> = {};
for (const [k, v] of Object.entries(jsonableObject)) {
result[k] = deserializePiledriverObjectFromJsonableObject(v, enclosingSeq, seqs);
}
return result;
}
}
default: {
@ -274,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) => {
@ -311,8 +678,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, new Set(), 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 };
});
},

15
apps/bulldozer-js/src/otel.ts
View File

@ -28,3 +28,18 @@ export async function traceSpan<T>(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<T>(optionsOrDescription: string | { description: string, attributes?: Record<string, TraceAttributeValue> }, fn: (span: TraceSpan) => Promise<T>): Promise<T> {
if (!hotPathTracingEnabled) return await fn(noopTraceSpan);
return await traceSpan(optionsOrDescription, fn);
}

2
apps/bulldozer-js/src/payments/schema/index.ts
View File

@ -668,7 +668,7 @@ export function createPaymentsSchema() {
const subscription = rowObject<SubscriptionRow>(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 => {