diff --git a/release-image/Dockerfile.dockerignore b/release-image/Dockerfile.dockerignore index edafd9d97055..078a10aa92be 100644 --- a/release-image/Dockerfile.dockerignore +++ b/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/ diff --git a/yarn-project/bb-prover/src/verifier/batch_chonk_verifier.ts b/yarn-project/bb-prover/src/verifier/batch_chonk_verifier.ts index b64811b9240c..6a0d51621c37 100644 --- a/yarn-project/bb-prover/src/verifier/batch_chonk_verifier.ts +++ b/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(); /** 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}`); diff --git a/yarn-project/bb-prover/src/verifier/queued_chonk_verifier.ts b/yarn-project/bb-prover/src/verifier/queued_chonk_verifier.ts index c7d62565fa6a..7fa09b6a2161 100644 --- a/yarn-project/bb-prover/src/verifier/queued_chonk_verifier.ts +++ b/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; diff --git a/yarn-project/end-to-end/src/spartan/n_tps.test.ts b/yarn-project/end-to-end/src/spartan/n_tps.test.ts index 3dfe0a0d94de..ecaa0f1fee74 100644 --- a/yarn-project/end-to-end/src/spartan/n_tps.test.ts +++ b/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 { diff --git a/yarn-project/end-to-end/src/spartan/n_tps_prove.test.ts b/yarn-project/end-to-end/src/spartan/n_tps_prove.test.ts index 89d091ee3ef7..7186c9b3fabb 100644 --- a/yarn-project/end-to-end/src/spartan/n_tps_prove.test.ts +++ b/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) diff --git a/yarn-project/end-to-end/src/spartan/utils/gossip_metrics.ts b/yarn-project/end-to-end/src/spartan/utils/gossip_metrics.ts new file mode 100644 index 000000000000..f151b072afc7 --- /dev/null +++ b/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; 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, + 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; value: number }[], label: string): Record { + const out: Record = {}; + 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 { + 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, + }); +} diff --git a/yarn-project/end-to-end/src/spartan/utils/index.ts b/yarn-project/end-to-end/src/spartan/utils/index.ts index c9a05844c184..8d88da5a8147 100644 --- a/yarn-project/end-to-end/src/spartan/utils/index.ts +++ b/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'; diff --git a/yarn-project/p2p/src/mem_pools/tx_pool_v2/archive/tx_archive.ts b/yarn-project/p2p/src/mem_pools/tx_pool_v2/archive/tx_archive.ts index baf7686356cc..172552a042f2 100644 --- a/yarn-project/p2p/src/mem_pools/tx_pool_v2/archive/tx_archive.ts +++ b/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 { - if (!this.isEnabled() || txs.length === 0) { + archiveTxs(txs: Tx[]): Promise { + 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 { + 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 { - 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 { - 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; } } diff --git a/yarn-project/p2p/src/mem_pools/tx_pool_v2/instrumentation.ts b/yarn-project/p2p/src/mem_pools/tx_pool_v2/instrumentation.ts index 6ec711bb826c..9dac77baecf0 100644 --- a/yarn-project/p2p/src/mem_pools/tx_pool_v2/instrumentation.ts +++ b/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); + } +} diff --git a/yarn-project/p2p/src/mem_pools/tx_pool_v2/tx_pool_v2.ts b/yarn-project/p2p/src/mem_pools/tx_pool_v2/tx_pool_v2.ts index b7094ec40b0f..ad4cfe75f4d2 100644 --- a/yarn-project/p2p/src/mem_pools/tx_pool_v2/tx_pool_v2.ts +++ b/yarn-project/p2p/src/mem_pools/tx_pool_v2/tx_pool_v2.ts @@ -1,7 +1,8 @@ import { SlotNumber } from '@aztec/foundation/branded-types'; +import { chunk } from '@aztec/foundation/collection'; import { type Logger, createLogger } from '@aztec/foundation/log'; import { SerialQueue } from '@aztec/foundation/queue'; -import { DateProvider } from '@aztec/foundation/timer'; +import { DateProvider, Timer } from '@aztec/foundation/timer'; import type { TypedEventEmitter } from '@aztec/foundation/types'; import type { AztecAsyncKVStore } from '@aztec/kv-store'; import type { L2Block, L2BlockId } from '@aztec/stdlib/block'; @@ -11,6 +12,7 @@ import { type TelemetryClient, getTelemetryClient } from '@aztec/telemetry-clien import EventEmitter from 'node:events'; import { PoolInstrumentation, PoolName } from '../instrumentation.js'; +import { TxPoolQueueInstrumentation } from './instrumentation.js'; import type { AddTxsResult, PoolReadAccess, @@ -21,7 +23,7 @@ import type { } from './interfaces.js'; import type { TxState } from './tx_metadata.js'; import type { MinedTxInfo } from './tx_pool_v2_impl.js'; -import { TxPoolV2Impl } from './tx_pool_v2_impl.js'; +import { FINALIZE_BLOCK_CHUNK_SIZE, TxPoolV2Impl } from './tx_pool_v2_impl.js'; /** * Implementation of TxPoolV2 with explicit state management. @@ -31,6 +33,9 @@ import { TxPoolV2Impl } from './tx_pool_v2_impl.js'; */ export class AztecKVTxPoolV2 extends (EventEmitter as new () => TypedEventEmitter) implements TxPoolV2 { #queue: SerialQueue; + #queueMetrics: TxPoolQueueInstrumentation; + /** Chains finalizations so their chunked queue items never interleave with each other. */ + #finalizationChain: Promise = Promise.resolve(); #impl: TxPoolV2Impl; #metrics?: PoolInstrumentation; #store: AztecAsyncKVStore; @@ -53,6 +58,7 @@ export class AztecKVTxPoolV2 extends (EventEmitter as new () => TypedEventEmitte this.#telemetry = telemetry; this.#log = log; this.#queue = new SerialQueue(); + this.#queueMetrics = new TxPoolQueueInstrumentation(telemetry, () => this.#queue.length()); // Create callbacks that the impl uses to notify us about events and metrics const callbacks = { @@ -84,106 +90,148 @@ export class AztecKVTxPoolV2 extends (EventEmitter as new () => TypedEventEmitte // PUBLIC API - All methods queue to the implementation // ============================================================================ + /** + * Enqueues an operation on the serial queue, recording how long it waited behind other queued + * work and how long it took to execute once running. + */ + #run(operation: string, fn: () => T | Promise): Promise> { + const waitTimer = new Timer(); + return this.#queue.put(async () => { + const waitMs = waitTimer.ms(); + const executionTimer = new Timer(); + try { + return await fn(); + } finally { + this.#queueMetrics.record(operation, waitMs, executionTimer.ms()); + } + }); + } + // === Core Operations === addPendingTxs(txs: Tx[], opts: { source?: string; feeComparisonOnly?: boolean } = {}): Promise { - return this.#queue.put(() => this.#impl.addPendingTxs(txs, opts)); + return this.#run('addPendingTxs', () => this.#impl.addPendingTxs(txs, opts)); } canAddPendingTx(tx: Tx): Promise<'accepted' | 'ignored'> { - return this.#queue.put(() => this.#impl.canAddPendingTx(tx)); + return this.#run('canAddPendingTx', () => this.#impl.canAddPendingTx(tx)); } addProtectedTxs(txs: Tx[], block: BlockHeader, opts: { source?: string } = {}): Promise { - return this.#queue.put(() => this.#impl.addProtectedTxs(txs, block, opts)); + return this.#run('addProtectedTxs', () => this.#impl.addProtectedTxs(txs, block, opts)); } protectTxs(txHashes: TxHash[], block: BlockHeader): Promise { - return this.#queue.put(() => this.#impl.protectTxs(txHashes, block)); + return this.#run('protectTxs', () => this.#impl.protectTxs(txHashes, block)); } addMinedTxs(txs: Tx[], block: BlockHeader, opts: { source?: string } = {}): Promise { - return this.#queue.put(() => this.#impl.addMinedTxs(txs, block, opts)); + return this.#run('addMinedTxs', () => this.#impl.addMinedTxs(txs, block, opts)); } // === State Transition Handlers === handleMinedBlock(block: L2Block): Promise { - return this.#queue.put(() => this.#impl.handleMinedBlock(block)); + return this.#run('handleMinedBlock', () => this.#impl.handleMinedBlock(block)); } prepareForSlot(slotNumber: SlotNumber): Promise { - return this.#queue.put(() => this.#impl.prepareForSlot(slotNumber)); + return this.#run('prepareForSlot', () => this.#impl.prepareForSlot(slotNumber)); } unprotectTxs(txHashes: TxHash[], slotNumber: SlotNumber): Promise { - return this.#queue.put(() => this.#impl.unprotectTxs(txHashes, slotNumber)); + return this.#run('unprotectTxs', () => this.#impl.unprotectTxs(txHashes, slotNumber)); } handlePrunedBlocks(latestBlock: L2BlockId, options?: { deleteAllTxs?: boolean }): Promise { - return this.#queue.put(() => this.#impl.handlePrunedBlocks(latestBlock, options)); + return this.#run('handlePrunedBlocks', () => this.#impl.handlePrunedBlocks(latestBlock, options)); } handleFailedExecution(txHashes: TxHash[]): Promise { - return this.#queue.put(() => this.#impl.handleFailedExecution(txHashes)); + return this.#run('handleFailedExecution', () => this.#impl.handleFailedExecution(txHashes)); } + /** + * Handles a finalized block by archiving and deleting the mined txs it finalizes. The work is + * split into chunk-sized serial-queue items rather than one long-running item, so gossip-driven + * pool operations (canAddPendingTx / addPendingTxs) interleave with finalization instead of + * stalling behind an entire epoch's worth of tx processing. Finalizations are chained so two + * concurrent calls never interleave their chunks with each other. + */ handleFinalizedBlock(block: BlockHeader): Promise { - return this.#queue.put(() => this.#impl.handleFinalizedBlock(block)); + const run = this.#finalizationChain.then(() => this.#handleFinalizedBlock(block)); + this.#finalizationChain = run.catch(() => {}); + return run; + } + + async #handleFinalizedBlock(block: BlockHeader): Promise { + const { cutoffBlock, txHashes } = await this.#run('prepareFinalization', () => + this.#impl.prepareFinalization(block), + ); + const batches = chunk(txHashes, FINALIZE_BLOCK_CHUNK_SIZE); + for (const batch of batches) { + await this.#run('archiveFinalizedTxs', () => this.#impl.archiveFinalizedTxs(batch)); + } + for (const batch of batches) { + await this.#run('deleteFinalizedTxs', () => this.#impl.deleteFinalizedTxs(batch, cutoffBlock)); + } + await this.#run('completeFinalization', () => + this.#impl.completeFinalization(txHashes, cutoffBlock, block.globalVariables.blockNumber), + ); } // === Queries === getTxByHash(txHash: TxHash, opts?: { includeProof?: boolean }): Promise { - return this.#queue.put(() => this.#impl.getTxByHash(txHash, opts)); + return this.#run('getTxByHash', () => this.#impl.getTxByHash(txHash, opts)); } getTxsByHash(txHashes: TxHash[], opts?: { includeProof?: boolean }): Promise<(Tx | undefined)[]> { - return this.#queue.put(() => this.#impl.getTxsByHash(txHashes, opts)); + return this.#run('getTxsByHash', () => this.#impl.getTxsByHash(txHashes, opts)); } hasTxs(txHashes: TxHash[]): Promise { - return this.#queue.put(() => this.#impl.hasTxs(txHashes)); + return this.#run('hasTxs', () => this.#impl.hasTxs(txHashes)); } getTxStatus(txHash: TxHash): Promise { - return this.#queue.put(() => Promise.resolve(this.#impl.getTxStatus(txHash))); + return this.#run('getTxStatus', () => this.#impl.getTxStatus(txHash)); } getPendingTxHashes(): Promise { - return this.#queue.put(() => Promise.resolve(this.#impl.getPendingTxHashes())); + return this.#run('getPendingTxHashes', () => this.#impl.getPendingTxHashes()); } getEligiblePendingTxHashes(): Promise { - return this.#queue.put(() => Promise.resolve(this.#impl.getEligiblePendingTxHashes())); + return this.#run('getEligiblePendingTxHashes', () => this.#impl.getEligiblePendingTxHashes()); } getPendingTxCount(): Promise { - return this.#queue.put(() => Promise.resolve(this.#impl.getPendingTxCount())); + return this.#run('getPendingTxCount', () => this.#impl.getPendingTxCount()); } hasEligiblePendingTxs(minCount: number): Promise { - return this.#queue.put(() => Promise.resolve(this.#impl.hasEligiblePendingTxs(minCount))); + return this.#run('hasEligiblePendingTxs', () => this.#impl.hasEligiblePendingTxs(minCount)); } getMinedTxHashes(): Promise<[TxHash, L2BlockId][]> { - return this.#queue.put(() => Promise.resolve(this.#impl.getMinedTxHashes())); + return this.#run('getMinedTxHashes', () => this.#impl.getMinedTxHashes()); } getMinedTxCount(): Promise { - return this.#queue.put(() => Promise.resolve(this.#impl.getMinedTxCount())); + return this.#run('getMinedTxCount', () => this.#impl.getMinedTxCount()); } isEmpty(): Promise { - return this.#queue.put(() => Promise.resolve(this.#impl.isEmpty())); + return this.#run('isEmpty', () => this.#impl.isEmpty()); } getArchivedTxByHash(txHash: TxHash): Promise { - return this.#queue.put(() => this.#impl.getArchivedTxByHash(txHash)); + return this.#run('getArchivedTxByHash', () => this.#impl.getArchivedTxByHash(txHash)); } getLowestPriorityPending(limit: number): Promise { - return this.#queue.put(() => Promise.resolve(this.#impl.getLowestPriorityPending(limit))); + return this.#run('getLowestPriorityPending', () => this.#impl.getLowestPriorityPending(limit)); } /** Returns read-only access to the pool. Used for testing. */ diff --git a/yarn-project/p2p/src/mem_pools/tx_pool_v2/tx_pool_v2_finalization.test.ts b/yarn-project/p2p/src/mem_pools/tx_pool_v2/tx_pool_v2_finalization.test.ts new file mode 100644 index 000000000000..7bb27b0d8768 --- /dev/null +++ b/yarn-project/p2p/src/mem_pools/tx_pool_v2/tx_pool_v2_finalization.test.ts @@ -0,0 +1,193 @@ +import { BlockNumber, CheckpointNumber, IndexWithinCheckpoint, SlotNumber } from '@aztec/foundation/branded-types'; +import { Fr } from '@aztec/foundation/curves/bn254'; +import { createLogger } from '@aztec/foundation/log'; +import { Timer } from '@aztec/foundation/timer'; +import { openTmpStore } from '@aztec/kv-store/lmdb-v2'; +import { RevertCode } from '@aztec/stdlib/avm'; +import { AztecAddress } from '@aztec/stdlib/aztec-address'; +import { Body, L2Block, type L2BlockSource } from '@aztec/stdlib/block'; +import { GasFees } from '@aztec/stdlib/gas'; +import type { MerkleTreeReadOperations, WorldStateSynchronizer } from '@aztec/stdlib/interfaces/server'; +import { mockTx } from '@aztec/stdlib/testing'; +import { + AppendOnlyTreeSnapshot, + MerkleTreeId, + PublicDataTreeLeaf, + PublicDataTreeLeafPreimage, +} from '@aztec/stdlib/trees'; +import { BlockHeader, GlobalVariables, type Tx, TxEffect, type TxValidator } from '@aztec/stdlib/tx'; + +import { jest } from '@jest/globals'; +import { type MockProxy, mock } from 'jest-mock-extended'; + +import type { TxMetaData } from './tx_metadata.js'; +import { AztecKVTxPoolV2 } from './tx_pool_v2.js'; + +const alwaysValidValidator: TxValidator = { + validateTx: () => Promise.resolve({ result: 'valid' }), +}; + +jest.setTimeout(300_000); + +// Reproduction for A-1656: gossip tx validation on mainnet stalls for 10-40s whenever the pool +// finalizes an epoch's worth of mined txs. Finalization runs as a single serial-queue item, so a +// concurrent canAddPendingTx / addPendingTxs (issued by gossip validation) waits for the whole +// bulk operation to complete. This test measures that wait and bounds it. +describe('TxPoolV2 finalization stall', () => { + const logger = createLogger('p2p:tx_pool_v2:finalization_stall_test'); + + // Matches the tx counts seen finalized per epoch tick on mainnet (~100-200). + const MINED_TX_COUNT = 128; + // Upper bound for how long a gossip-validation pool operation may wait behind maintenance work. + const MAX_STALL_MS = 1_000; + + const feePayers = [ + AztecAddress.fromBigIntUnsafe(1n), + AztecAddress.fromBigIntUnsafe(2n), + AztecAddress.fromBigIntUnsafe(3n), + ]; + + let mockL2BlockSource: MockProxy; + let mockWorldState: MockProxy; + let db: MockProxy; + let minedTxs: Tx[]; + let incomingTx: Tx; + + const makeHeader = (n: number) => + BlockHeader.empty({ + globalVariables: GlobalVariables.empty({ + blockNumber: BlockNumber(n), + slotNumber: SlotNumber(n), + }), + }); + + const makeBlock = (txs: Tx[], header: BlockHeader): L2Block => { + const txEffects = txs.map(tx => { + const nullifiers = tx.data.getNonEmptyNullifiers(); + return new TxEffect(RevertCode.OK, tx.getTxHash(), Fr.ZERO, [], nullifiers, [], [], [], [], []); + }); + const body = new Body(txEffects); + const archive = new AppendOnlyTreeSnapshot(Fr.random(), header.globalVariables.blockNumber + 1); + return new L2Block( + archive, + header, + body, + CheckpointNumber(Number(header.globalVariables.blockNumber)), + IndexWithinCheckpoint(0), + ); + }; + + // Large public calldata mimics the complex public-heavy txs seen on mainnet when the stalls occurred. + const createTxBatch = (count: number, startSeed = 1): Promise => + Promise.all( + Array.from({ length: count }, (_, i) => + mockTx((startSeed + i) * 100, { + publicCalldataSize: 1000, + maxPriorityFeesPerGas: new GasFees(((startSeed + i) % 100) + 1, ((startSeed + i) % 100) + 1), + feePayer: feePayers[(startSeed + i) % feePayers.length], + }), + ), + ); + + const createPool = async (archivedTxLimit: number) => { + const store = await openTmpStore('p2p-finalization-stall'); + const archiveStore = await openTmpStore('archive-finalization-stall'); + const pool = new AztecKVTxPoolV2( + store, + archiveStore, + { + l2BlockSource: mockL2BlockSource, + worldStateSynchronizer: mockWorldState, + createTxValidator: () => Promise.resolve(alwaysValidValidator), + checkAllowedSetupCalls: () => Promise.resolve(true), + blockMinFeesProvider: { getCurrentMinFees: () => Promise.resolve(GasFees.empty()) }, + }, + undefined, + { archivedTxLimit }, + ); + await pool.start(); + const cleanup = async () => { + await pool.stop(); + await store.delete(); + await archiveStore.delete(); + }; + return { pool, cleanup }; + }; + + beforeAll(async () => { + minedTxs = await createTxBatch(MINED_TX_COUNT); + [incomingTx] = await createTxBatch(1, (MINED_TX_COUNT + 1) * 100); + logger.info(`Created ${MINED_TX_COUNT} mined txs of ${minedTxs[0].toBuffer().length} bytes each`); + }); + + beforeEach(() => { + mockL2BlockSource = mock(); + mockL2BlockSource.getTxEffect.mockResolvedValue(undefined); + + mockWorldState = mock(); + db = mock(); + mockWorldState.getCommitted.mockReturnValue(db); + mockWorldState.getSnapshot.mockReturnValue(db); + db.getPreviousValueIndex.mockImplementation((_tree, slot) => + Promise.resolve({ index: slot, alreadyPresent: true }), + ); + db.getLeafPreimage.mockImplementation((tree, index) => + Promise.resolve( + tree === MerkleTreeId.PUBLIC_DATA_TREE + ? new PublicDataTreeLeafPreimage( + new PublicDataTreeLeaf(new Fr(index), new Fr(BigInt('1000000000000000000000000'))), + Fr.ONE, + 1n, + ) + : undefined, + ), + ); + db.findLeafIndices.mockImplementation((_tree, leaves) => + Promise.resolve((leaves as Fr[]).map((_, i) => BigInt(i + 1))), + ); + }); + + // Fills the pool with mined txs, then kicks off finalization and immediately issues the pool + // calls that gossip validation depends on. Returns how long each waited and the finalize time. + const measureStallDuringFinalization = async (archivedTxLimit: number) => { + const { pool, cleanup } = await createPool(archivedTxLimit); + try { + await pool.addPendingTxs(minedTxs); + await pool.handleMinedBlock(makeBlock(minedTxs, makeHeader(1))); + + const finalizeTimer = new Timer(); + const finalizePromise = pool.handleFinalizedBlock(makeHeader(1)); + + const precheckTimer = new Timer(); + const precheckPromise = pool.canAddPendingTx(incomingTx).then(() => precheckTimer.ms()); + + const addTimer = new Timer(); + const addPromise = pool.addPendingTxs([incomingTx], { source: 'gossip' }).then(() => addTimer.ms()); + + const [precheckMs, addMs] = await Promise.all([precheckPromise, addPromise]); + await finalizePromise; + const finalizeMs = finalizeTimer.ms(); + + logger.info( + `Finalized ${MINED_TX_COUNT} txs in ${Math.round(finalizeMs)}ms with archivedTxLimit=${archivedTxLimit} ` + + `(canAddPendingTx waited ${Math.round(precheckMs)}ms, addPendingTxs waited ${Math.round(addMs)}ms)`, + { finalizeMs, precheckMs, addMs, archivedTxLimit }, + ); + return { finalizeMs, precheckMs, addMs }; + } finally { + await cleanup(); + } + }; + + it('does not stall gossip pool operations while finalizing mined txs with archiving disabled', async () => { + const { precheckMs, addMs } = await measureStallDuringFinalization(0); + expect(precheckMs).toBeLessThan(MAX_STALL_MS); + expect(addMs).toBeLessThan(MAX_STALL_MS); + }); + + it('does not stall gossip pool operations while finalizing mined txs with archiving enabled', async () => { + const { precheckMs, addMs } = await measureStallDuringFinalization(10_000); + expect(precheckMs).toBeLessThan(MAX_STALL_MS); + expect(addMs).toBeLessThan(MAX_STALL_MS); + }); +}); diff --git a/yarn-project/p2p/src/mem_pools/tx_pool_v2/tx_pool_v2_impl.ts b/yarn-project/p2p/src/mem_pools/tx_pool_v2/tx_pool_v2_impl.ts index ddd014c49550..15b31242e77e 100644 --- a/yarn-project/p2p/src/mem_pools/tx_pool_v2/tx_pool_v2_impl.ts +++ b/yarn-project/p2p/src/mem_pools/tx_pool_v2/tx_pool_v2_impl.ts @@ -42,11 +42,12 @@ import { type TxMetaData, type TxState, buildTxMetaData, checkNullifierConflict import { TxPoolIndices } from './tx_pool_indices.js'; /** - * Maximum number of full transactions to load into memory at once when finalizing a block. - * Bounds peak memory while archiving and hard-deleting mined txs (~23k txs/epoch at 10 TPS would - * otherwise OOM the node). + * Maximum number of finalized txs to archive or hard-delete per serial-queue item. Bounds both + * peak memory (~23k txs/epoch at 10 TPS would otherwise OOM the node) and, more importantly, how + * long finalization occupies the pool's serial queue per item: gossip tx validation waits on that + * queue, so each item must stay well under the gossipsub mcache eviction window. */ -const FINALIZE_BLOCK_CHUNK_SIZE = 100; +export const FINALIZE_BLOCK_CHUNK_SIZE = 16; /** * Callbacks for the implementation to notify the outer class about events and metrics. @@ -725,45 +726,71 @@ export class TxPoolV2Impl { this.#log.info(`Deleted ${txHashes.length} failed txs`, { txHashes: txHashes.map(h => h.toString()) }); } - async handleFinalizedBlock(block: BlockHeader): Promise { - const blockNumber = block.globalVariables.blockNumber; - + /** + * Resolves what a finalized block event should process: the cutoff block and the mined txs at or + * before it. The wrapper feeds the result to archiveFinalizedTxs / finalizeTxs in chunks, each as + * its own serial-queue item, so gossip-driven pool operations can interleave with finalization + * instead of stalling behind an entire epoch's worth of tx processing. + */ + async prepareFinalization(block: BlockHeader): Promise<{ cutoffBlock: BlockNumber; txHashes: string[] }> { // Hold finalized txs for a configurable margin behind the finalized tip so a prover still // proving an epoch with already-finalized blocks isn't starved of its txs. 0 deletes at the finalized tip. const cutoffBlock = await this.#finalizationCutoffBlock(block); + return { cutoffBlock, txHashes: this.#indices.findTxsMinedAtOrBefore(cutoffBlock) }; + } - // Step 1: Find mined txs at or before the cutoff block - const minedTxsToFinalize = this.#indices.findTxsMinedAtOrBefore(cutoffBlock); - - // Step 2: Archive in chunks if archiving is enabled. Hydrating an entire epoch's worth of - // mined txs at once would OOM under load. When archiving is disabled there is no need to hydrate the txs at all. - if (this.#archive.isEnabled()) { - for (let i = 0; i < minedTxsToFinalize.length; i += FINALIZE_BLOCK_CHUNK_SIZE) { - const chunk = minedTxsToFinalize.slice(i, i + FINALIZE_BLOCK_CHUNK_SIZE); - const txsToArchive: Tx[] = []; - for (const txHashStr of chunk) { - const buffer = await this.#txsDB.getAsync(txHashStr); - if (buffer) { - txsToArchive.push(Tx.fromBuffer(buffer)); - } - } - if (txsToArchive.length > 0) { - await this.#archive.archiveTxs(txsToArchive); - } + /** + * Copies the given finalized txs into the archive. The pool stores txs proof-stripped, which is + * exactly the archive format, so buffers are copied as-is without deserialization. + */ + async archiveFinalizedTxs(txHashes: string[]): Promise { + if (!this.#archive.isEnabled()) { + return; + } + const entries: { txHash: string; buffer: Buffer }[] = []; + for (const txHash of txHashes) { + const buffer = await this.#txsDB.getAsync(txHash); + if (buffer) { + entries.push({ txHash, buffer }); } } + await this.#archive.archiveTxBuffers(entries); + } + + /** + * Deletes a batch of finalized mined txs from the active pool. Callers may invoke this in + * chunks: a crash between chunks leaves some finalized txs mined, and the next finalized-block + * event lists and deletes them again (the operation is idempotent), so per-chunk atomicity is + * sufficient. Since other pool operations may interleave between the finalization plan being + * computed and this call, each tx is re-checked to still be mined at or before the cutoff. + */ + async deleteFinalizedTxs(txHashes: string[], cutoffBlock: BlockNumber): Promise { + const stillFinalized = txHashes.filter(txHash => { + const minedBlock = this.#indices.getMetadata(txHash)?.minedL2BlockId?.number; + return minedBlock !== undefined && minedBlock <= cutoffBlock; + }); + if (stillFinalized.length === 0) { + return; + } + await this.#store.transactionAsync(async () => { + await this.#deleteTxsBatch(stillFinalized); + }); + } - // Step 3: Delete mined txs from the active pool and finalize soft-deleted txs in one - // transaction. Only tx hashes are touched here, so memory is bounded and atomicity is preserved. + /** Finalizes soft-deleted txs up to the cutoff and logs the completed finalization. */ + async completeFinalization( + txHashes: string[], + cutoffBlock: BlockNumber, + finalizedBlockNumber: BlockNumber, + ): Promise { await this.#store.transactionAsync(async () => { - await this.#deleteTxsBatch(minedTxsToFinalize); await this.#deletedPool.finalizeBlock(cutoffBlock); }); - if (minedTxsToFinalize.length > 0) { - this.#log.info(`Finalized ${minedTxsToFinalize.length} mined txs from blocks up to ${cutoffBlock}`, { - txHashes: minedTxsToFinalize, - finalizedBlockNumber: blockNumber, + if (txHashes.length > 0) { + this.#log.info(`Finalized ${txHashes.length} mined txs from blocks up to ${cutoffBlock}`, { + txHashes, + finalizedBlockNumber, cutoffBlock, }); } diff --git a/yarn-project/p2p/src/services/libp2p/instrumentation.ts b/yarn-project/p2p/src/services/libp2p/instrumentation.ts index 7ec5e0a9a894..c0be0c07f1ee 100644 --- a/yarn-project/p2p/src/services/libp2p/instrumentation.ts +++ b/yarn-project/p2p/src/services/libp2p/instrumentation.ts @@ -19,6 +19,7 @@ export class P2PInstrumentation { private messageLatency: Histogram; private txReceivedCount: UpDownCounter; private slowValidationCount: UpDownCounter; + private txValidationStageDuration: Histogram; private aggLatencyHisto = new Map(); private aggValidationHisto = new Map(); @@ -47,6 +48,8 @@ export class P2PInstrumentation { this.messageLatency = meter.createHistogram(Metrics.P2P_GOSSIP_MESSAGE_LATENCY); + this.txValidationStageDuration = meter.createHistogram(Metrics.P2P_GOSSIP_TX_VALIDATION_STAGE_DURATION); + this.txReceivedCount = createUpDownCounterWithDefault(meter, Metrics.P2P_GOSSIP_TX_RECEIVED_COUNT); this.slowValidationCount = createUpDownCounterWithDefault(meter, Metrics.P2P_GOSSIP_SLOW_VALIDATION_COUNT, { @@ -101,6 +104,11 @@ export class P2PInstrumentation { this.slowValidationCount.add(1, { [Attributes.TOPIC_NAME]: topicName }); } + /** Records the duration of a single stage of gossiped tx validation. */ + public recordTxValidationStage(stage: string, ms: number) { + this.txValidationStageDuration.record(Math.ceil(ms), { [Attributes.TX_VALIDATION_STAGE]: stage }); + } + public incMessagePrevalidationStatus(passed: boolean, topicName: TopicType | undefined) { this.messagePrevalidationCount.add(1, { [Attributes.TOPIC_NAME]: topicName, [Attributes.OK]: passed }); } diff --git a/yarn-project/p2p/src/services/libp2p/libp2p_service.ts b/yarn-project/p2p/src/services/libp2p/libp2p_service.ts index ddeb5d15901d..2d748627e146 100644 --- a/yarn-project/p2p/src/services/libp2p/libp2p_service.ts +++ b/yarn-project/p2p/src/services/libp2p/libp2p_service.ts @@ -1007,6 +1007,7 @@ export class LibP2PService extends WithTracer implements P2PService { msgId: string, source: PeerId, topicType: TopicType, + stageTimings?: Record, ): Promise> { // Default to reject result with a penalty if validation function throws an error let resultAndObj: ReceivedMessageValidationResult = { @@ -1027,12 +1028,18 @@ export class LibP2PService extends WithTracer implements P2PService { this.logger.warn( `Gossip validation for ${topicType} took ${validationTimeMs}ms, approaching mcache eviction window of ${mcacheWindowMs}ms. ` + `Message forwarding may be skipped if validation exceeds the window.`, - { msgId, source: source.toString(), topicType, validationTimeMs, mcacheWindowMs }, + { msgId, source: source.toString(), topicType, validationTimeMs, mcacheWindowMs, stageTimings }, ); } if (resultAndObj.result === TopicValidatorResult.Accept) { - this.logger.debug(`Message ${topicType} accepted by validator`, { msgId, source: source.toString(), topicType }); + this.logger.debug(`Message ${topicType} accepted by validator`, { + msgId, + source: source.toString(), + topicType, + validationTimeMs, + stageTimings, + }); this.instrumentation.recordMessageValidation(topicType, timer); } else if (resultAndObj.result === TopicValidatorResult.Reject) { this.logger.warn(`Message ${topicType} rejected by validator with severity ${resultAndObj.severity}`, { @@ -1064,18 +1071,38 @@ export class LibP2PService extends WithTracer implements P2PService { } protected async handleGossipedTx(payloadData: Buffer, msgId: string, source: PeerId) { + // Per-stage timings so slow validations can be attributed to a specific stage (see A-1656). + const stageTimings: Record = {}; + const timed = async (stage: string, fn: () => T | Promise): Promise => { + const timer = new Timer(); + try { + return await fn(); + } finally { + const ms = timer.ms(); + stageTimings[stage] = Math.ceil(ms); + this.instrumentation.recordTxValidationStage(stage, ms); + } + }; + const validationFunc: () => Promise> = async () => { - const tx = this.tryDeserialize(() => Tx.fromBuffer(payloadData), msgId, source); + const tx = await timed('deserialize', () => this.tryDeserialize(() => Tx.fromBuffer(payloadData), msgId, source)); if (!tx) { return { result: TopicValidatorResult.Reject, severity: PeerErrorSeverity.LowToleranceError }; } - const currentBlockNumber = await this.archiver.getBlockNumber(); - const { ts: nextSlotTimestamp } = this.epochCache.getEpochAndSlotInNextL1Slot(); + // Stage 1 setup: chain tip, epoch info, gas fees, and L1 constants reads + const { currentBlockNumber, firstStageValidators } = await timed('stage1_setup', async () => { + const currentBlockNumber = await this.archiver.getBlockNumber(); + const { ts: nextSlotTimestamp } = this.epochCache.getEpochAndSlotInNextL1Slot(); + const firstStageValidators = await this.createFirstStageMessageValidators( + currentBlockNumber, + nextSlotTimestamp, + ); + return { currentBlockNumber, firstStageValidators }; + }); // Stage 1: fast validators (metadata, data, timestamps, double-spend, gas, phases, block header) - const firstStageValidators = await this.createFirstStageMessageValidators(currentBlockNumber, nextSlotTimestamp); - const firstStageOutcome = await this.runValidations(tx, firstStageValidators); + const firstStageOutcome = await timed('stage1', () => this.runValidations(tx, firstStageValidators)); if (!firstStageOutcome.allPassed) { const { name } = firstStageOutcome.failure; let { severity } = firstStageOutcome.failure; @@ -1097,7 +1124,7 @@ export class LibP2PService extends WithTracer implements P2PService { } // Pool pre-check: see if the pool would accept this tx before doing expensive proof verification - const canAdd = await this.mempools.txPool.canAddPendingTx(tx); + const canAdd = await timed('pool_precheck', () => this.mempools.txPool.canAddPendingTx(tx)); if (canAdd === 'ignored') { this.logger.verbose(`Ignoring gossiped tx ${tx.getTxHash().toString()}: pool pre-check returned ignored`, { source: source.toString(), @@ -1107,7 +1134,7 @@ export class LibP2PService extends WithTracer implements P2PService { // Stage 2: expensive proof verification const secondStageValidators = this.createSecondStageMessageValidators(); - const secondStageOutcome = await this.runValidations(tx, secondStageValidators); + const secondStageOutcome = await timed('proof_verify', () => this.runValidations(tx, secondStageValidators)); if (!secondStageOutcome.allPassed) { const { severity, name } = secondStageOutcome.failure; this.logger.verbose(`Rejecting gossiped tx ${tx.getTxHash().toString()}: stage 2 validation failed`, { @@ -1120,7 +1147,7 @@ export class LibP2PService extends WithTracer implements P2PService { // Pool add: persist the tx const txHash = tx.getTxHash(); - const addResult = await this.mempools.txPool.addPendingTxs([tx], { source: 'gossip' }); + const addResult = await timed('pool_add', () => this.mempools.txPool.addPendingTxs([tx], { source: 'gossip' })); const wasAccepted = addResult.accepted.some(h => h.equals(txHash)); const wasIgnored = addResult.ignored.some(h => h.equals(txHash)); @@ -1144,7 +1171,13 @@ export class LibP2PService extends WithTracer implements P2PService { } }; - const { result, obj: tx } = await this.validateReceivedMessage(validationFunc, msgId, source, TopicType.tx); + const { result, obj: tx } = await this.validateReceivedMessage( + validationFunc, + msgId, + source, + TopicType.tx, + stageTimings, + ); if (result !== TopicValidatorResult.Accept || !tx) { return; } diff --git a/yarn-project/telemetry-client/src/attributes.ts b/yarn-project/telemetry-client/src/attributes.ts index 8a262c03efe6..728a9b1f41d9 100644 --- a/yarn-project/telemetry-client/src/attributes.ts +++ b/yarn-project/telemetry-client/src/attributes.ts @@ -133,6 +133,12 @@ export const NODEJS_EVENT_LOOP_STATE = 'nodejs.eventloop.state'; export const TOPIC_NAME = 'aztec.gossip.topic_name'; +/** Stage of gossiped tx validation (deserialize, stage1_setup, stage1, pool_precheck, proof_verify, pool_add) */ +export const TX_VALIDATION_STAGE = 'aztec.p2p.tx_validation_stage'; + +/** Operation type enqueued on the mempool serial queue */ +export const MEMPOOL_OPERATION = 'aztec.mempool.operation'; + /** The reason a transaction was evicted from the tx pool */ export const TX_POOL_EVICTION_REASON = 'aztec.mempool.eviction_reason'; diff --git a/yarn-project/telemetry-client/src/metrics.ts b/yarn-project/telemetry-client/src/metrics.ts index 6255f44aeda2..1bdb0217a689 100644 --- a/yarn-project/telemetry-client/src/metrics.ts +++ b/yarn-project/telemetry-client/src/metrics.ts @@ -204,6 +204,24 @@ export const MEMPOOL_TX_POOL_V2_METADATA_MEMORY: MetricDefinition = { valueType: ValueType.INT, }; +export const MEMPOOL_TX_POOL_V2_QUEUE_WAIT: MetricDefinition = { + name: 'aztec.mempool.tx_pool_v2.queue_wait', + description: 'Time an operation waits in the tx pool serial queue before executing, keyed by mempool.operation', + unit: 'ms', + valueType: ValueType.INT, +}; +export const MEMPOOL_TX_POOL_V2_QUEUE_EXECUTION: MetricDefinition = { + name: 'aztec.mempool.tx_pool_v2.queue_execution', + description: 'Time an operation spends executing in the tx pool serial queue, keyed by mempool.operation', + unit: 'ms', + valueType: ValueType.INT, +}; +export const MEMPOOL_TX_POOL_V2_QUEUE_LENGTH: MetricDefinition = { + name: 'aztec.mempool.tx_pool_v2.queue_length', + description: 'Number of operations waiting in the tx pool serial queue', + valueType: ValueType.INT, +}; + export const MEMPOOL_TX_POOL_V2_DUPLICATE_ADD: MetricDefinition = { name: 'aztec.mempool.tx_pool_v2.duplicate_add', description: 'Transactions received via addPendingTxs that were already in the pool', @@ -1022,6 +1040,13 @@ export const P2P_GOSSIP_SLOW_VALIDATION_COUNT: MetricDefinition = { valueType: ValueType.INT, }; +export const P2P_GOSSIP_TX_VALIDATION_STAGE_DURATION: MetricDefinition = { + name: 'aztec.p2p.gossip.tx_validation_stage_duration', + description: 'Duration of each stage of gossiped tx validation, keyed by the tx_validation_stage attribute', + unit: 'ms', + valueType: ValueType.INT, +}; + export const PUBLIC_PROCESSOR_TX_DURATION: MetricDefinition = { name: 'aztec.public_processor.tx_duration', description: 'Duration to process a public transaction',