fix(p2p): gossip tx validation stalls behind tx pool finalization (A-1656)

release-image/Dockerfile.dockerignore

@@ -18,6 +18,7 @@
 !/yarn-project/noir-protocol-circuits-types/artifacts/
 !/yarn-project/protocol-contracts/artifacts/
 !/yarn-project/standard-contracts/artifacts/
+!/yarn-project/standard-contracts/artifacts-historical/
 !/yarn-project/noir-contracts.js/artifacts/
 !/yarn-project/noir-test-contracts.js/artifacts/
 !/yarn-project/simulator/artifacts/

yarn-project/bb-prover/src/verifier/batch_chonk_verifier.ts

@@ -6,6 +6,7 @@ import { Timer } from '@aztec/foundation/timer';
 import { ProtocolCircuitVks } from '@aztec/noir-protocol-circuits-types/server/vks';
 import type { ClientProtocolCircuitVerifier, IVCProofVerificationResult } from '@aztec/stdlib/interfaces/server';
 import type { Tx } from '@aztec/stdlib/tx';
+import { getTelemetryClient } from '@aztec/telemetry-client';
 
 import { Unpackr } from 'msgpackr';
 import { execFile } from 'node:child_process';
@@ -16,6 +17,7 @@ import * as path from 'node:path';
 import { promisify } from 'node:util';
 
 import type { BBConfig } from '../config.js';
+import { IVCVerifierMetrics } from './queued_chonk_verifier.js';
 
 const execFileAsync = promisify(execFile);
 
@@ -54,6 +56,7 @@ export class BatchChonkVerifier implements ClientProtocolCircuitVerifier {
   private sendQueue: SerialQueue;
   private fifoReader: FifoFrameReader;
   private logger = createLogger('bb-prover:batch_chonk_verifier');
+  private metrics: IVCVerifierMetrics;
   /** Maps artifact name to VK index in the batch verifier. */
   private vkIndexMap = new Map<string, number>();
   /** Bound cleanup handler for process exit signals. */
@@ -69,6 +72,7 @@ export class BatchChonkVerifier implements ClientProtocolCircuitVerifier {
     this.fifoReader = new FifoFrameReader();
     this.sendQueue = new SerialQueue();
     this.sendQueue.start(1);
+    this.metrics = new IVCVerifierMetrics(getTelemetryClient(), `BatchChonkVerifier-${label}`);
   }
 
   /** Create and start a BatchChonkVerifier using the protocol circuit VKs. */
@@ -237,6 +241,7 @@ export class BatchChonkVerifier implements ClientProtocolCircuitVerifier {
     const totalDurationMs = pending.totalTimer.ms();
 
     const ivcResult: IVCProofVerificationResult = { valid, durationMs, totalDurationMs };
+    this.metrics.recordIVCVerification(ivcResult);
 
     if (!valid) {
       this.logger.warn(`Proof verification failed for request_id=${result.request_id}: ${result.error_message}`);

yarn-project/bb-prover/src/verifier/queued_chonk_verifier.ts

@@ -16,7 +16,8 @@ import {
 
 import { createHistogram } from 'node:perf_hooks';
 
-class IVCVerifierMetrics {
+/** Records verification timing and failure metrics for an IVC (chonk) proof verifier. */
+export class IVCVerifierMetrics {
   private ivcVerificationHistogram: Histogram;
   private ivcTotalVerificationHistogram: Histogram;
   private ivcFailureCount: UpDownCounter;

yarn-project/end-to-end/src/spartan/n_tps.test.ts

@@ -38,6 +38,7 @@ import {
   getRPCEndpoint,
   hasDeployedHelmRelease,
   installChaosMeshChart,
+  logGossipTxValidationMetrics,
   setupEnvironment,
   startPortForwardForPrometeheus,
   uninstallChaosMesh,
@@ -127,6 +128,15 @@ describe('sustained N TPS test', () => {
 
   afterAll(async () => {
     logger.info('Collecting benchmark metrics and cleaning up...');
+
+    // Log the gossip tx validation breakdown (per-stage timings, tx pool queue stats, chonk verifier
+    // timings) so slow gossip validations observed during the run can be attributed to a stage.
+    if (prometheusClient) {
+      await logGossipTxValidationMetrics(prometheusClient, config.NAMESPACE, TEST_DURATION_SECONDS + 60, logger).catch(
+        err => logger.warn(`Failed to scrape gossip validation metrics: ${err}`, { err }),
+      );
+    }
+
     if (process.env.BENCH_OUTPUT) {
       for (const topic of Object.values(TopicType)) {
         try {

yarn-project/end-to-end/src/spartan/n_tps_prove.test.ts

@@ -35,6 +35,7 @@ import {
   type ServiceEndpoint,
   getEthereumEndpoint,
   getExternalIP,
+  logGossipTxValidationMetrics,
   setupEnvironment,
   startPortForwardForPrometeheus,
 } from './utils.js';
@@ -145,6 +146,15 @@ describe(`prove ${TARGET_TPS}TPS test`, () => {
         server: new URL(`http://127.0.0.1:${freshPromForward.port}`),
       });
 
+      // Log the gossip tx validation breakdown (per-stage timings, tx pool queue stats, chonk
+      // verifier timings) so slow gossip validations during the run can be attributed to a stage.
+      await logGossipTxValidationMetrics(
+        prometheusClient,
+        config.NAMESPACE,
+        epochDurationSeconds + SLOTS_BUFFER * slotDurationSeconds,
+        logger,
+      ).catch(err => logger.warn(`Failed to scrape gossip validation metrics: ${err}`, { err }));
+
       const endSnapshot = await captureMetricsSnapshot(prometheusClient, logger);
 
       // Helper to compute delta, clamping negative values to 0 (handles pod restarts)

yarn-project/end-to-end/src/spartan/utils/gossip_metrics.ts

@@ -0,0 +1,108 @@
+import type { Logger } from '@aztec/foundation/log';
+
+import type { PrometheusClient } from '../../quality_of_service/prometheus_client.js';
+
+/** Runs a PromQL query and returns each series as a label-record plus value. Returns [] on error. */
+async function queryVector(
+  prometheus: PrometheusClient,
+  query: string,
+  logger: Logger,
+): Promise<{ labels: Record<string, string>; value: number }[]> {
+  try {
+    const resp = await prometheus.queryRaw(query);
+    if (resp.status !== 'success' || resp.data.resultType !== 'vector') {
+      logger.warn(`Unexpected Prometheus response for query`, { query, resp });
+      return [];
+    }
+    return resp.data.result.map(({ metric, value }) => ({
+      labels: (metric ?? {}) as Record<string, string>,
+      value: parseFloat(value[1]),
+    }));
+  } catch (err) {
+    logger.warn(`Failed to run Prometheus query: ${err}`, { query });
+    return [];
+  }
+}
+
+/** Formats a grouped vector result as a { labelValue: value } record for structured logging. */
+function toRecord(series: { labels: Record<string, string>; value: number }[], label: string): Record<string, number> {
+  const out: Record<string, number> = {};
+  for (const { labels, value } of series) {
+    out[labels[label] ?? 'unknown'] = Math.round(value * 100) / 100;
+  }
+  return out;
+}
+
+/**
+ * Scrapes and logs gossip tx validation timing breakdowns from Prometheus: per-stage validation
+ * durations, tx pool serial queue wait/execution times, and chonk (IVC) proof verifier timings.
+ * Used by the spartan TPS benchmarks to attribute slow gossip validations to a specific stage.
+ */
+export async function logGossipTxValidationMetrics(
+  prometheus: PrometheusClient,
+  namespace: string,
+  windowSeconds: number,
+  logger: Logger,
+): Promise<void> {
+  const ns = `k8s_namespace_name="${namespace}"`;
+  const window = `[${Math.max(60, Math.ceil(windowSeconds))}s]`;
+
+  const stageQuantile = (perc: string) =>
+    `histogram_quantile(${perc}, sum(rate(aztec_p2p_gossip_tx_validation_stage_duration_milliseconds_bucket{${ns}}${window})) by (le, aztec_p2p_tx_validation_stage))`;
+  const stageAvg = () =>
+    `sum(rate(aztec_p2p_gossip_tx_validation_stage_duration_milliseconds_sum{${ns}}${window})) by (aztec_p2p_tx_validation_stage) / ` +
+    `sum(rate(aztec_p2p_gossip_tx_validation_stage_duration_milliseconds_count{${ns}}${window})) by (aztec_p2p_tx_validation_stage)`;
+  const validationQuantile = (perc: string) =>
+    `histogram_quantile(${perc}, sum(rate(aztec_p2p_gossip_message_validation_duration_milliseconds_bucket{${ns}}${window})) by (le, aztec_gossip_topic_name))`;
+  const queueQuantile = (metric: string, perc: string) =>
+    `topk(10, histogram_quantile(${perc}, sum(rate(${metric}{${ns}}${window})) by (le, aztec_mempool_operation)))`;
+  const ivcQuantile = (metric: string, perc: string) =>
+    `histogram_quantile(${perc}, sum(rate(${metric}{${ns}}${window})) by (le))`;
+
+  const stageLabel = 'aztec_p2p_tx_validation_stage';
+  const topicLabel = 'aztec_gossip_topic_name';
+  const operationLabel = 'aztec_mempool_operation';
+
+  const [stageP50, stageP95, stageAvgs, validationP95, slowCount] = await Promise.all([
+    queryVector(prometheus, stageQuantile('0.50'), logger),
+    queryVector(prometheus, stageQuantile('0.95'), logger),
+    queryVector(prometheus, stageAvg(), logger),
+    queryVector(prometheus, validationQuantile('0.95'), logger),
+    queryVector(prometheus, `sum(aztec_p2p_gossip_slow_validation_count{${ns}}) by (${topicLabel})`, logger),
+  ]);
+
+  logger.info('Gossip tx validation stage timings (ms)', {
+    stageP50: toRecord(stageP50, stageLabel),
+    stageP95: toRecord(stageP95, stageLabel),
+    stageAvg: toRecord(stageAvgs, stageLabel),
+    validationP95ByTopic: toRecord(validationP95, topicLabel),
+    slowValidationCountByTopic: toRecord(slowCount, topicLabel),
+  });
+
+  const [queueWaitP95, queueExecutionP95, queueLengthMax] = await Promise.all([
+    queryVector(prometheus, queueQuantile('aztec_mempool_tx_pool_v2_queue_wait_milliseconds_bucket', '0.95'), logger),
+    queryVector(
+      prometheus,
+      queueQuantile('aztec_mempool_tx_pool_v2_queue_execution_milliseconds_bucket', '0.95'),
+      logger,
+    ),
+    queryVector(prometheus, `max(max_over_time(aztec_mempool_tx_pool_v2_queue_length{${ns}}${window}))`, logger),
+  ]);
+
+  logger.info('Tx pool serial queue stats (ms)', {
+    queueWaitP95: toRecord(queueWaitP95, operationLabel),
+    queueExecutionP95: toRecord(queueExecutionP95, operationLabel),
+    queueLengthMax: queueLengthMax[0]?.value,
+  });
+
+  const [ivcVerifyP95, ivcTotalP95] = await Promise.all([
+    queryVector(prometheus, ivcQuantile('aztec_ivc_verifier_time_milliseconds_bucket', '0.95'), logger),
+    queryVector(prometheus, ivcQuantile('aztec_ivc_verifier_total_time_milliseconds_bucket', '0.95'), logger),
+  ]);
+
+  // A large gap between total (queue + verify) and verify indicates a pile-up in the verifier queue.
+  logger.info('Chonk (IVC) proof verifier timings (ms)', {
+    verifyP95: ivcVerifyP95[0]?.value,
+    totalP95: ivcTotalP95[0]?.value,
+  });
+}

yarn-project/end-to-end/src/spartan/utils/index.ts

@@ -69,3 +69,6 @@ export { ChainHealth, type ChainHealthSnapshot } from './health.js';
 
 // Pod log extraction
 export { type BlockBuiltLogEntry, fetchBlockBuiltLogs } from './pod_logs.js';
+
+// Gossip validation metrics scraping
+export { logGossipTxValidationMetrics } from './gossip_metrics.js';

yarn-project/p2p/src/mem_pools/tx_pool_v2/archive/tx_archive.ts

@@ -46,8 +46,19 @@ export class TxArchive {
    * Archives transactions, stripping their proofs.
    * Evicts oldest transactions if the limit is exceeded.
    */
-  async archiveTxs(txs: Tx[]): Promise<void> {
-    if (!this.isEnabled() || txs.length === 0) {
+  archiveTxs(txs: Tx[]): Promise<void> {
+    return this.archiveTxBuffers(
+      txs.map(tx => ({ txHash: tx.getTxHash().toString(), buffer: tx.withoutProof().toBuffer() })),
+    );
+  }
+
+  /**
+   * Archives already-serialized proof-less tx buffers, avoiding any deserialization. This is the
+   * hot path used at finalization time, where the pool already stores txs proof-stripped.
+   * Evicts oldest transactions if the limit is exceeded.
+   */
+  async archiveTxBuffers(entries: { txHash: string; buffer: Buffer }[]): Promise<void> {
+    if (!this.isEnabled() || entries.length === 0) {
       return;
     }
 
@@ -57,7 +68,15 @@ export class TxArchive {
         let headIdx = await this.getHeadIndex();
         let tailIdx = await this.getTailIndex();
 
-        for (const tx of txs) {
+        for (const { txHash, buffer } of entries) {
+          // Skip txs that are already archived. Re-archiving (a retried finalization, or a crash
+          // between archiving and deleting) would append a second FIFO index entry pointing at the
+          // same stored value; evicting the older entry would then delete the value out from under
+          // the newer one.
+          if (await this.#txs.hasAsync(txHash)) {
+            continue;
+          }
+
           // Evict oldest entries if at capacity
           while (headIdx - tailIdx >= this.#limit) {
             const txHashToEvict = await this.#indices.getAsync(tailIdx);
@@ -68,15 +87,12 @@ export class TxArchive {
             tailIdx++;
           }
 
-          // Archive the transaction with stripped proof
-          const archivedTx = tx.withoutProof();
-          const txHash = tx.getTxHash().toString();
-          await this.#txs.set(txHash, archivedTx.toBuffer());
+          await this.#txs.set(txHash, buffer);
           await this.#indices.set(headIdx, txHash);
           headIdx++;
         }
 
-        this.#log.debug(`Archived ${txs.length} txs, total: ${headIdx - tailIdx}`);
+        this.#log.debug(`Archived ${entries.length} txs, total: ${headIdx - tailIdx}`);
       });
     } catch (error) {
       this.#log.error('Error archiving transactions', { error });
@@ -100,13 +116,22 @@ export class TxArchive {
     return head - tail;
   }
 
+  // NOTE: these must consume the iterator via for-await so the underlying LMDB cursor is closed.
+  // Calling .next() once and abandoning the generator leaks a cursor slot: inside a write
+  // transaction the committed-state iterator is unbounded (the limit is applied by the wrapper),
+  // and the abandoned generator never runs its finally block, so CLOSE_CURSOR is never sent. The
+  // store has maxReaders - 1 cursor slots; leaking them deadlocks every later iteration on the store.
   private async getHeadIndex(): Promise<number> {
-    const entry = await this.#indices.entriesAsync({ limit: 1, reverse: true }).next();
-    return (entry.value?.[0] ?? -1) + 1;
+    for await (const [index] of this.#indices.entriesAsync({ limit: 1, reverse: true })) {
+      return index + 1;
+    }
+    return 0;
   }
 
   private async getTailIndex(): Promise<number> {
-    const entry = await this.#indices.entriesAsync({ limit: 1 }).next();
-    return entry.value?.[0] ?? 0;
+    for await (const [index] of this.#indices.entriesAsync({ limit: 1 })) {
+      return index;
+    }
+    return 0;
   }
 }

yarn-project/p2p/src/mem_pools/tx_pool_v2/instrumentation.ts

@@ -1,5 +1,6 @@
 import {
   Attributes,
+  type Histogram,
   type Meter,
   Metrics,
   type ObservableGauge,
@@ -67,3 +68,29 @@ export class TxPoolV2Instrumentation {
     this.#missingPreviouslyEvictedCounter.add(count);
   }
 }
+
+/**
+ * Instrumentation for the tx pool serial queue: how long operations wait behind other queued work,
+ * how long they take to execute once running, and the current queue depth. All pool operations share
+ * a single serial queue, so contention here directly delays gossip tx validation.
+ */
+export class TxPoolQueueInstrumentation {
+  #queueWait: Histogram;
+  #queueExecution: Histogram;
+
+  constructor(telemetry: TelemetryClient, getQueueLength: () => number) {
+    const meter: Meter = telemetry.getMeter('TxPoolQueue');
+    this.#queueWait = meter.createHistogram(Metrics.MEMPOOL_TX_POOL_V2_QUEUE_WAIT);
+    this.#queueExecution = meter.createHistogram(Metrics.MEMPOOL_TX_POOL_V2_QUEUE_EXECUTION);
+    const queueLength = meter.createObservableGauge(Metrics.MEMPOOL_TX_POOL_V2_QUEUE_LENGTH);
+    queueLength.addCallback((result: ObservableResult) => {
+      result.observe(getQueueLength());
+    });
+  }
+
+  record(operation: string, waitMs: number, executionMs: number) {
+    const attributes = { [Attributes.MEMPOOL_OPERATION]: operation };
+    this.#queueWait.record(Math.ceil(waitMs), attributes);
+    this.#queueExecution.record(Math.ceil(executionMs), attributes);
+  }
+}