From e20c006e8b85ab9c7b8ca5b972b476df9148099a Mon Sep 17 00:00:00 2001 From: Santiago Palladino Date: Tue, 25 Aug 2026 16:33:48 -0300 Subject: [PATCH] fix(archiver): speed up log-by-tag queries Tag log queries fan the per-tag scans out 8 wide instead of walking tags one at a time, and bounded scans now read a single cursor page instead of paging in tens of entries at a time. `SINGLE_PAGE_LIMIT` caps how large a bounded scan may be before it goes back to paging. Supporting changes on the read path: an empty write batch short circuits to a plain read, `Fr.fromHexString` rejects over-long input on byte length rather than after parsing, and `hexSchemaFor` runs a single transform instead of chaining two. The queries stay inside `db.transactionAsync`, so they serialize with queued writes by design and cannot return a torn result. `node_rpc_perf` gains the production query shape (100 tags per call, real tags, a reference block anchor, `includeEffects` on) plus a probe that measures how much a query queues behind five concurrent siblings. --- yarn-project/archiver/src/store/log_store.ts | 23 +++- .../store/log_store_write_contention.test.ts | 98 +++++++++++++++++ .../src/bench/node_rpc_perf.test.ts | 103 +++++++++++++++++- .../foundation/src/curves/bn254/field.ts | 10 +- yarn-project/foundation/src/schemas/utils.ts | 49 ++++++--- yarn-project/kv-store/src/lmdb-v2/message.ts | 7 ++ .../kv-store/src/lmdb-v2/read_transaction.ts | 8 +- .../kv-store/src/lmdb-v2/write_transaction.ts | 16 +++ 8 files changed, 291 insertions(+), 23 deletions(-) create mode 100644 yarn-project/archiver/src/store/log_store_write_contention.test.ts diff --git a/yarn-project/archiver/src/store/log_store.ts b/yarn-project/archiver/src/store/log_store.ts index 490f84eb072..b9dde9f02c6 100644 --- a/yarn-project/archiver/src/store/log_store.ts +++ b/yarn-project/archiver/src/store/log_store.ts @@ -1,4 +1,5 @@ import { INITIAL_L2_BLOCK_NUM } from '@aztec/constants'; +import { asyncPool } from '@aztec/foundation/async-pool'; import { Fr } from '@aztec/foundation/curves/bn254'; import { createLogger } from '@aztec/foundation/log'; import type { AztecAsyncKVStore, AztecAsyncMap } from '@aztec/kv-store'; @@ -29,6 +30,9 @@ import { tagHexForLog, } from './log_store_codec.js'; +/** How many per-tag range scans a single tag query runs concurrently. */ +const TAG_SCAN_CONCURRENCY = 8; + /** * Indexes every emitted private and public log under a composite hex-string key * `[contractAddress (public only)]-tag-blockNumber-txIndexWithinBlock-logIndexWithinTx`, @@ -43,6 +47,12 @@ import { * scan by block (block isn't the leading key segment). * * Contract-class logs are no longer stored or served by the log store. + * + * Tag queries run inside `db.transactionAsync` so the `referenceBlock` reorg check and every per-tag scan see the same + * snapshot of the store and cannot return a torn result. On lmdb-v2 that helper routes its callback through the store's + * single serial writer queue, so a query waits for any queued write (notably block ingestion) to commit first; that + * cost is accepted in exchange for the consistency. The per-tag scans within one query still run concurrently + * ({@link TAG_SCAN_CONCURRENCY} at a time), each on its own cursor over the transaction's snapshot. */ export class LogStore { /** Primary map: composite private key (tag + tail = 96 hex chars + separators) -> serialized {@link StoredLogValue}. */ @@ -254,8 +264,12 @@ export class LogStore { const fromBlock = query.fromBlock ?? INITIAL_L2_BLOCK_NUM; const includeEffects = query.includeEffects === true; - const perTagResults: LogResult[][] = []; - for (const tagEntry of tags) { + const limit = query.limitPerTag ?? MAX_LOGS_PER_TAG; + + // Each scan opens its own native cursor over the transaction's snapshot, so they overlap instead of queueing. The + // pool bound stays well under the store's concurrent-cursor budget (maxReaders - 1) so other components can still + // read while a query is in flight. + const perTagResults = await asyncPool(TAG_SCAN_CONCURRENCY, [...tags], async tagEntry => { const { tagHex, afterLog } = normalizeTagEntry(tagEntry); const prefix = contractHex !== undefined ? encodePublicPrefix(contractHex, tagHex) : tagHex; @@ -275,7 +289,6 @@ export class LogStore { start = encodeKey(prefix, fromBlock, 0, 0); } - const limit = query.limitPerTag ?? MAX_LOGS_PER_TAG; const out: LogResult[] = []; for await (const [rawKey, rawVal] of primaryMap.entriesAsync({ start, end, limit })) { const tail = decodeKeyTail(rawKey); @@ -290,8 +303,8 @@ export class LogStore { logIndexWithinTx: tail.logIndexWithinTx, }); } - perTagResults.push(out); - } + return out; + }); if (includeEffects) { // Dedupe by txHash across the entire page so a tx with many tagged logs costs one fetch. diff --git a/yarn-project/archiver/src/store/log_store_write_contention.test.ts b/yarn-project/archiver/src/store/log_store_write_contention.test.ts new file mode 100644 index 00000000000..b39d57c636b --- /dev/null +++ b/yarn-project/archiver/src/store/log_store_write_contention.test.ts @@ -0,0 +1,98 @@ +import { createLogger } from '@aztec/foundation/log'; +import { sleep } from '@aztec/foundation/sleep'; +import { Timer } from '@aztec/foundation/timer'; +import type { AztecAsyncKVStore } from '@aztec/kv-store'; +import { openTmpStore } from '@aztec/kv-store/lmdb-v2'; +import { GENESIS_BLOCK_HEADER_HASH } from '@aztec/stdlib/block'; +import type { PublishedCheckpoint } from '@aztec/stdlib/checkpoint'; +import { SiloedTag } from '@aztec/stdlib/logs'; +import type { AppendOnlyTreeSnapshot } from '@aztec/stdlib/trees'; + +import { jest } from '@jest/globals'; + +import { makeCheckpointWithLogs } from '../test/mock_structs.js'; +import { BlockStore } from './block_store.js'; +import { LogStore } from './log_store.js'; + +const BLOCKS_TO_SEED = 5; +const TXS_PER_BLOCK = 4; +const LOGS_PER_TX = 5; +const TAGS_PER_QUERY = 100; + +/** Number of artificial write transactions queued in front of the contended read. */ +const QUEUED_WRITES = 5; +/** Duration of each artificial write transaction. */ +const WRITE_DURATION_MS = 50; + +describe('LogStore write contention', () => { + jest.setTimeout(60_000); + + const logger = createLogger('archiver:log_store_write_contention_test'); + + let db: AztecAsyncKVStore; + let blockStore: BlockStore; + let logStore: LogStore; + let tags: SiloedTag[]; + + beforeEach(async () => { + db = await openTmpStore('log_store_write_contention_test'); + blockStore = new BlockStore(db); + logStore = new LogStore(db, blockStore, GENESIS_BLOCK_HEADER_HASH); + + const checkpoints: PublishedCheckpoint[] = []; + let previousArchive: AppendOnlyTreeSnapshot | undefined; + for (let blockNumber = 1; blockNumber <= BLOCKS_TO_SEED; blockNumber++) { + const ckpt = await makeCheckpointWithLogs(blockNumber, { + numTxsPerBlock: TXS_PER_BLOCK, + privateLogs: { numLogsPerTx: LOGS_PER_TX }, + previousArchive, + }); + previousArchive = ckpt.checkpoint.blocks[0].archive; + checkpoints.push(ckpt); + } + + const blocks = checkpoints.map(c => c.checkpoint.blocks[0]); + await blockStore.addCheckpoints(checkpoints); + await logStore.addLogs(blocks); + + const harvested: SiloedTag[] = []; + for (const block of blocks) { + for (const txEffect of block.body.txEffects) { + for (const log of txEffect.privateLogs) { + harvested.push(new SiloedTag(log.fields[0])); + } + } + } + tags = Array.from({ length: TAGS_PER_QUERY }, (_, i) => harvested[i % harvested.length]); + }); + + afterEach(async () => { + await db.close(); + }); + + it('returns the same results for a tag query racing queued write transactions', async () => { + const baselineTimer = new Timer(); + const baseline = await logStore.getPrivateLogsByTags({ tags }); + const baselineMs = baselineTimer.ms(); + + // Fill the store's serial writer queue without awaiting it. Tag queries are routed through that same queue, so + // the contended query below only resolves once these commit. + const writes = Array.from({ length: QUEUED_WRITES }, () => db.transactionAsync(() => sleep(WRITE_DURATION_MS))); + + const contendedTimer = new Timer(); + const contended = await logStore.getPrivateLogsByTags({ tags }); + const contendedMs = contendedTimer.ms(); + + logger.info(`Baseline read ${baselineMs.toFixed(2)}ms, contended read ${contendedMs.toFixed(2)}ms`, { + baselineMs, + contendedMs, + queuedWriteMs: QUEUED_WRITES * WRITE_DURATION_MS, + }); + + expect(contended.length).toBe(TAGS_PER_QUERY); + expect(contended.every(logs => logs.length > 0)).toBe(true); + expect(contended).toEqual(baseline); + + await Promise.all(writes); + }); +}); diff --git a/yarn-project/end-to-end/src/bench/node_rpc_perf.test.ts b/yarn-project/end-to-end/src/bench/node_rpc_perf.test.ts index 098e033ad63..9ea5124db96 100644 --- a/yarn-project/end-to-end/src/bench/node_rpc_perf.test.ts +++ b/yarn-project/end-to-end/src/bench/node_rpc_perf.test.ts @@ -39,6 +39,12 @@ const BENCHMARK_ITERATIONS_SLOW = 5; /** Number of blocks to build before benchmarking */ const BLOCKS_TO_BUILD = 5; +/** Tags per call for the batched log queries, matching the production PXE limit (MAX_RPC_LEN) */ +const TAGS_PER_CALL = 100; + +/** Number of log queries kept in flight when measuring how much a query queues behind its siblings */ +const CONCURRENT_LOG_QUERIES = 5; + /** Result structure for benchmark data */ interface BenchmarkResult { name: string; @@ -72,6 +78,18 @@ function calculateStats(timings: number[]): TimingStats { }; } +/** + * Builds a list of exactly `size` tags by cycling through the harvested ones, falling back to random tags when + * nothing was harvested. + */ +function cycleTags(size: number, source: T[], makeRandom: () => T): T[] { + const tags: T[] = []; + for (let i = 0; i < size; i++) { + tags.push(source.length > 0 ? source[i % source.length] : makeRandom()); + } + return tags; +} + /** * Run a benchmark for a given async function */ @@ -118,6 +136,9 @@ describe('e2e_node_rpc_perf', () => { const txHashes: TxHash[] = []; let tokenContract: TokenContract; let sampleTx: Tx; // A sample proven tx for benchmarking simulation/validation APIs + let referenceBlock: BlockHash; + const privateTags: SiloedTag[] = []; + const publicTags: Tag[] = []; afterEach(() => { jest.restoreAllMocks(); @@ -179,6 +200,8 @@ describe('e2e_node_rpc_perf', () => { const block = await aztecNode.getBlock(BlockNumber(blockNumber)); blockArchive = block!.header.lastArchive.root; + await harvestTags(); + // Create a sample tx for benchmarking simulation/validation APIs logger.info('Creating sample tx for simulation/validation benchmarks...'); sampleTx = await proveInteraction(wallet, tokenContract.methods.mint_to_public(ownerAddress, 1n), { @@ -193,7 +216,7 @@ describe('e2e_node_rpc_perf', () => { const mintAmount = 100n; for (let block = 0; block < BLOCKS_TO_BUILD; block++) { - const provenTx = await proveInteraction(wallet, tokenContract.methods.mint_to_public(ownerAddress, mintAmount), { + const provenTx = await proveInteraction(wallet, tokenContract.methods.mint_to_private(ownerAddress, mintAmount), { from: ownerAddress, }); @@ -204,6 +227,26 @@ describe('e2e_node_rpc_perf', () => { } } + async function harvestTags() { + const blocks = await aztecNode.getBlocks(BlockNumber(1), blockNumber, { includeTransactions: true }); + for (const block of blocks) { + for (const txEffect of block.body.txEffects) { + for (const log of txEffect.privateLogs) { + privateTags.push(new SiloedTag(log.fields[0])); + } + for (const log of txEffect.publicLogs) { + if (log.contractAddress.equals(contractAddress) && log.fields.length > 0) { + publicTags.push(new Tag(log.fields[0])); + } + } + } + } + referenceBlock = blocks[blocks.length - 1].hash; + logger.info( + `Harvested ${privateTags.length} private and ${publicTags.length} public tags from ${blocks.length} blocks`, + ); + } + function addResult(name: string, stats: TimingStats, unit = 'ms') { benchmarkResults.push({ name: `node_rpc/${name}/avg`, @@ -579,6 +622,64 @@ describe('e2e_node_rpc_perf', () => { addResult('getPublicLogsByTags', stats); expect(stats.avg).toBeLessThan(3000); }); + + it('benchmarks getPrivateLogsByTags with 100 tags (recipient sync shape)', async () => { + // The recipient-sync shape measured at the node boundary: ~100 tags per call, includeEffects on, and a + // near-total miss rate (0.45% of tags matched anything) — one matching tag among 100 keeps the hit path + // covered without over-weighting response serialization. + const tags = [ + privateTags[0] ?? SiloedTag.random(), + ...Array.from({ length: TAGS_PER_CALL - 1 }, () => SiloedTag.random()), + ]; + const { stats } = await benchmark( + 'getPrivateLogsByTags_100tags', + () => aztecNode.getPrivateLogsByTags({ tags, referenceBlock, includeEffects: true }), + BENCHMARK_ITERATIONS_FAST, + ); + addResult('getPrivateLogsByTags_100tags', stats); + expect(stats.avg).toBeLessThan(3000); + }); + + it('benchmarks getPrivateLogsByTags with 100 matching tags', async () => { + const tags = cycleTags(TAGS_PER_CALL, privateTags, () => SiloedTag.random()); + const { stats } = await benchmark( + 'getPrivateLogsByTags_100tags_allhits', + () => aztecNode.getPrivateLogsByTags({ tags, referenceBlock, includeEffects: true }), + BENCHMARK_ITERATIONS_FAST, + ); + addResult('getPrivateLogsByTags_100tags_allhits', stats); + expect(stats.avg).toBeLessThan(3000); + }); + + it('benchmarks getPublicLogsByTags with 100 tags', async () => { + const tags = cycleTags(TAGS_PER_CALL, publicTags, () => Tag.random()); + const { stats } = await benchmark( + 'getPublicLogsByTags_100tags', + () => aztecNode.getPublicLogsByTags({ contractAddress, tags, referenceBlock }), + BENCHMARK_ITERATIONS_FAST, + ); + addResult('getPublicLogsByTags_100tags', stats); + expect(stats.avg).toBeLessThan(3000); + }); + + it('benchmarks getPrivateLogsByTags queued behind concurrent calls', async () => { + // Issue CONCURRENT_LOG_QUERIES calls without awaiting them, then time one more issued right after. If the + // store serializes these against each other, the measured latency grows with the queue depth. + const timings: number[] = []; + for (let i = 0; i < BENCHMARK_ITERATIONS_FAST; i++) { + const tags = cycleTags(TAGS_PER_CALL, privateTags, () => SiloedTag.random()); + const inFlight = Array.from({ length: CONCURRENT_LOG_QUERIES }, () => + aztecNode.getPrivateLogsByTags({ tags, referenceBlock, includeEffects: true }), + ); + const timer = new Timer(); + await aztecNode.getPrivateLogsByTags({ tags, referenceBlock, includeEffects: true }); + timings.push(timer.ms()); + await Promise.all(inFlight); + } + const stats = calculateStats(timings); + addResult('getPrivateLogsByTags_100tags_queued', stats); + expect(stats.avg).toBeLessThan(3000); + }); }); describe('validator APIs', () => { diff --git a/yarn-project/foundation/src/curves/bn254/field.ts b/yarn-project/foundation/src/curves/bn254/field.ts index 715d6816b71..e7fcfc06118 100644 --- a/yarn-project/foundation/src/curves/bn254/field.ts +++ b/yarn-project/foundation/src/curves/bn254/field.ts @@ -186,9 +186,15 @@ function fromHexString(buf: string, f: DerivedField) { throw new Error(`Invalid hex-encoded string: "${buf}"`); } - const buffer = Buffer.from(checked.length % 2 === 1 ? '0' + checked : checked, 'hex'); + // Going straight to a bigint rather than through a Buffer makes this ~5x faster, which matters when + // deserializing thousands of field elements from JSON-RPC responses. The explicit length check preserves + // the rejection of over-long values that the Buffer-based constructor used to perform. + const byteLength = Math.ceil(checked.length / 2); + if (byteLength > BaseField.SIZE_IN_BYTES) { + throw new Error(`Value length ${byteLength} exceeds ${BaseField.SIZE_IN_BYTES}`); + } - return new f(toBigIntBE(buffer)); + return new f(BigInt(`0x${checked}`)); } /** diff --git a/yarn-project/foundation/src/schemas/utils.ts b/yarn-project/foundation/src/schemas/utils.ts index 0ba2fa4f72b..5a802395127 100644 --- a/yarn-project/foundation/src/schemas/utils.ts +++ b/yarn-project/foundation/src/schemas/utils.ts @@ -42,25 +42,48 @@ export function optional(schema: T) { type ToJsonIs = T extends { toJSON(): TRet } ? T : never; +type HexHydratable = { fromString(str: string): any } | { fromBuffer(buf: Buffer): any }; + +type HexInstanceOf = TClass extends + | { fromString(str: string): infer TInstance } + | { fromBuffer(buf: Buffer): infer TInstance } + ? ToJsonIs + : never; + /** - * Creates a schema that accepts a hex string and uses it to hydrate an instance. + * Creates a schema that accepts a hex string and uses it to hydrate an instance. Validation and hydration run in + * a single zod transform (rather than a chain of refinements): each extra zod layer is measurable when parsing + * thousands of field elements from a JSON-RPC response. * @param klazz - Class that implements either fromString or fromBuffer. * @returns A schema for the class. */ -export function hexSchemaFor( +export function hexSchemaFor( klazz: TClass, refinement?: (input: string) => boolean, -): ZodType< - TClass extends { fromString(str: string): infer TInstance } | { fromBuffer(buf: Buffer): infer TInstance } - ? ToJsonIs - : never, - string -> { - const stringSchema = refinement ? z.string().refine(refinement, `Not a valid instance`) : z.string(); - const hexSchema = stringSchema.refine(isHex, 'Not a valid hex string'); - return 'fromString' in klazz - ? hexSchema.transform(klazz.fromString.bind(klazz)) - : hexSchema.transform(str => Buffer.from(withoutHexPrefix(str), 'hex')).transform(klazz.fromBuffer.bind(klazz)); +): ZodType, string> { + const construct = + 'fromString' in klazz + ? (str: string) => klazz.fromString(str) + : (str: string) => klazz.fromBuffer(Buffer.from(withoutHexPrefix(str), 'hex')); + return z.string().transform((str, ctx) => { + const fail = (message: string) => { + ctx.addIssue({ code: z.ZodIssueCode.custom, message }); + return z.NEVER; + }; + if (refinement && !refinement(str)) { + return fail(`Not a valid instance`); + } + // isHex is checked even though the class does its own parsing: several fromString implementations are laxer + // than hex (Fr.fromString also accepts decimal strings), and the wire format must not silently accept those. + if (!isHex(str)) { + return fail('Not a valid hex string'); + } + try { + return construct(str); + } catch (err) { + return fail(err instanceof Error ? err.message : String(err)); + } + }); } /** diff --git a/yarn-project/kv-store/src/lmdb-v2/message.ts b/yarn-project/kv-store/src/lmdb-v2/message.ts index 8c165936b1d..e3d2b10f710 100644 --- a/yarn-project/kv-store/src/lmdb-v2/message.ts +++ b/yarn-project/kv-store/src/lmdb-v2/message.ts @@ -7,6 +7,13 @@ export const CURSOR_PAGE_SIZE = 10; /** Keys per GET message when a batched read is split into chunks; bounds message size and how long one read holds its tx. */ export const DEFAULT_GET_CHUNK_SIZE = 1024; +/** + * Largest `limit` that an iteration will ask the native side to return in a single page. Bounded scans up to this + * size are answered by one START_CURSOR round trip with no cursor left open, so they neither pay for + * ADVANCE_CURSOR/CLOSE_CURSOR nor hold one of the store's limited cursor slots. + */ +export const SINGLE_PAGE_LIMIT = 128; + export enum LMDBMessageType { OPEN_DATABASE = 100, GET, diff --git a/yarn-project/kv-store/src/lmdb-v2/read_transaction.ts b/yarn-project/kv-store/src/lmdb-v2/read_transaction.ts index ce3ef9d3fac..b3808ef261e 100644 --- a/yarn-project/kv-store/src/lmdb-v2/read_transaction.ts +++ b/yarn-project/kv-store/src/lmdb-v2/read_transaction.ts @@ -7,6 +7,7 @@ import { Database, type LMDBMessageChannel, LMDBMessageType, + SINGLE_PAGE_LIMIT, } from './message.js'; /** @@ -120,11 +121,14 @@ export class ReadTransaction { let cursor: number | undefined; try { + // A bounded scan asks for the whole limit up front: `endKey` is filtered client side but in-range entries are + // contiguous from `startKey`, so the first `limit` entries always contain every entry we may return. + const singlePageLimit = typeof limit === 'number' && limit <= SINGLE_PAGE_LIMIT ? limit : undefined; const response = await this.channel.sendMessage(LMDBMessageType.START_CURSOR, { key: startKey, reverse, - count: typeof limit === 'number' ? Math.min(limit, CURSOR_PAGE_SIZE) : CURSOR_PAGE_SIZE, - onePage: typeof limit === 'number' && limit < CURSOR_PAGE_SIZE, + count: singlePageLimit ?? CURSOR_PAGE_SIZE, + onePage: singlePageLimit !== undefined, db, txId: this.txId ?? null, }); diff --git a/yarn-project/kv-store/src/lmdb-v2/write_transaction.ts b/yarn-project/kv-store/src/lmdb-v2/write_transaction.ts index e5170019418..2ab9a8f0ae2 100644 --- a/yarn-project/kv-store/src/lmdb-v2/write_transaction.ts +++ b/yarn-project/kv-store/src/lmdb-v2/write_transaction.ts @@ -202,6 +202,13 @@ export class WriteTransaction extends ReadTransaction { reverse?: boolean, limit?: number, ): AsyncIterable<[Uint8Array, Uint8Array]> { + if (isEmptyBatch(this.dataBatch)) { + // Nothing pending can shadow or suppress a committed entry, so the committed data can be served verbatim and + // `limit` can be pushed down to the cursor. With a non-empty batch that is unsafe: pending removals within the + // first `limit` committed entries would require reading past it. + yield* super.iterate(startKey, endKey, reverse, limit); + return; + } yield* this.#iterate( super.iterate(startKey, endKey, reverse), this.dataBatch, @@ -220,6 +227,11 @@ export class WriteTransaction extends ReadTransaction { reverse?: boolean, limit?: number, ): AsyncIterable<[Uint8Array, Uint8Array[]]> { + if (isEmptyBatch(this.indexBatch)) { + // See the note in `iterate`. + yield* super.iterateIndex(startKey, endKey, reverse, limit); + return; + } yield* this.#iterate( super.iterateIndex(startKey, endKey, reverse), this.indexBatch, @@ -348,3 +360,7 @@ export class WriteTransaction extends ReadTransaction { }); } } + +function isEmptyBatch(batch: Batch): boolean { + return batch.addEntries.length === 0 && batch.removeEntries.length === 0; +}