From a142e115e71d5e41d81790f4e183c3a7ae641607 Mon Sep 17 00:00:00 2001 From: Tuyen Nguyen Date: Wed, 18 Mar 2026 14:01:26 +0700 Subject: [PATCH 1/4] refactor: generalize network processor from attestation-specific to all gossip messages MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Rename `searchUnknownSlotRoot` → `searchUnknownBlock` across network interface - Flatten `awaitingGossipsubMessagesByRootBySlot` (MapDef of MapDef) into `awaitingBlockByRoot` (single MapDef) for simpler lookups - Rename `unknownRootsBySlot` → `unknownBlocksBySlot` for clarity - Rename metrics `reprocessGossipAttestations` → `awaitingBlockGossipMessages` with added `topic` label - Add `BlockInputSource.network_processor` to label messages queued internally - Update dashboard panels to use new metric names and topic breakdowns Co-Authored-By: Claude Sonnet 4.6 --- dashboards/lodestar_networking.json | 28 +-- .../src/api/impl/validator/index.ts | 2 +- .../src/chain/blocks/blockInput/types.ts | 1 + .../src/metrics/metrics/lodestar.ts | 42 ++-- packages/beacon-node/src/network/interface.ts | 2 +- packages/beacon-node/src/network/network.ts | 4 +- .../src/network/processor/gossipHandlers.ts | 2 +- .../src/network/processor/index.ts | 194 ++++++++++-------- packages/beacon-node/src/sync/unknownBlock.ts | 2 +- 9 files changed, 148 insertions(+), 129 deletions(-) diff --git a/dashboards/lodestar_networking.json b/dashboards/lodestar_networking.json index a8964cf0d4bf..b02c8121028f 100644 --- a/dashboards/lodestar_networking.json +++ b/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" }, { diff --git a/packages/beacon-node/src/api/impl/validator/index.ts b/packages/beacon-node/src/api/impl/validator/index.ts index 2a1e9b1d9fc9..96f0e1aab13a 100644 --- a/packages/beacon-node/src/api/impl/validator/index.ts +++ b/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); } diff --git a/packages/beacon-node/src/chain/blocks/blockInput/types.ts b/packages/beacon-node/src/chain/blocks/blockInput/types.ts index e932360a9269..0ca5c4b43f62 100644 --- a/packages/beacon-node/src/chain/blocks/blockInput/types.ts +++ b/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", diff --git a/packages/beacon-node/src/metrics/metrics/lodestar.ts b/packages/beacon-node/src/metrics/metrics/lodestar.ts index 32955d5a9e4d..67cc4df58922 100644 --- a/packages/beacon-node/src/metrics/metrics/lodestar.ts +++ b/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"], }), }, diff --git a/packages/beacon-node/src/network/interface.ts b/packages/beacon-node/src/network/interface.ts index 225202317620..25abc19d6c51 100644 --- a/packages/beacon-node/src/network/interface.ts +++ b/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; - searchUnknownSlotRoot(slotRoot: SlotRootHex, source: BlockInputSource, peer?: PeerIdStr): void; + searchUnknownBlock(slotRoot: SlotRootHex, source: BlockInputSource, peer?: PeerIdStr): void; // ReqResp sendBeaconBlocksByRange(peerId: PeerIdStr, request: phase0.BeaconBlocksByRangeRequest): Promise; sendBeaconBlocksByRoot(peerId: PeerIdStr, request: BeaconBlocksByRootRequest): Promise; diff --git a/packages/beacon-node/src/network/network.ts b/packages/beacon-node/src/network/network.ts index 6afe85dbad2e..9cc9756dbfc7 100644 --- a/packages/beacon-node/src/network/network.ts +++ b/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 { diff --git a/packages/beacon-node/src/network/processor/gossipHandlers.ts b/packages/beacon-node/src/network/processor/gossipHandlers.ts index b022a1866470..3d14ea2c10e6 100644 --- a/packages/beacon-node/src/network/processor/gossipHandlers.ts +++ b/packages/beacon-node/src/network/processor/gossipHandlers.ts @@ -1007,7 +1007,7 @@ export async function validateGossipFnRetryUnknownRoot( 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) { diff --git a/packages/beacon-node/src/network/processor/index.ts b/packages/beacon-node/src/network/processor/index.ts index cf20ab63d2e3..cf597bd95639 100644 --- a/packages/beacon-node/src/network/processor/index.ts +++ b/packages/beacon-node/src/network/processor/index.ts @@ -1,8 +1,6 @@ import {routes} from "@lodestar/api"; -import {ForkSeq} from "@lodestar/params"; -import {computeStartSlotAtEpoch} from "@lodestar/state-transition"; import {RootHex, Slot, SlotRootHex} from "@lodestar/types"; -import {Logger, MapDef, mapValues, pruneSetToMax, sleep} from "@lodestar/utils"; +import {Logger, MapDef, mapValues, sleep} from "@lodestar/utils"; import {BlockInputSource} from "../../chain/blocks/blockInput/types.js"; import {ChainEvent} from "../../chain/emitter.js"; import {GossipErrorCode} from "../../chain/errors/gossipValidation.js"; @@ -87,27 +85,27 @@ const executeGossipWorkOrder = Object.keys(executeGossipWorkOrderObj) as (keyof // TODO: Arbitrary constant, check metrics const MAX_JOBS_SUBMITTED_PER_TICK = 128; -// How many attestations (aggregate + unaggregate) we keep before new ones get dropped. +// How many gossip messages we keep before new ones get dropped. const MAX_QUEUED_UNKNOWN_BLOCK_GOSSIP_OBJECTS = 16_384; -// We don't want to process too many attestations in a single tick -// As seen on mainnet, attestation concurrency metric ranges from 1000 to 2000 +// We don't want to process too many gossip messages in a single tick +// As seen on mainnet, gossip messages concurrency metric ranges from 1000 to 2000 // so make this constant a little bit conservative -const MAX_UNKNOWN_BLOCK_GOSSIP_OBJECTS_PER_TICK = 1024; +const MAX_AWAITING_GOSSIP_OBJECTS_PER_TICK = 1024; // Same motivation to JobItemQueue, we don't want to block the event loop -const PROCESS_UNKNOWN_BLOCK_GOSSIP_OBJECTS_YIELD_EVERY_MS = 50; +const AWAITING_GOSSIP_OBJECTS_YIELD_EVERY_MS = 50; /** * Reprocess reject reason for metrics */ export enum ReprocessRejectReason { /** - * There are too many attestations that have unknown block root. + * There are too many gossip messages that have unknown block root. */ reached_limit = "reached_limit", /** - * The awaiting attestation is pruned per clock slot. + * The awaiting gossip message is pruned per clock slot. */ expired = "expired", } @@ -137,7 +135,7 @@ export enum CannotAcceptWorkReason { * * ### PendingGossipsubMessage beacon_attestation example * - * For attestations, processing the message includes the steps: + * For gossip messages, processing the message includes the steps: * 1. Pre shuffling sync validation * 2. Retrieve shuffling: async + goes into the regen queue and can be expensive * 3. Pre sig validation sync validation @@ -156,11 +154,11 @@ export class NetworkProcessor { private readonly gossipQueues: ReturnType; private readonly gossipTopicConcurrency: {[K in GossipType]: number}; private readonly extractBlockSlotRootFns = createExtractBlockSlotRootFns(); - // we may not receive the block for Attestation and SignedAggregateAndProof messages, in that case PendingGossipsubMessage needs + // we may not receive the block for messages like Attestation and SignedAggregateAndProof messages, in that case PendingGossipsubMessage needs // to be stored in this Map and reprocessed once the block comes - private readonly awaitingGossipsubMessagesByRootBySlot: MapDef>>; + private readonly awaitingBlockByRoot: MapDef>; private unknownBlockGossipsubMessagesCount = 0; - private unknownRootsBySlot = new MapDef>(() => new Set()); + private unknownBlocksBySlot = new MapDef>(() => new Set()); constructor( modules: NetworkProcessorModules, @@ -184,9 +182,7 @@ export class NetworkProcessor { this.chain.emitter.on(routes.events.EventType.block, this.onBlockProcessed.bind(this)); this.chain.clock.on(ClockEvent.slot, this.onClockSlot.bind(this)); - this.awaitingGossipsubMessagesByRootBySlot = new MapDef( - () => new MapDef>(() => new Set()) - ); + this.awaitingBlockByRoot = new MapDef>(() => new Set()); // TODO: Implement queues and priorization for ReqResp incoming requests // Listens to NetworkEvent.reqRespIncomingRequest event @@ -198,7 +194,7 @@ export class NetworkProcessor { metrics.gossipValidationQueue.keySize.set({topic}, this.gossipQueues[topic].keySize); metrics.gossipValidationQueue.concurrency.set({topic}, this.gossipTopicConcurrency[topic]); } - metrics.reprocessGossipAttestations.countPerSlot.set(this.unknownBlockGossipsubMessagesCount); + metrics.awaitingBlockGossipMessages.countPerSlot.set(this.unknownBlockGossipsubMessagesCount); // specific metric for beacon_attestation topic metrics.gossipValidationQueue.keyAge.reset(); for (const ageMs of this.gossipQueues.beacon_attestation.getDataAgeMs()) { @@ -233,64 +229,73 @@ export class NetworkProcessor { return queue.getAll(); } - searchUnknownSlotRoot({slot, root}: SlotRootHex, source: BlockInputSource, peer?: PeerIdStr): void { - if (this.chain.seenBlock(root) || this.unknownRootsBySlot.getOrDefault(slot).has(root)) { + /** + * Search block via `ChainEvent.unknownBlockRoot` event + * Note that slot is not necessarily the same to the block's slot but it can be used for a good prune strategy. + * In the rare case, if 2 messages on 2 slots search for the same root (for example beacon_attestation) we may emit the same root twice but BlockInputSync should handle it well. + */ + searchUnknownBlock({slot, root}: SlotRootHex, source: BlockInputSource, peer?: PeerIdStr): void { + if ( + this.chain.seenBlock(root) || + this.awaitingBlockByRoot.has(root) || + this.unknownBlocksBySlot.getOrDefault(slot).has(root) + ) { return; } // Search for the unknown block - this.unknownRootsBySlot.getOrDefault(slot).add(root); + this.unknownBlocksBySlot.getOrDefault(slot).add(root); this.chain.emitter.emit(ChainEvent.unknownBlockRoot, {rootHex: root, peer, source}); } private onPendingGossipsubMessage(message: PendingGossipsubMessage): void { const topicType = message.topic.type; const extractBlockSlotRootFn = this.extractBlockSlotRootFns[topicType]; - // check block root of Attestation and SignedAggregateAndProof messages - if (extractBlockSlotRootFn) { - const slotRoot = extractBlockSlotRootFn(message.msg.data, message.topic.boundary.fork); - // if slotRoot is null, it means the msg.data is invalid - // in that case message will be rejected when deserializing data in later phase (gossipValidatorFn) - if (slotRoot) { - // DOS protection: avoid processing messages that are too old - const {slot, root} = slotRoot; - const clockSlot = this.chain.clock.currentSlot; - const {fork} = message.topic.boundary; - let earliestPermissableSlot = clockSlot - DEFAULT_EARLIEST_PERMISSIBLE_SLOT_DISTANCE; - if (ForkSeq[fork] >= ForkSeq.deneb && topicType === GossipType.beacon_attestation) { - // post deneb, the attestations could be in current or previous epoch - earliestPermissableSlot = computeStartSlotAtEpoch(this.chain.clock.currentEpoch - 1); - } - if (slot < earliestPermissableSlot) { - // TODO: Should report the dropped job to gossip? It will be eventually pruned from the mcache - this.metrics?.networkProcessor.gossipValidationError.inc({ - topic: topicType, - error: GossipErrorCode.PAST_SLOT, - }); - return; - } - message.msgSlot = slot; - // check if we processed a block with this root - // no need to check if root is a descendant of the current finalized block, it will be checked once we validate the message if needed - if (root && !this.chain.forkChoice.hasBlockHexUnsafe(root)) { - this.searchUnknownSlotRoot({slot, root}, BlockInputSource.gossip, message.propagationSource.toString()); - - if (this.unknownBlockGossipsubMessagesCount > MAX_QUEUED_UNKNOWN_BLOCK_GOSSIP_OBJECTS) { - // TODO: Should report the dropped job to gossip? It will be eventually pruned from the mcache - this.metrics?.reprocessGossipAttestations.reject.inc({reason: ReprocessRejectReason.reached_limit}); - return; - } + const slotRoot = extractBlockSlotRootFn + ? extractBlockSlotRootFn(message.msg.data, message.topic.boundary.fork) + : null; + if (slotRoot === null) { + // some messages don't have slot and root + // if the msg.data is invalid, message will be rejected when deserializing data in later phase (gossipValidatorFn) + this.pushPendingGossipsubMessageToQueue(message); + return; + } - this.metrics?.reprocessGossipAttestations.total.inc(); - const awaitingGossipsubMessagesByRoot = this.awaitingGossipsubMessagesByRootBySlot.getOrDefault(slot); - const awaitingGossipsubMessages = awaitingGossipsubMessagesByRoot.getOrDefault(root); - awaitingGossipsubMessages.add(message); - this.unknownBlockGossipsubMessagesCount++; - return; - } + // common check for all topics + // DOS protection: avoid processing messages that are too old + const {slot, root} = slotRoot; + const clockSlot = this.chain.clock.currentSlot; + const earliestPermissableSlot = clockSlot - DEFAULT_EARLIEST_PERMISSIBLE_SLOT_DISTANCE; + if (slot < earliestPermissableSlot) { + // No need to report the dropped job to gossip. It will be eventually pruned from the mcache + this.metrics?.networkProcessor.gossipValidationError.inc({ + topic: topicType, + error: GossipErrorCode.PAST_SLOT, + }); + return; + } + + message.msgSlot = slot; + + // no need to check if root is a descendant of the current finalized block, it will be checked once we validate the message if needed + if (root && !this.chain.forkChoice.hasBlockHexUnsafe(root)) { + this.searchUnknownBlock({slot, root}, BlockInputSource.network_processor, message.propagationSource.toString()); + + if (this.unknownBlockGossipsubMessagesCount > MAX_QUEUED_UNKNOWN_BLOCK_GOSSIP_OBJECTS) { + // No need to report the dropped job to gossip. It will be eventually pruned from the mcache + this.metrics?.awaitingBlockGossipMessages.reject.inc({ + reason: ReprocessRejectReason.reached_limit, + topic: topicType, + }); + return; } + + this.metrics?.awaitingBlockGossipMessages.queue.inc({topic: topicType}); + const awaitingGossipsubMessages = this.awaitingBlockByRoot.getOrDefault(root); + awaitingGossipsubMessages.add(message); + this.unknownBlockGossipsubMessagesCount++; + return; } - // bypass the check for other messages this.pushPendingGossipsubMessageToQueue(message); } @@ -298,7 +303,7 @@ export class NetworkProcessor { const topicType = message.topic.type; const droppedCount = this.gossipQueues[topicType].add(message); if (droppedCount) { - // TODO: Should report the dropped job to gossip? It will be eventually pruned from the mcache + // No need to report the dropped job to gossip. It will be eventually pruned from the mcache this.metrics?.gossipValidationQueue.droppedJobs.inc({topic: message.topic.type}, droppedCount); } @@ -306,58 +311,67 @@ export class NetworkProcessor { this.executeWork(); } - private async onBlockProcessed({ - slot, - block: rootHex, - }: { - slot: Slot; - block: string; - executionOptimistic: boolean; - }): Promise { - const byRootGossipsubMessages = this.awaitingGossipsubMessagesByRootBySlot.getOrDefault(slot); - const waitingGossipsubMessages = byRootGossipsubMessages.getOrDefault(rootHex); + private async onBlockProcessed({block: rootHex}: {block: string; executionOptimistic: boolean}): Promise { + const waitingGossipsubMessages = this.awaitingBlockByRoot.getOrDefault(rootHex); if (waitingGossipsubMessages.size === 0) { return; } - this.metrics?.reprocessGossipAttestations.resolve.inc(waitingGossipsubMessages.size); const nowSec = Date.now() / 1000; let count = 0; // TODO: we can group attestations to process in batches but since we have the SeenAttestationDatas // cache, it may not be necessary at this time for (const message of waitingGossipsubMessages) { - this.metrics?.reprocessGossipAttestations.waitSecBeforeResolve.set(nowSec - message.seenTimestampSec); + const topicType = message.topic.type; + this.metrics?.awaitingBlockGossipMessages.waitSecBeforeResolve.set( + {topic: topicType}, + nowSec - message.seenTimestampSec + ); + this.metrics?.awaitingBlockGossipMessages.resolve.inc({topic: topicType}); this.pushPendingGossipsubMessageToQueue(message); count++; // don't want to block the event loop, worse case it'd wait for 16_084 / 1024 * 50ms = 800ms which is not a big deal - if (count === MAX_UNKNOWN_BLOCK_GOSSIP_OBJECTS_PER_TICK) { + if (count === MAX_AWAITING_GOSSIP_OBJECTS_PER_TICK) { count = 0; - await sleep(PROCESS_UNKNOWN_BLOCK_GOSSIP_OBJECTS_YIELD_EVERY_MS); + await sleep(AWAITING_GOSSIP_OBJECTS_YIELD_EVERY_MS); } } - byRootGossipsubMessages.delete(rootHex); + this.unknownBlockGossipsubMessagesCount -= waitingGossipsubMessages.size; + this.awaitingBlockByRoot.delete(rootHex); } private onClockSlot(clockSlot: Slot): void { const nowSec = Date.now() / 1000; - for (const [slot, gossipMessagesByRoot] of this.awaitingGossipsubMessagesByRootBySlot.entries()) { - if (slot < clockSlot) { - for (const gossipMessages of gossipMessagesByRoot.values()) { - for (const message of gossipMessages) { - this.metrics?.reprocessGossipAttestations.reject.inc({reason: ReprocessRejectReason.expired}); - this.metrics?.reprocessGossipAttestations.waitSecBeforeReject.set( - {reason: ReprocessRejectReason.expired}, + const minSlot = clockSlot - MAX_UNKNOWN_ROOTS_SLOT_CACHE_SIZE; + + for (const [slot, roots] of this.unknownBlocksBySlot) { + if (slot > minSlot) continue; + for (const rootHex of roots) { + const gossipMessagesByRoot = this.awaitingBlockByRoot.get(rootHex); + if (gossipMessagesByRoot !== undefined) { + for (const message of gossipMessagesByRoot) { + const topicType = message.topic.type; + this.metrics?.awaitingBlockGossipMessages.reject.inc({ + topic: topicType, + reason: ReprocessRejectReason.expired, + }); + this.metrics?.awaitingBlockGossipMessages.waitSecBeforeReject.set( + {topic: topicType, reason: ReprocessRejectReason.expired}, nowSec - message.seenTimestampSec ); - // TODO: Should report the dropped job to gossip? It will be eventually pruned from the mcache + // No need to report the dropped job to gossip. It will be eventually pruned from the mcache } + this.unknownBlockGossipsubMessagesCount -= gossipMessagesByRoot.size; + this.awaitingBlockByRoot.delete(rootHex); } - this.awaitingGossipsubMessagesByRootBySlot.delete(slot); } + this.unknownBlocksBySlot.delete(slot); + } + + if (this.unknownBlocksBySlot.size === 0) { + this.unknownBlockGossipsubMessagesCount = 0; } - pruneSetToMax(this.unknownRootsBySlot, MAX_UNKNOWN_ROOTS_SLOT_CACHE_SIZE); - this.unknownBlockGossipsubMessagesCount = 0; } private executeWork(): void { diff --git a/packages/beacon-node/src/sync/unknownBlock.ts b/packages/beacon-node/src/sync/unknownBlock.ts index ad3e5cbc2ad9..fcb7e961d9b9 100644 --- a/packages/beacon-node/src/sync/unknownBlock.ts +++ b/packages/beacon-node/src/sync/unknownBlock.ts @@ -64,7 +64,7 @@ enum FetchResult { * * - publishBlock * - gossipHandlers - * - searchUnknownSlotRoot + * - searchUnknownBlock * = produceSyncCommitteeContribution * = validateGossipFnRetryUnknownRoot * * submitPoolAttestationsV2 From e12fbeaba7426cbdeff871dc490d77da76bf842c Mon Sep 17 00:00:00 2001 From: Tuyen Nguyen Date: Thu, 19 Mar 2026 13:52:46 +0700 Subject: [PATCH 2/4] fix: keep earliestPermissableSlot logic the same --- packages/beacon-node/src/network/processor/index.ts | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/packages/beacon-node/src/network/processor/index.ts b/packages/beacon-node/src/network/processor/index.ts index cf597bd95639..f8433e84e8ce 100644 --- a/packages/beacon-node/src/network/processor/index.ts +++ b/packages/beacon-node/src/network/processor/index.ts @@ -1,4 +1,6 @@ import {routes} from "@lodestar/api"; +import {ForkSeq} from "@lodestar/params"; +import {computeStartSlotAtEpoch} from "@lodestar/state-transition"; import {RootHex, Slot, SlotRootHex} from "@lodestar/types"; import {Logger, MapDef, mapValues, sleep} from "@lodestar/utils"; import {BlockInputSource} from "../../chain/blocks/blockInput/types.js"; @@ -264,7 +266,12 @@ export class NetworkProcessor { // DOS protection: avoid processing messages that are too old const {slot, root} = slotRoot; const clockSlot = this.chain.clock.currentSlot; - const earliestPermissableSlot = clockSlot - DEFAULT_EARLIEST_PERMISSIBLE_SLOT_DISTANCE; + const {fork} = message.topic.boundary; + let earliestPermissableSlot = clockSlot - DEFAULT_EARLIEST_PERMISSIBLE_SLOT_DISTANCE; + if (ForkSeq[fork] >= ForkSeq.deneb && topicType === GossipType.beacon_attestation) { + // post deneb, the attestations could be in current or previous epoch + earliestPermissableSlot = computeStartSlotAtEpoch(this.chain.clock.currentEpoch - 1); + } if (slot < earliestPermissableSlot) { // No need to report the dropped job to gossip. It will be eventually pruned from the mcache this.metrics?.networkProcessor.gossipValidationError.inc({ @@ -312,8 +319,8 @@ export class NetworkProcessor { } private async onBlockProcessed({block: rootHex}: {block: string; executionOptimistic: boolean}): Promise { - const waitingGossipsubMessages = this.awaitingBlockByRoot.getOrDefault(rootHex); - if (waitingGossipsubMessages.size === 0) { + const waitingGossipsubMessages = this.awaitingBlockByRoot.get(rootHex); + if (!waitingGossipsubMessages || waitingGossipsubMessages.size === 0) { return; } From 75aa8e37563fc1fd3ef37a9c4d646b48ed17f7e8 Mon Sep 17 00:00:00 2001 From: Tuyen Nguyen Date: Mon, 23 Mar 2026 14:22:24 +0700 Subject: [PATCH 3/4] fix: implement unknownBlockGossipsubMessagesCount() as getter --- .../network/processor/extractSlotRootFns.ts | 2 +- .../src/network/processor/index.ts | 22 +++++++++---------- 2 files changed, 12 insertions(+), 12 deletions(-) diff --git a/packages/beacon-node/src/network/processor/extractSlotRootFns.ts b/packages/beacon-node/src/network/processor/extractSlotRootFns.ts index 1cf09716bbc6..77a4ecdd4ec1 100644 --- a/packages/beacon-node/src/network/processor/extractSlotRootFns.ts +++ b/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 { diff --git a/packages/beacon-node/src/network/processor/index.ts b/packages/beacon-node/src/network/processor/index.ts index f8433e84e8ce..0766dd9b5882 100644 --- a/packages/beacon-node/src/network/processor/index.ts +++ b/packages/beacon-node/src/network/processor/index.ts @@ -159,7 +159,6 @@ export class NetworkProcessor { // we may not receive the block for messages like Attestation and SignedAggregateAndProof messages, in that case PendingGossipsubMessage needs // to be stored in this Map and reprocessed once the block comes private readonly awaitingBlockByRoot: MapDef>; - private unknownBlockGossipsubMessagesCount = 0; private unknownBlocksBySlot = new MapDef>(() => new Set()); constructor( @@ -299,7 +298,6 @@ export class NetworkProcessor { this.metrics?.awaitingBlockGossipMessages.queue.inc({topic: topicType}); const awaitingGossipsubMessages = this.awaitingBlockByRoot.getOrDefault(root); awaitingGossipsubMessages.add(message); - this.unknownBlockGossipsubMessagesCount++; return; } @@ -344,7 +342,6 @@ export class NetworkProcessor { } } - this.unknownBlockGossipsubMessagesCount -= waitingGossipsubMessages.size; this.awaitingBlockByRoot.delete(rootHex); } @@ -355,9 +352,9 @@ export class NetworkProcessor { for (const [slot, roots] of this.unknownBlocksBySlot) { if (slot > minSlot) continue; for (const rootHex of roots) { - const gossipMessagesByRoot = this.awaitingBlockByRoot.get(rootHex); - if (gossipMessagesByRoot !== undefined) { - for (const message of gossipMessagesByRoot) { + const gossipMessages = this.awaitingBlockByRoot.get(rootHex); + if (gossipMessages !== undefined) { + for (const message of gossipMessages) { const topicType = message.topic.type; this.metrics?.awaitingBlockGossipMessages.reject.inc({ topic: topicType, @@ -369,16 +366,11 @@ export class NetworkProcessor { ); // No need to report the dropped job to gossip. It will be eventually pruned from the mcache } - this.unknownBlockGossipsubMessagesCount -= gossipMessagesByRoot.size; this.awaitingBlockByRoot.delete(rootHex); } } this.unknownBlocksBySlot.delete(slot); } - - if (this.unknownBlocksBySlot.size === 0) { - this.unknownBlockGossipsubMessagesCount = 0; - } } private executeWork(): void { @@ -517,4 +509,12 @@ export class NetworkProcessor { return null; } + + private get unknownBlockGossipsubMessagesCount(): number { + let count = 0; + for (const messages of this.awaitingBlockByRoot.values()) { + count += messages.size; + } + return count; + } } From 305cc7a5c05b7f8442169372ed3d05dc8aaa8a8d Mon Sep 17 00:00:00 2001 From: Tuyen Nguyen Date: Mon, 23 Mar 2026 14:29:56 +0700 Subject: [PATCH 4/4] =?UTF-8?q?refactor:=20rename=20awaitingBlockByRoot?= =?UTF-8?q?=E2=86=92awaitingMessagesByBlockRoot?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../beacon-node/src/network/processor/index.ts | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) diff --git a/packages/beacon-node/src/network/processor/index.ts b/packages/beacon-node/src/network/processor/index.ts index 0766dd9b5882..73e52c382487 100644 --- a/packages/beacon-node/src/network/processor/index.ts +++ b/packages/beacon-node/src/network/processor/index.ts @@ -158,7 +158,7 @@ export class NetworkProcessor { private readonly extractBlockSlotRootFns = createExtractBlockSlotRootFns(); // we may not receive the block for messages like Attestation and SignedAggregateAndProof messages, in that case PendingGossipsubMessage needs // to be stored in this Map and reprocessed once the block comes - private readonly awaitingBlockByRoot: MapDef>; + private readonly awaitingMessagesByBlockRoot: MapDef>; private unknownBlocksBySlot = new MapDef>(() => new Set()); constructor( @@ -183,7 +183,7 @@ export class NetworkProcessor { this.chain.emitter.on(routes.events.EventType.block, this.onBlockProcessed.bind(this)); this.chain.clock.on(ClockEvent.slot, this.onClockSlot.bind(this)); - this.awaitingBlockByRoot = new MapDef>(() => new Set()); + this.awaitingMessagesByBlockRoot = new MapDef>(() => new Set()); // TODO: Implement queues and priorization for ReqResp incoming requests // Listens to NetworkEvent.reqRespIncomingRequest event @@ -238,7 +238,7 @@ export class NetworkProcessor { searchUnknownBlock({slot, root}: SlotRootHex, source: BlockInputSource, peer?: PeerIdStr): void { if ( this.chain.seenBlock(root) || - this.awaitingBlockByRoot.has(root) || + this.awaitingMessagesByBlockRoot.has(root) || this.unknownBlocksBySlot.getOrDefault(slot).has(root) ) { return; @@ -296,7 +296,7 @@ export class NetworkProcessor { } this.metrics?.awaitingBlockGossipMessages.queue.inc({topic: topicType}); - const awaitingGossipsubMessages = this.awaitingBlockByRoot.getOrDefault(root); + const awaitingGossipsubMessages = this.awaitingMessagesByBlockRoot.getOrDefault(root); awaitingGossipsubMessages.add(message); return; } @@ -317,7 +317,7 @@ export class NetworkProcessor { } private async onBlockProcessed({block: rootHex}: {block: string; executionOptimistic: boolean}): Promise { - const waitingGossipsubMessages = this.awaitingBlockByRoot.get(rootHex); + const waitingGossipsubMessages = this.awaitingMessagesByBlockRoot.get(rootHex); if (!waitingGossipsubMessages || waitingGossipsubMessages.size === 0) { return; } @@ -342,7 +342,7 @@ export class NetworkProcessor { } } - this.awaitingBlockByRoot.delete(rootHex); + this.awaitingMessagesByBlockRoot.delete(rootHex); } private onClockSlot(clockSlot: Slot): void { @@ -352,7 +352,7 @@ export class NetworkProcessor { for (const [slot, roots] of this.unknownBlocksBySlot) { if (slot > minSlot) continue; for (const rootHex of roots) { - const gossipMessages = this.awaitingBlockByRoot.get(rootHex); + const gossipMessages = this.awaitingMessagesByBlockRoot.get(rootHex); if (gossipMessages !== undefined) { for (const message of gossipMessages) { const topicType = message.topic.type; @@ -366,7 +366,7 @@ export class NetworkProcessor { ); // No need to report the dropped job to gossip. It will be eventually pruned from the mcache } - this.awaitingBlockByRoot.delete(rootHex); + this.awaitingMessagesByBlockRoot.delete(rootHex); } } this.unknownBlocksBySlot.delete(slot); @@ -512,7 +512,7 @@ export class NetworkProcessor { private get unknownBlockGossipsubMessagesCount(): number { let count = 0; - for (const messages of this.awaitingBlockByRoot.values()) { + for (const messages of this.awaitingMessagesByBlockRoot.values()) { count += messages.size; } return count;