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/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/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..73e52c382487 100644 --- a/packages/beacon-node/src/network/processor/index.ts +++ b/packages/beacon-node/src/network/processor/index.ts @@ -2,7 +2,7 @@ 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 +87,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 +137,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 +156,10 @@ 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 unknownBlockGossipsubMessagesCount = 0; - private unknownRootsBySlot = new MapDef>(() => new Set()); + private readonly awaitingMessagesByBlockRoot: MapDef>; + private unknownBlocksBySlot = new MapDef>(() => new Set()); constructor( modules: NetworkProcessorModules, @@ -184,9 +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.awaitingGossipsubMessagesByRootBySlot = new MapDef( - () => new MapDef>(() => new Set()) - ); + this.awaitingMessagesByBlockRoot = new MapDef>(() => new Set()); // TODO: Implement queues and priorization for ReqResp incoming requests // Listens to NetworkEvent.reqRespIncomingRequest event @@ -198,7 +195,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 +230,77 @@ 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.awaitingMessagesByBlockRoot.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 {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({ + 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.awaitingMessagesByBlockRoot.getOrDefault(root); + awaitingGossipsubMessages.add(message); + return; } - // bypass the check for other messages this.pushPendingGossipsubMessageToQueue(message); } @@ -298,7 +308,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 +316,61 @@ 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); - if (waitingGossipsubMessages.size === 0) { + private async onBlockProcessed({block: rootHex}: {block: string; executionOptimistic: boolean}): Promise { + const waitingGossipsubMessages = this.awaitingMessagesByBlockRoot.get(rootHex); + if (!waitingGossipsubMessages || 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.awaitingMessagesByBlockRoot.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()) { + 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 gossipMessages = this.awaitingMessagesByBlockRoot.get(rootHex); + if (gossipMessages !== undefined) { for (const message of gossipMessages) { - this.metrics?.reprocessGossipAttestations.reject.inc({reason: ReprocessRejectReason.expired}); - this.metrics?.reprocessGossipAttestations.waitSecBeforeReject.set( - {reason: ReprocessRejectReason.expired}, + 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.awaitingMessagesByBlockRoot.delete(rootHex); } - this.awaitingGossipsubMessagesByRootBySlot.delete(slot); } + this.unknownBlocksBySlot.delete(slot); } - pruneSetToMax(this.unknownRootsBySlot, MAX_UNKNOWN_ROOTS_SLOT_CACHE_SIZE); - this.unknownBlockGossipsubMessagesCount = 0; } private executeWork(): void { @@ -496,4 +509,12 @@ export class NetworkProcessor { return null; } + + private get unknownBlockGossipsubMessagesCount(): number { + let count = 0; + for (const messages of this.awaitingMessagesByBlockRoot.values()) { + count += messages.size; + } + return count; + } } 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