refactor: generalize NetworkProcessor awaiting-block mechanism for all gossip types

dashboards/lodestar_networking.json

@@ -4383,7 +4383,7 @@
             "uid": "${DS_PROMETHEUS}"
           },
           "editorMode": "code",
-          "expr": "lodestar_reprocess_gossip_attestations_per_slot_total",
+          "expr": "lodestar_awaiting_block_gossip_messages_per_slot_total",
           "legendFormat": "Per Slot",
           "range": true,
           "refId": "A"
@@ -4394,14 +4394,14 @@
             "uid": "${DS_PROMETHEUS}"
           },
           "editorMode": "code",
-          "expr": "rate(lodestar_reprocess_gossip_attestations_total[$rate_interval]) * 12",
+          "expr": "sum by (topic)(rate(lodestar_awaiting_block_gossip_messages_total[$rate_interval])) * 12",
           "hide": false,
-          "legendFormat": "Avg Per Slot",
+          "legendFormat": "{{topic}}",
           "range": true,
           "refId": "B"
         }
       ],
-      "title": "Reprocess",
+      "title": "Awaiting Block Gossip Messages",
       "type": "timeseries"
     },
     {
@@ -4574,8 +4574,8 @@
             "uid": "${DS_PROMETHEUS}"
           },
           "editorMode": "code",
-          "expr": "rate(lodestar_reprocess_gossip_attestations_resolve_total[$rate_interval]) * 12",
-          "legendFormat": "Per Slot",
+          "expr": "sum by (topic)(rate(lodestar_awaiting_block_gossip_messages_resolve_total[$rate_interval])) * 12",
+          "legendFormat": "{{topic}}",
           "range": true,
           "refId": "A"
         },
@@ -4585,14 +4585,14 @@
             "uid": "${DS_PROMETHEUS}"
           },
           "editorMode": "code",
-          "expr": "lodestar_reprocess_gossip_attestations_wait_time_resolve_seconds",
+          "expr": "lodestar_awaiting_block_gossip_messages_wait_time_resolve_seconds",
           "hide": false,
-          "legendFormat": "Wait Sec",
+          "legendFormat": "{{topic}}",
           "range": true,
           "refId": "B"
         }
       ],
-      "title": "Reprocess - Resolved",
+      "title": "Awaiting Block Gossip Messages - Resolved",
       "type": "timeseries"
     },
     {
@@ -4684,8 +4684,8 @@
             "uid": "${DS_PROMETHEUS}"
           },
           "editorMode": "code",
-          "expr": "rate(lodestar_reprocess_gossip_attestations_reject_total[$rate_interval]) * 12",
-          "legendFormat": "{{reason}}",
+          "expr": "sum by (reason, topic)(rate(lodestar_awaiting_block_gossip_messages_reject_total[$rate_interval])) * 12",
+          "legendFormat": "{{reason}} - {{topic}}",
           "range": true,
           "refId": "A"
         },
@@ -4695,14 +4695,14 @@
             "uid": "${DS_PROMETHEUS}"
           },
           "editorMode": "code",
-          "expr": "lodestar_reprocess_gossip_attestations_wait_time_reject_seconds",
+          "expr": "lodestar_awaiting_block_gossip_messages_wait_time_reject_seconds",
           "hide": false,
-          "legendFormat": "Wait Sec",
+          "legendFormat": "{{reason}} - {{topic}}",
           "range": true,
           "refId": "B"
         }
       ],
-      "title": "Reprocess - Reject",
+      "title": "Awaiting Block Gossip Messages - Rejected",
       "type": "timeseries"
     },
     {

packages/beacon-node/src/api/impl/validator/index.ts

@@ -1068,7 +1068,7 @@ export function getValidatorApi(
       // see https://github.com/ChainSafe/lodestar/issues/5063
       if (!chain.forkChoice.hasBlock(beaconBlockRoot)) {
         const rootHex = toRootHex(beaconBlockRoot);
-        network.searchUnknownSlotRoot({slot, root: rootHex}, BlockInputSource.api);
+        network.searchUnknownBlock({slot, root: rootHex}, BlockInputSource.api);
         // if result of this call is false, i.e. block hasn't seen after 1 slot then the below notOnOptimisticBlockRoot call will throw error
         await chain.waitForBlock(slot, rootHex);
       }

packages/beacon-node/src/chain/blocks/blockInput/types.ts

@@ -16,6 +16,7 @@ export type DAData = null | deneb.BlobSidecars | fulu.DataColumnSidecars;
  * sources so each should be labelled individually.
  */
 export enum BlockInputSource {
+  network_processor = "network_processor",
   gossip = "gossip",
   api = "api",
   engine = "engine",

packages/beacon-node/src/metrics/metrics/lodestar.ts

@@ -1635,33 +1635,37 @@ export function createLodestarMetrics(
       }),
     },
 
-    // reprocess gossip attestations
-    reprocessGossipAttestations: {
-      total: register.gauge({
-        name: "lodestar_reprocess_gossip_attestations_total",
-        help: "Total number of gossip attestations waiting to reprocess",
+    // some gossip messages need to wait for block to be processed before they can be processed
+    awaitingBlockGossipMessages: {
+      queue: register.gauge<{topic: GossipType}>({
+        name: "lodestar_awaiting_block_gossip_messages_total",
+        help: "Total number of gossip messages waiting for block to be processed",
+        labelNames: ["topic"],
       }),
       countPerSlot: register.gauge({
-        name: "lodestar_reprocess_gossip_attestations_per_slot_total",
-        help: "Total number of gossip attestations waiting to reprocess pet slot",
+        name: "lodestar_awaiting_block_gossip_messages_per_slot_total",
+        help: "Total number of gossip messages waiting for block to be processed per slot",
       }),
-      resolve: register.gauge({
-        name: "lodestar_reprocess_gossip_attestations_resolve_total",
-        help: "Total number of gossip attestations are reprocessed",
+      resolve: register.gauge<{topic: GossipType}>({
+        name: "lodestar_awaiting_block_gossip_messages_resolve_total",
+        help: "Total number of gossip messages are reprocessed",
+        labelNames: ["topic"],
       }),
-      waitSecBeforeResolve: register.gauge({
-        name: "lodestar_reprocess_gossip_attestations_wait_time_resolve_seconds",
+      waitSecBeforeResolve: register.gauge<{topic: GossipType}>({
+        name: "lodestar_awaiting_block_gossip_messages_wait_time_resolve_seconds",
         help: "Time to wait for unknown block in seconds",
+        labelNames: ["topic"],
       }),
-      reject: register.gauge<{reason: ReprocessRejectReason}>({
-        name: "lodestar_reprocess_gossip_attestations_reject_total",
-        help: "Total number of attestations are rejected to reprocess",
-        labelNames: ["reason"],
+      // having 2 labels here is not great for performance, however it's rarely happening and having the reason label is important for debugging
+      reject: register.gauge<{reason: ReprocessRejectReason; topic: GossipType}>({
+        name: "lodestar_awaiting_block_gossip_messages_reject_total",
+        help: "Total number of gossip messages are rejected to reprocess",
+        labelNames: ["reason", "topic"],
       }),
-      waitSecBeforeReject: register.gauge<{reason: ReprocessRejectReason}>({
-        name: "lodestar_reprocess_gossip_attestations_wait_time_reject_seconds",
+      waitSecBeforeReject: register.gauge<{reason: ReprocessRejectReason; topic: GossipType}>({
+        name: "lodestar_awaiting_block_gossip_messages_wait_time_reject_seconds",
         help: "Time to wait for unknown block before being rejected",
-        labelNames: ["reason"],
+        labelNames: ["reason", "topic"],
       }),
     },
 

packages/beacon-node/src/network/interface.ts

@@ -68,7 +68,7 @@ export interface INetwork extends INetworkCorePublic {
   reportPeer(peer: PeerIdStr, action: PeerAction, actionName: string): void;
   shouldAggregate(subnet: SubnetID, slot: Slot): boolean;
   reStatusPeers(peers: PeerIdStr[]): Promise<void>;
-  searchUnknownSlotRoot(slotRoot: SlotRootHex, source: BlockInputSource, peer?: PeerIdStr): void;
+  searchUnknownBlock(slotRoot: SlotRootHex, source: BlockInputSource, peer?: PeerIdStr): void;
   // ReqResp
   sendBeaconBlocksByRange(peerId: PeerIdStr, request: phase0.BeaconBlocksByRangeRequest): Promise<SignedBeaconBlock[]>;
   sendBeaconBlocksByRoot(peerId: PeerIdStr, request: BeaconBlocksByRootRequest): Promise<SignedBeaconBlock[]>;

packages/beacon-node/src/network/network.ts

@@ -277,8 +277,8 @@ export class Network implements INetwork {
     return this.core.reStatusPeers(peers);
   }
 
-  searchUnknownSlotRoot(slotRoot: SlotRootHex, source: BlockInputSource, peer?: PeerIdStr): void {
-    this.networkProcessor.searchUnknownSlotRoot(slotRoot, source, peer);
+  searchUnknownBlock(slotRoot: SlotRootHex, source: BlockInputSource, peer?: PeerIdStr): void {
+    this.networkProcessor.searchUnknownBlock(slotRoot, source, peer);
   }
 
   async reportPeer(peer: PeerIdStr, action: PeerAction, actionName: string): Promise<void> {

packages/beacon-node/src/network/processor/extractSlotRootFns.ts

@@ -14,7 +14,7 @@ import {ExtractSlotRootFns} from "./types.js";
 
 /**
  * Extract the slot and block root of a gossip message form serialized data.
- * Only applicable for beacon_attestation and beacon_aggregate_and_proof topics.
+ * Not applicable for all topics.
  */
 export function createExtractBlockSlotRootFns(): ExtractSlotRootFns {
   return {

packages/beacon-node/src/network/processor/gossipHandlers.ts

@@ -1007,7 +1007,7 @@ export async function validateGossipFnRetryUnknownRoot<T>(
         if (unknownBlockRootRetries === 0) {
           // Trigger unknown block root search here
           const rootHex = toRootHex(blockRoot);
-          network.searchUnknownSlotRoot({slot, root: rootHex}, BlockInputSource.gossip);
+          network.searchUnknownBlock({slot, root: rootHex}, BlockInputSource.gossip);
         }
 
         if (unknownBlockRootRetries++ < MAX_UNKNOWN_BLOCK_ROOT_RETRIES) {