spalladino · Aug 25, 2026
diff --git 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
@@ -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
@@ -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<T>(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
@@ -186,9 +186,15 @@ function fromHexString<T extends BaseField>(buf: string, f: DerivedField<T>) {
     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
@@ -42,25 +42,48 @@ export function optional<T extends ZodTypeAny>(schema: T) {
 
 type ToJsonIs<T, TRet> = T extends { toJSON(): TRet } ? T : never;
 
+type HexHydratable = { fromString(str: string): any } | { fromBuffer(buf: Buffer): any };
+
+type HexInstanceOf<TClass> = TClass extends
+  | { fromString(str: string): infer TInstance }
+  | { fromBuffer(buf: Buffer): infer TInstance }
+  ? ToJsonIs<TInstance, string>
+  : 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<TClass extends { fromString(str: string): any } | { fromBuffer(buf: Buffer): any }>(
+export function hexSchemaFor<TClass extends HexHydratable>(
   klazz: TClass,
   refinement?: (input: string) => boolean,
-): ZodType<
-  TClass extends { fromString(str: string): infer TInstance } | { fromBuffer(buf: Buffer): infer TInstance }
-    ? ToJsonIs<TInstance, string>
-    : 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<HexInstanceOf<TClass>, 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));
+    }
+  });
 }
 
 /**