feat: message propagation metrics (#19265)

Ref: A-413, A-412

Span to track attestation propagation on `AttestToProposal` function
Peer connection time metric

Also changed metrics from previous pr:
1. track deleted transactions in `cleanupDeletedMinedTxs`
3. Record p2p requested transaction even if we requested zero

Author: mralj

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

@@ -1,5 +1,5 @@
 import type { Gossipable } from '@aztec/stdlib/p2p';
-import type { Tx, TxHash } from '@aztec/stdlib/tx';
+import type { Tx } from '@aztec/stdlib/tx';
 import {
   Attributes,
   type BatchObservableResult,
@@ -72,7 +72,7 @@ export class PoolInstrumentation<PoolObject extends Gossipable> {
   private defaultAttributes;
   private meter: Meter;
 
-  private txAddedTimestamp: Map<string, number> = new Map<string, number>();
+  private txAddedTimestamp: Map<bigint, number> = new Map<bigint, number>();
 
   constructor(
     telemetry: TelemetryClient,
@@ -124,17 +124,16 @@ export class PoolInstrumentation<PoolObject extends Gossipable> {
   public transactionsAdded(transactions: Tx[]) {
     const timestamp = Date.now();
     for (const transaction of transactions) {
-      this.txAddedTimestamp.set(transaction.txHash.toString(), timestamp);
+      this.txAddedTimestamp.set(transaction.txHash.toBigInt(), timestamp);
     }
   }
 
-  public transactionsRemoved(hashes: TxHash[]) {
+  public transactionsRemoved(hashes: Iterable<bigint> | Iterable<string>) {
     const timestamp = Date.now();
     for (const hash of hashes) {
-      const key = hash.toString();
-      const addedAt = this.txAddedTimestamp.get(key);
+      const addedAt = this.txAddedTimestamp.get(BigInt(hash));
       if (addedAt) {
-        this.txAddedTimestamp.delete(key);
+        this.txAddedTimestamp.delete(BigInt(hash));
         this.minedDelay.record(addedAt - timestamp);
       }
     }

yarn-project/p2p/src/mem_pools/tx_pool/aztec_kv_tx_pool.ts

@@ -187,7 +187,7 @@ export class AztecKVTxPool extends (EventEmitter as new () => TypedEventEmitter<
 
       await this.evictInvalidTxsAfterMining(txHashes, blockHeader, minedNullifiers, minedFeePayers);
     });
-    this.#metrics.transactionsRemoved(txHashes);
+    this.#metrics.transactionsRemoved(txHashes.map(hash => hash.toBigInt()));
     // We update this after the transaction above. This ensures that the non-evictable transactions are not evicted
     // until any that have been mined are marked as such.
     // The non-evictable set is not considered when evicting transactions that are invalid after a block is mined.
@@ -386,7 +386,7 @@ export class AztecKVTxPool extends (EventEmitter as new () => TypedEventEmitter<
 
       await this.#pendingTxSize.set(pendingTxSize);
     });
-    this.#metrics.transactionsRemoved(txHashes);
+    this.#metrics.transactionsRemoved(txHashes.map(hash => hash.toBigInt()));
     this.#log.debug(`Deleted ${txHashes.length} txs from pool`, { txHashes });
     return this.#archivedTxLimit ? poolDbTx.then(() => this.archiveTxs(deletedTxs)) : poolDbTx;
   }
@@ -464,6 +464,7 @@ export class AztecKVTxPool extends (EventEmitter as new () => TypedEventEmitter<
           deletedCount++;
         }
       }
+      this.#metrics.transactionsRemoved(txHashesToDelete);
 
       // Clean up block-to-hash mapping - delete all values for each block
       for (const block of blocksToDelete) {

yarn-project/p2p/src/mem_pools/tx_pool/memory_tx_pool.ts

@@ -76,7 +76,7 @@ export class InMemoryTxPool extends (EventEmitter as new () => TypedEventEmitter
       this.minedTxs.set(key, blockHeader.globalVariables.blockNumber);
       this.pendingTxs.delete(key);
     }
-    this.metrics.transactionsRemoved(txHashes);
+    this.metrics.transactionsRemoved(txHashes.map(hash => hash.toBigInt()));
     return Promise.resolve();
   }
 
@@ -226,7 +226,7 @@ export class InMemoryTxPool extends (EventEmitter as new () => TypedEventEmitter
         }
       }
     }
-    this.metrics.transactionsRemoved(txHashes);
+    this.metrics.transactionsRemoved(txHashes.map(hash => hash.toBigInt()));
 
     return Promise.resolve();
   }
@@ -271,6 +271,7 @@ export class InMemoryTxPool extends (EventEmitter as new () => TypedEventEmitter
           this.deletedMinedTxHashes.delete(txHash);
           deletedCount++;
         }
+        this.metrics.transactionsRemoved(txHashes);
         blocksToDelete.push(block);
       }
     }

yarn-project/p2p/src/services/peer-manager/metrics.ts

@@ -1,6 +1,7 @@
 import {
   Attributes,
   type Gauge,
+  type Histogram,
   Metrics,
   type TelemetryClient,
   type Tracer,
@@ -9,13 +10,18 @@ import {
   getTelemetryClient,
 } from '@aztec/telemetry-client';
 
+import type { PeerId } from '@libp2p/interface';
+
 import { type GoodByeReason, prettyGoodbyeReason } from '../reqresp/protocols/index.js';
 
 export class PeerManagerMetrics {
   private sentGoodbyes: UpDownCounter;
   private receivedGoodbyes: UpDownCounter;
   private peerCount: Gauge;
   private lowScoreDisconnects: UpDownCounter;
+  private peerConnectionDuration: Histogram;
+
+  private peerConnectedAt: Map<string, number> = new Map<string, number>();
 
   public readonly tracer: Tracer;
 
@@ -46,6 +52,11 @@ export class PeerManagerMetrics {
       unit: 'peers',
       valueType: ValueType.INT,
     });
+    this.peerConnectionDuration = meter.createHistogram(Metrics.PEER_MANAGER_PEER_CONNECTION_DURATION, {
+      description: 'Time duration between peer connection and disconnection',
+      unit: 'ms',
+      valueType: ValueType.INT,
+    });
   }
 
   public recordGoodbyeSent(reason: GoodByeReason) {
@@ -63,4 +74,15 @@ export class PeerManagerMetrics {
   public recordLowScoreDisconnect(scoreState: 'Banned' | 'Disconnect') {
     this.lowScoreDisconnects.add(1, { [Attributes.P2P_PEER_SCORE_STATE]: scoreState });
   }
+
+  public peerConnected(id: PeerId) {
+    this.peerConnectedAt.set(id.toString(), Date.now());
+  }
+
+  public peerDisconnected(id: PeerId) {
+    const connectedAt = this.peerConnectedAt.get(id.toString());
+    if (connectedAt) {
+      this.peerConnectionDuration.record(Date.now() - connectedAt);
+    }
+  }
 }

yarn-project/p2p/src/services/peer-manager/peer_manager.ts

@@ -278,6 +278,7 @@ export class PeerManager implements PeerManagerInterface {
   private handleConnectedPeerEvent(e: CustomEvent<PeerId>) {
     const peerId = e.detail;
     this.logger.verbose(`Connected to peer ${peerId.toString()}`);
+    this.metrics.peerConnected(peerId);
     if (this.config.p2pDisableStatusHandshake) {
       return;
     }
@@ -303,6 +304,7 @@ export class PeerManager implements PeerManagerInterface {
    */
   private handleDisconnectedPeerEvent(e: CustomEvent<PeerId>) {
     const peerId = e.detail;
+    this.metrics.peerDisconnected(peerId);
     this.logger.verbose(`Disconnected from peer ${peerId.toString()}`);
     const validatorAddress = this.authenticatedPeerIdToValidatorAddress.get(peerId.toString());
     if (validatorAddress !== undefined) {

yarn-project/p2p/src/services/tx_provider.ts

@@ -137,6 +137,7 @@ export class TxProvider implements ITxProvider {
     );
 
     if (missingTxHashes.size === 0) {
+      this.instrumentation.incTxsFromP2P(0, txHashes.length);
       return { txsFromMempool };
     }
 
@@ -155,6 +156,7 @@ export class TxProvider implements ITxProvider {
 
     if (missingTxHashes.size === 0) {
       await this.processProposalTxs(txsFromProposal);
+      this.instrumentation.incTxsFromP2P(0, txHashes.length);
       return { txsFromMempool, txsFromProposal };
     }
 
@@ -166,13 +168,14 @@ export class TxProvider implements ITxProvider {
 
     if (txsFromNetwork.length > 0) {
       txsFromNetwork.forEach(tx => missingTxHashes.delete(tx.txHash.toString()));
-      this.instrumentation.incTxsFromP2P(txsFromNetwork.length, txHashes.length);
       this.log.debug(
         `Retrieved ${txsFromNetwork.length} txs from network for block proposal (${missingTxHashes.size} pending)`,
         { ...blockInfo, missingTxHashes: [...missingTxHashes] },
       );
     }
 
+    this.instrumentation.incTxsFromP2P(txsFromNetwork.length, txHashes.length);
+
     if (missingTxHashes.size === 0) {
       return { txsFromNetwork, txsFromMempool, txsFromProposal };
     }

yarn-project/sequencer-client/src/sequencer/checkpoint_proposal_job.ts

@@ -108,6 +108,10 @@ export class CheckpointProposalJob {
     // Wait until the voting promises have resolved, so all requests are enqueued (not sent)
     await Promise.all(votesPromises);
 
+    if (checkpoint) {
+      this.metrics.recordBlockProposalSuccess();
+    }
+
     // Do not post anything to L1 if we are fishermen, but do perform L1 fee analysis
     if (this.config.fishermanMode) {
       await this.handleCheckpointEndAsFisherman(checkpoint);
@@ -669,7 +673,6 @@ export class CheckpointProposalJob {
         ...checkpoint.getStats(),
         feeAnalysisId: feeAnalysis?.id,
       });
-      this.metrics.recordBlockProposalSuccess();
     } else {
       this.log.warn(`Validation block building FAILED for slot ${this.slot}`, {
         slot: this.slot,

yarn-project/telemetry-client/src/attributes.ts

@@ -99,6 +99,8 @@ export const P2P_REQ_RESP_BATCH_REQUESTS_COUNT = 'aztec.p2p.req_resp.batch_reque
 export const P2P_PEER_SCORE_STATE = 'aztec.p2p.peer_score_state';
 export const POOL_NAME = 'aztec.pool.name';
 
+export const PEER_ID = 'aztec.p2p.peer_id';
+
 export const SEQUENCER_STATE = 'aztec.sequencer.state';
 
 export const SIMULATOR_PHASE = 'aztec.simulator.phase';

yarn-project/telemetry-client/src/metrics.ts

@@ -143,6 +143,7 @@ export const PEER_MANAGER_GOODBYES_SENT = 'aztec.peer_manager.goodbyes_sent';
 export const PEER_MANAGER_GOODBYES_RECEIVED = 'aztec.peer_manager.goodbyes_received';
 export const PEER_MANAGER_PEER_COUNT = 'aztec.peer_manager.peer_count';
 export const PEER_MANAGER_LOW_SCORE_DISCONNECTS = 'aztec.peer_manager.low_score_disconnects';
+export const PEER_MANAGER_PEER_CONNECTION_DURATION = 'aztec.peer_manager.peer_connection_duration';
 export const P2P_PEER_STATE_COUNT = 'aztec.p2p.peer_state_count';
 
 export const P2P_REQ_RESP_SENT_REQUESTS = 'aztec.p2p.req_resp.sent_requests';

yarn-project/validator-client/src/validator.ts

@@ -21,7 +21,7 @@ import type { BlockAttestation, BlockProposal, BlockProposalOptions } from '@azt
 import type { CheckpointHeader } from '@aztec/stdlib/rollup';
 import type { Tx } from '@aztec/stdlib/tx';
 import { AttestationTimeoutError } from '@aztec/stdlib/validators';
-import { type TelemetryClient, type Tracer, getTelemetryClient } from '@aztec/telemetry-client';
+import { Attributes, type TelemetryClient, type Tracer, getTelemetryClient, trackSpan } from '@aztec/telemetry-client';
 
 import { EventEmitter } from 'events';
 import type { TypedDataDefinition } from 'viem';
@@ -264,6 +264,10 @@ export class ValidatorClient extends (EventEmitter as new () => WatcherEmitter)
     }
   }
 
+  @trackSpan('validator.attestToProposal', (proposal, proposalSender) => ({
+    [Attributes.BLOCK_HASH]: proposal.payload.header.hash.toString(),
+    [Attributes.PEER_ID]: proposalSender.toString(),
+  }))
   async attestToProposal(proposal: BlockProposal, proposalSender: PeerId): Promise<BlockAttestation[] | undefined> {
     const slotNumber = proposal.slotNumber;
     const proposer = proposal.getSender();