diff --git a/packages/beacon-node/src/chain/errors/gossipValidation.ts b/packages/beacon-node/src/chain/errors/gossipValidation.ts index fbf9fd7086d6..322696c274dc 100644 --- a/packages/beacon-node/src/chain/errors/gossipValidation.ts +++ b/packages/beacon-node/src/chain/errors/gossipValidation.ts @@ -3,6 +3,7 @@ import {LodestarError} from "@lodestar/utils"; export enum GossipAction { IGNORE = "IGNORE", REJECT = "REJECT", + RETRY_UNKNOWN_BLOCK = "RETRY_UNKNOWN_BLOCK", } export class GossipActionError extends LodestarError { diff --git a/packages/beacon-node/src/chain/reprocess.ts b/packages/beacon-node/src/chain/reprocess.ts index 013b05bf0a9a..ab05f6c2faa6 100644 --- a/packages/beacon-node/src/chain/reprocess.ts +++ b/packages/beacon-node/src/chain/reprocess.ts @@ -1,4 +1,4 @@ -import {Slot, RootHex} from "@lodestar/types"; +import {Slot, RootHex, SlotRoot} from "@lodestar/types"; import {MapDef} from "@lodestar/utils"; import {Metrics} from "../metrics/index.js"; @@ -34,8 +34,6 @@ type AwaitingAttestationPromise = { // How many attestations (aggregate + unaggregate) we keep before new ones get dropped. const MAXIMUM_QUEUED_ATTESTATIONS = 16_384; -type SlotRoot = {slot: Slot; root: RootHex}; - /** * Some attestations may reach our node before the voted block, so we manage a cache to reprocess them * when the block come. @@ -61,10 +59,10 @@ export class ReprocessController { * @returns true if blockFound */ waitForBlockOfAttestation(slot: Slot, root: RootHex): Promise { - this.metrics?.reprocessAttestations.total.inc(); + this.metrics?.reprocessApiAttestations.total.inc(); if (this.awaitingPromisesCount >= MAXIMUM_QUEUED_ATTESTATIONS) { - this.metrics?.reprocessAttestations.reject.inc({reason: ReprocessStatus.reached_limit}); + this.metrics?.reprocessApiAttestations.reject.inc({reason: ReprocessStatus.reached_limit}); return Promise.resolve(false); } @@ -116,8 +114,8 @@ export class ReprocessController { const {resolve, addedTimeMs, awaitingAttestationsCount} = awaitingPromise; resolve(true); this.awaitingPromisesCount -= awaitingAttestationsCount; - this.metrics?.reprocessAttestations.resolve.inc(awaitingAttestationsCount); - this.metrics?.reprocessAttestations.waitTimeBeforeResolve.set((Date.now() - addedTimeMs) / 1000); + this.metrics?.reprocessApiAttestations.resolve.inc(awaitingAttestationsCount); + this.metrics?.reprocessApiAttestations.waitSecBeforeResolve.set((Date.now() - addedTimeMs) / 1000); } // prune @@ -140,8 +138,8 @@ export class ReprocessController { for (const awaitingPromise of awaitingPromisesByRoot.values()) { const {resolve, addedTimeMs} = awaitingPromise; resolve(false); - this.metrics?.reprocessAttestations.waitTimeBeforeReject.set((now - addedTimeMs) / 1000); - this.metrics?.reprocessAttestations.reject.inc({reason: ReprocessStatus.expired}); + this.metrics?.reprocessApiAttestations.waitSecBeforeReject.set((now - addedTimeMs) / 1000); + this.metrics?.reprocessApiAttestations.reject.inc({reason: ReprocessStatus.expired}); } // prune diff --git a/packages/beacon-node/src/chain/validation/attestation.ts b/packages/beacon-node/src/chain/validation/attestation.ts index b463dbdf9533..2280f565f654 100644 --- a/packages/beacon-node/src/chain/validation/attestation.ts +++ b/packages/beacon-node/src/chain/validation/attestation.ts @@ -226,7 +226,8 @@ function verifyHeadBlockIsKnown(chain: IBeaconChain, beaconBlockRoot: Root): Pro const headBlock = chain.forkChoice.getBlock(beaconBlockRoot); if (headBlock === null) { - throw new AttestationError(GossipAction.IGNORE, { + // should retry the attestation when a block with beaconBlockRoot comes + throw new AttestationError(GossipAction.RETRY_UNKNOWN_BLOCK, { code: AttestationErrorCode.UNKNOWN_OR_PREFINALIZED_BEACON_BLOCK_ROOT, root: toHexString(beaconBlockRoot as typeof beaconBlockRoot), }); diff --git a/packages/beacon-node/src/metrics/metrics/lodestar.ts b/packages/beacon-node/src/metrics/metrics/lodestar.ts index 2c2f0cc825b3..bffdeae0d985 100644 --- a/packages/beacon-node/src/metrics/metrics/lodestar.ts +++ b/packages/beacon-node/src/metrics/metrics/lodestar.ts @@ -237,6 +237,16 @@ export function createLodestarMetrics( help: "Count of total gossip validation reject", labelNames: ["topic"], }), + gossipValidationRetry: register.gauge<"topic">({ + name: "lodestar_gossip_validation_retry_total", + help: "Count of total gossip validation retry", + labelNames: ["topic"], + }), + gossipValidationReprocess: register.gauge<"topic">({ + name: "lodestar_gossip_validation_reprocess_total", + help: "Count of total gossip validation reprocess", + labelNames: ["topic"], + }), gossipValidationError: register.gauge<"topic" | "error">({ name: "lodestar_gossip_validation_error_total", help: "Count of total gossip validation errors detailed", @@ -1188,8 +1198,8 @@ export function createLodestarMetrics( }), }, - // reprocess attestations - reprocessAttestations: { + // reprocess api attestations + reprocessApiAttestations: { total: register.gauge({ name: "lodestar_reprocess_attestations_total", help: "Total number of attestations waiting to reprocess", @@ -1198,7 +1208,7 @@ export function createLodestarMetrics( name: "lodestar_reprocess_attestations_resolve_total", help: "Total number of attestations are reprocessed", }), - waitTimeBeforeResolve: register.gauge({ + waitSecBeforeResolve: register.gauge({ name: "lodestar_reprocess_attestations_wait_time_resolve_seconds", help: "Time to wait for unknown block in seconds", }), @@ -1207,12 +1217,41 @@ export function createLodestarMetrics( help: "Total number of attestations are rejected to reprocess", labelNames: ["reason"], }), - waitTimeBeforeReject: register.gauge<"reason">({ + waitSecBeforeReject: register.gauge<"reason">({ name: "lodestar_reprocess_attestations_wait_time_reject_seconds", help: "Time to wait for unknown block before being rejected", }), }, + // reprocess gossip attestations + reprocessGossipAttestations: { + total: register.gauge({ + name: "lodestar_reprocess_gossip_attestations_total", + help: "Total number of gossip attestations waiting to reprocess", + }), + countPerSlot: register.gauge({ + name: "lodestar_reprocess_gossip_attestations_per_slot_total", + help: "Total number of gossip attestations waiting to reprocess pet slot", + }), + resolve: register.gauge({ + name: "lodestar_reprocess_gossip_attestations_resolve_total", + help: "Total number of gossip attestations are reprocessed", + }), + waitSecBeforeResolve: register.gauge({ + name: "lodestar_reprocess_gossip_attestations_wait_time_resolve_seconds", + help: "Time to wait for unknown block in seconds", + }), + reject: register.gauge<"reason">({ + name: "lodestar_reprocess_gossip_attestations_reject_total", + help: "Total number of attestations are rejected to reprocess", + labelNames: ["reason"], + }), + waitSecBeforeReject: register.gauge<"reason">({ + name: "lodestar_reprocess_gossip_attestations_wait_time_reject_seconds", + help: "Time to wait for unknown block before being rejected", + }), + }, + lightclientServer: { onSyncAggregate: register.gauge<"event">({ name: "lodestar_lightclient_server_on_sync_aggregate_event_total", diff --git a/packages/beacon-node/src/network/events.ts b/packages/beacon-node/src/network/events.ts index 2111afaa4c04..cdce6989f312 100644 --- a/packages/beacon-node/src/network/events.ts +++ b/packages/beacon-node/src/network/events.ts @@ -19,15 +19,21 @@ export enum NetworkEvent { // Network processor events pendingGossipsubMessage = "gossip.pendingGossipsubMessage", + reprocessGossipsubMessage = "gossip.reprocessGossipsubMessage", gossipMessageValidationResult = "gossip.messageValidationResult", } +export enum ReprocessGossipMessageType { + unknownBlock = "unknownBlock", +} + export type NetworkEvents = { [NetworkEvent.peerConnected]: (peer: PeerId, status: phase0.Status) => void; [NetworkEvent.peerDisconnected]: (peer: PeerId) => void; [NetworkEvent.reqRespRequest]: (request: RequestTypedContainer, peer: PeerId) => void; [NetworkEvent.unknownBlockParent]: (blockInput: BlockInput, peerIdStr: string) => void; [NetworkEvent.pendingGossipsubMessage]: (data: PendingGossipsubMessage) => void; + [NetworkEvent.reprocessGossipsubMessage]: (data: PendingGossipsubMessage, type: ReprocessGossipMessageType) => void; [NetworkEvent.gossipMessageValidationResult]: ( msgId: string, propagationSource: PeerId, diff --git a/packages/beacon-node/src/network/gossip/gossipsub.ts b/packages/beacon-node/src/network/gossip/gossipsub.ts index 8625e45a9192..e9f925e00b9c 100644 --- a/packages/beacon-node/src/network/gossip/gossipsub.ts +++ b/packages/beacon-node/src/network/gossip/gossipsub.ts @@ -404,6 +404,7 @@ export class Eth2Gossipsub extends GossipSub implements GossipBeaconNode { propagationSource, seenTimestampSec, startProcessUnixSec: null, + gossipObject: null, }); } diff --git a/packages/beacon-node/src/network/gossip/interface.ts b/packages/beacon-node/src/network/gossip/interface.ts index 4fbc550f457e..77dc357f2dbc 100644 --- a/packages/beacon-node/src/network/gossip/interface.ts +++ b/packages/beacon-node/src/network/gossip/interface.ts @@ -4,7 +4,7 @@ import {Message, TopicValidatorResult} from "@libp2p/interface-pubsub"; import StrictEventEmitter from "strict-event-emitter-types"; import {PeerIdStr} from "@chainsafe/libp2p-gossipsub/types"; import {ForkName} from "@lodestar/params"; -import {allForks, altair, capella, deneb, phase0} from "@lodestar/types"; +import {allForks, altair, capella, deneb, phase0, SlotRoot} from "@lodestar/types"; import {BeaconConfig} from "@lodestar/config"; import {Logger} from "@lodestar/utils"; import {IBeaconChain} from "../../chain/index.js"; @@ -151,9 +151,12 @@ export type GossipBeaconNode = { export type GossipValidatorFn = ( topic: GossipTopic, msg: Message, + object: GossipTypeMap[GossipType] | null, propagationSource: PeerIdStr, seenTimestampSec: number -) => Promise; +) => Promise< + {type: "done"; result: TopicValidatorResult} | {type: "retryUnknownBlock"; gossipObject: GossipTypeMap[GossipType]} +>; export type ValidatorFnsByType = {[K in GossipType]: GossipValidatorFn}; @@ -167,6 +170,7 @@ export type GossipHandlerFn = ( peerIdStr: string, seenTimestampSec: number ) => Promise; + export type GossipHandlers = { [K in GossipType]: ( object: GossipTypeMap[K], @@ -176,6 +180,11 @@ export type GossipHandlers = { ) => Promise; }; +export type UnknownBlockFns = { + [K in GossipType]: (object: GossipTypeMap[K]) => SlotRoot; +}; +export type UnknownBlockFromGossipObjectFn = (object: GossipTypeMap[GossipType]) => SlotRoot; + // eslint-disable-next-line @typescript-eslint/no-explicit-any export type ResolvedType Promise> = F extends (...args: any) => Promise ? T diff --git a/packages/beacon-node/src/network/processor/gossipHandlers.ts b/packages/beacon-node/src/network/processor/gossipHandlers.ts index 24debc783fd0..452eea272404 100644 --- a/packages/beacon-node/src/network/processor/gossipHandlers.ts +++ b/packages/beacon-node/src/network/processor/gossipHandlers.ts @@ -194,14 +194,9 @@ export function getGossipHandlers(modules: ValidatorFnsModules, options: GossipH [GossipType.beacon_aggregate_and_proof]: async (signedAggregateAndProof, _topic, _peer, seenTimestampSec) => { let validationResult: {indexedAttestation: phase0.IndexedAttestation; committeeIndices: number[]}; try { - // If an attestation refers to a block root that's not known, it will wait for 1 slot max - // See https://github.com/ChainSafe/lodestar/pull/3564 for reasoning and results - // Waiting here requires minimal code and automatically affects attestation, and aggregate validation - // both from gossip and the API. I also prevents having to catch and re-throw in multiple places. - // eslint-disable-next-line @typescript-eslint/explicit-function-return-type - const validateFn = () => validateGossipAggregateAndProof(chain, signedAggregateAndProof); - const {slot, beaconBlockRoot} = signedAggregateAndProof.message.aggregate.data; - validationResult = await validateGossipFnRetryUnknownRoot(validateFn, chain, slot, beaconBlockRoot); + // validating attestations may throw UNKNOWN_BLOCK error. In that case the NetworkProcessor will have + // to reprocess GossipMessage when the block comes + validationResult = await validateGossipAggregateAndProof(chain, signedAggregateAndProof); } catch (e) { if (e instanceof AttestationError && e.action === GossipAction.REJECT) { chain.persistInvalidSszValue(ssz.phase0.SignedAggregateAndProof, signedAggregateAndProof, "gossip_reject"); @@ -236,14 +231,9 @@ export function getGossipHandlers(modules: ValidatorFnsModules, options: GossipH [GossipType.beacon_attestation]: async (attestation, {subnet}, _peer, seenTimestampSec) => { let validationResult: {indexedAttestation: phase0.IndexedAttestation; subnet: number}; try { - // eslint-disable-next-line @typescript-eslint/explicit-function-return-type - const validateFn = () => validateGossipAttestation(chain, attestation, subnet); - const {slot, beaconBlockRoot} = attestation.data; - // If an attestation refers to a block root that's not known, it will wait for 1 slot max - // See https://github.com/ChainSafe/lodestar/pull/3564 for reasoning and results - // Waiting here requires minimal code and automatically affects attestation, and aggregate validation - // both from gossip and the API. I also prevents having to catch and re-throw in multiple places. - validationResult = await validateGossipFnRetryUnknownRoot(validateFn, chain, slot, beaconBlockRoot); + // validating attestations may throw UNKNOWN_BLOCK error. In that case the NetworkProcessor will have + // to reprocess GossipMessage when the block comes + validationResult = await validateGossipAttestation(chain, attestation, subnet); } catch (e) { if (e instanceof AttestationError && e.action === GossipAction.REJECT) { chain.persistInvalidSszValue(ssz.phase0.Attestation, attestation, "gossip_reject"); diff --git a/packages/beacon-node/src/network/processor/gossipValidatorFn.ts b/packages/beacon-node/src/network/processor/gossipValidatorFn.ts index 00b5b3f4a6e9..e59888850543 100644 --- a/packages/beacon-node/src/network/processor/gossipValidatorFn.ts +++ b/packages/beacon-node/src/network/processor/gossipValidatorFn.ts @@ -29,31 +29,36 @@ export type ValidatorFnModules = { export function getGossipValidatorFn(gossipHandlers: GossipHandlers, modules: ValidatorFnModules): GossipValidatorFn { const {logger, metrics} = modules; - return async function gossipValidatorFn(topic, msg, propagationSource, seenTimestampSec) { + return async function gossipValidatorFn(topic, msg, obj, propagationSource, seenTimestampSec) { const type = topic.type; // Define in scope above try {} to be used in catch {} if object was parsed - let gossipObject; - try { - // Deserialize object from bytes ONLY after being picked up from the validation queue + let gossipObject = obj; + // Deserialize object from bytes ONLY after being picked up from the validation queue + // Do not need to deserialize if it's a retry + if (gossipObject) { + metrics?.gossipValidationReprocess.inc({topic: type}); + } else { try { const sszType = getGossipSSZType(topic); gossipObject = sszType.deserialize(msg.data); } catch (e) { // TODO: Log the error or do something better with it - return TopicValidatorResult.Reject; + return {type: "done", result: TopicValidatorResult.Reject}; } + } + try { await (gossipHandlers[topic.type] as GossipHandlerFn)(gossipObject, topic, propagationSource, seenTimestampSec); metrics?.gossipValidationAccept.inc({topic: type}); - return TopicValidatorResult.Accept; + return {type: "done", result: TopicValidatorResult.Accept}; } catch (e) { if (!(e instanceof GossipActionError)) { // not deserve to log error here, it looks too dangerous to users logger.debug(`Gossip validation ${type} threw a non-GossipActionError`, {}, e as Error); - return TopicValidatorResult.Ignore; + return {type: "done", result: TopicValidatorResult.Ignore}; } // Metrics on specific error reason @@ -63,11 +68,15 @@ export function getGossipValidatorFn(gossipHandlers: GossipHandlers, modules: Va switch (e.action) { case GossipAction.IGNORE: metrics?.gossipValidationIgnore.inc({topic: type}); - return TopicValidatorResult.Ignore; + return {type: "done", result: TopicValidatorResult.Ignore}; case GossipAction.REJECT: metrics?.gossipValidationReject.inc({topic: type}); - return TopicValidatorResult.Reject; + return {type: "done", result: TopicValidatorResult.Reject}; + + case GossipAction.RETRY_UNKNOWN_BLOCK: + metrics?.gossipValidationRetry.inc({topic: type}); + return {type: "retryUnknownBlock", gossipObject}; } } }; diff --git a/packages/beacon-node/src/network/processor/index.ts b/packages/beacon-node/src/network/processor/index.ts index e1f37f8da531..5cb162ca8eb8 100644 --- a/packages/beacon-node/src/network/processor/index.ts +++ b/packages/beacon-node/src/network/processor/index.ts @@ -1,12 +1,17 @@ -import {Logger, mapValues} from "@lodestar/utils"; +import {TopicValidatorResult} from "@libp2p/interface-pubsub"; +import {Logger, MapDef, mapValues} from "@lodestar/utils"; +import {RootHex, Slot} from "@lodestar/types"; +import {routes} from "@lodestar/api"; import {IBeaconChain} from "../../chain/interface.js"; import {Metrics} from "../../metrics/metrics.js"; -import {NetworkEvent, NetworkEventBus} from "../events.js"; -import {GossipType} from "../gossip/interface.js"; +import {NetworkEvent, NetworkEventBus, ReprocessGossipMessageType} from "../events.js"; +import {GossipType, UnknownBlockFromGossipObjectFn} from "../gossip/interface.js"; +import {ChainEvent} from "../../chain/emitter.js"; import {createGossipQueues} from "./gossipQueues.js"; import {NetworkWorker, NetworkWorkerModules} from "./worker.js"; -import {PendingGossipsubMessage} from "./types.js"; +import {PendingGossipsubMessage, WaitingGossipsubMessage} from "./types.js"; import {ValidatorFnsModules, GossipHandlerOpts} from "./gossipHandlers.js"; +import {createUnknownBlockFromGossipObjectFns} from "./unknownBlockFromGossipObject.js"; export type NetworkProcessorModules = NetworkWorkerModules & ValidatorFnsModules & { @@ -39,6 +44,23 @@ 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. +const MAXIMUM_QUEUED_UNKNOWN_BLOCK_GOSSIP_OBJECTS = 16_384; + +/** + * Reprocess reject reason for metrics + */ +enum ReprocessRejectReason { + /** + * There are too many attestations that have unknown block root. + */ + reached_limit = "reached_limit", + /** + * The awaiting attestation is pruned per clock slot. + */ + expired = "expired", +} + /** * Network processor handles the gossip queues and throtles processing to not overload the main thread * - Decides when to process work and what to process @@ -58,23 +80,44 @@ const MAX_JOBS_SUBMITTED_PER_TICK = 128; * * The gossip queues should receive "backpressue" from the regen and BLS workers queues. * Such that enough work is processed to fill either one of the queue. + * + * ### WaitingGossipsubMessage beacon_attestation example + * 1. beacon_attestation gossip message passes through gossipQueues and executeWork() + * 2. During gossip validation, it throws UNKNOWN_BLOCK error + * 3. A WaitingGossipsubMessage instance is created which is the same to PendingGossipsubMessage with addedTimeMs + * 4. WaitingGossipsubMessage is then put to a temporary queue implemented as slot/root hex map (deleted per slot) + * 5. Once the block comes with same slot/root WaitingGossipsubMessage is then pushed back to gossipQueues again */ export class NetworkProcessor { private readonly worker: NetworkWorker; private readonly chain: IBeaconChain; + private readonly events: NetworkEventBus; private readonly logger: Logger; private readonly metrics: Metrics | null; private readonly gossipQueues = createGossipQueues(); private readonly gossipTopicConcurrency = mapValues(this.gossipQueues, () => 0); + private readonly unknownBlockFns = createUnknownBlockFromGossipObjectFns(); + // validating GossipMessage may result in UNKNOWN_BLOCK error, in that case PendingGossipsubMessage needs + // to be stored in this Map and reprocessed once the block comes + private readonly awaitingGossipsubMessagesByRootBySlot: MapDef>>; + private unknownBlockGossipsubMessagesCount = 0; constructor(modules: NetworkProcessorModules, private readonly opts: NetworkProcessorOpts) { const {chain, events, logger, metrics} = modules; this.chain = chain; + this.events = events; this.metrics = metrics; this.logger = logger; this.worker = new NetworkWorker(modules, opts); events.on(NetworkEvent.pendingGossipsubMessage, this.onPendingGossipsubMessage.bind(this)); + events.on(NetworkEvent.reprocessGossipsubMessage, this.onReprocessGossipsubMessage.bind(this)); + this.chain.emitter.on(routes.events.EventType.block, this.onBlockProcessed.bind(this)); + this.chain.emitter.on(ChainEvent.clockSlot, this.onClockSlot.bind(this)); + + this.awaitingGossipsubMessagesByRootBySlot = new MapDef( + () => new MapDef>(() => new Set()) + ); if (metrics) { metrics.gossipValidationQueueLength.addCollect(() => { @@ -82,6 +125,7 @@ export class NetworkProcessor { metrics.gossipValidationQueueLength.set({topic}, this.gossipQueues[topic].length); metrics.gossipValidationQueueConcurrency.set({topic}, this.gossipTopicConcurrency[topic]); } + metrics.reprocessGossipAttestations.countPerSlot.set(this.unknownBlockGossipsubMessagesCount); }); } @@ -90,6 +134,13 @@ export class NetworkProcessor { // this.regen.onAvailable(() => this.executeWork()); } + async stop(): Promise { + this.events.off(NetworkEvent.pendingGossipsubMessage, this.onPendingGossipsubMessage); + this.events.off(NetworkEvent.reprocessGossipsubMessage, this.onReprocessGossipsubMessage); + this.chain.emitter.off(routes.events.EventType.block, this.onBlockProcessed); + this.chain.emitter.off(ChainEvent.clockSlot, this.onClockSlot); + } + dropAllJobs(): void { for (const topic of executeGossipWorkOrder) { this.gossipQueues[topic].clear(); @@ -116,6 +167,65 @@ export class NetworkProcessor { this.executeWork(); } + private onReprocessGossipsubMessage(data: PendingGossipsubMessage, reprocessType: ReprocessGossipMessageType): void { + if (!data.gossipObject) { + throw Error("Should have gossip object after the 1st gossip validation"); + } + if (reprocessType === ReprocessGossipMessageType.unknownBlock) { + if (this.unknownBlockGossipsubMessagesCount > MAXIMUM_QUEUED_UNKNOWN_BLOCK_GOSSIP_OBJECTS) { + this.metrics?.reprocessGossipAttestations.reject.inc({reason: ReprocessRejectReason.reached_limit}); + return; + } + + this.metrics?.reprocessGossipAttestations.total.inc(); + const {slot, root} = (this.unknownBlockFns[data.topic.type] as UnknownBlockFromGossipObjectFn)(data.gossipObject); + const awaitingGossipsubMessagesByRoot = this.awaitingGossipsubMessagesByRootBySlot.getOrDefault(slot); + const awaitingGossipsubMessages = awaitingGossipsubMessagesByRoot.getOrDefault(root); + (data as WaitingGossipsubMessage).addedTimeMs = Date.now(); + awaitingGossipsubMessages.add(data as WaitingGossipsubMessage); + this.unknownBlockGossipsubMessagesCount++; + } + } + + private onBlockProcessed({slot, block: rootHex}: {slot: Slot; block: string; executionOptimistic: boolean}): void { + const byRootGossipsubMessages = this.awaitingGossipsubMessagesByRootBySlot.getOrDefault(slot); + const waitingGossipsubMessages = byRootGossipsubMessages.getOrDefault(rootHex); + if (waitingGossipsubMessages.size === 0) { + return; + } + + this.metrics?.reprocessGossipAttestations.resolve.inc(waitingGossipsubMessages.size); + const now = Date.now(); + waitingGossipsubMessages.forEach((msg) => { + this.metrics?.reprocessGossipAttestations.waitSecBeforeResolve.set((now - msg.addedTimeMs) / 1000); + this.onPendingGossipsubMessage(msg); + }); + + byRootGossipsubMessages.delete(rootHex); + } + + private onClockSlot(clockSlot: Slot): void { + const now = Date.now(); + for (const [slot, gossipMessagesByRoot] of this.awaitingGossipsubMessagesByRootBySlot.entries()) { + if (slot < clockSlot) { + for (const gossipMessages of gossipMessagesByRoot.values()) { + gossipMessages.forEach((message) => { + this.metrics?.reprocessGossipAttestations.reject.inc({reason: ReprocessRejectReason.expired}); + this.metrics?.reprocessGossipAttestations.waitSecBeforeReject.set((now - message.addedTimeMs) / 1000); + this.events.emit( + NetworkEvent.gossipMessageValidationResult, + message.msgId, + message.propagationSource, + TopicValidatorResult.Ignore + ); + }); + } + this.awaitingGossipsubMessagesByRootBySlot.delete(slot); + } + } + this.unknownBlockGossipsubMessagesCount = 0; + } + private executeWork(): void { // TODO: Maybe de-bounce by timing the last time executeWork was run diff --git a/packages/beacon-node/src/network/processor/types.ts b/packages/beacon-node/src/network/processor/types.ts index ebff7e68d437..4989eaad5909 100644 --- a/packages/beacon-node/src/network/processor/types.ts +++ b/packages/beacon-node/src/network/processor/types.ts @@ -1,6 +1,6 @@ import {PeerId} from "@libp2p/interface-peer-id"; import {Message} from "@libp2p/interface-pubsub"; -import {GossipTopic} from "../gossip/index.js"; +import {GossipTopic, GossipType, GossipTypeMap} from "../gossip/index.js"; export type GossipAttestationsWork = { messages: PendingGossipsubMessage[]; @@ -14,4 +14,9 @@ export type PendingGossipsubMessage = { propagationSource: PeerId; seenTimestampSec: number; startProcessUnixSec: number | null; + gossipObject: GossipTypeMap[GossipType] | null; +}; + +export type WaitingGossipsubMessage = PendingGossipsubMessage & { + addedTimeMs: number; }; diff --git a/packages/beacon-node/src/network/processor/unknownBlockFromGossipObject.ts b/packages/beacon-node/src/network/processor/unknownBlockFromGossipObject.ts new file mode 100644 index 000000000000..f42f6242568e --- /dev/null +++ b/packages/beacon-node/src/network/processor/unknownBlockFromGossipObject.ts @@ -0,0 +1,34 @@ +import {toHexString} from "@chainsafe/ssz"; +import {UnknownBlockFns as UnknownBlockFromGossipObjectFns, GossipType} from "../gossip/index.js"; + +export function createUnknownBlockFromGossipObjectFns(): UnknownBlockFromGossipObjectFns { + const neverReturn = (): never => { + throw Error("Not expect this gossip message to return an unknown block"); + }; + return { + [GossipType.beacon_attestation]: (attestation) => { + const {slot, beaconBlockRoot} = attestation.data; + return { + slot, + root: toHexString(beaconBlockRoot), + }; + }, + [GossipType.beacon_aggregate_and_proof]: (signedAggregateAndProof) => { + const {slot, beaconBlockRoot} = signedAggregateAndProof.message.aggregate.data; + return { + slot, + root: toHexString(beaconBlockRoot), + }; + }, + [GossipType.sync_committee]: neverReturn, + [GossipType.sync_committee_contribution_and_proof]: neverReturn, + [GossipType.beacon_block]: neverReturn, + [GossipType.beacon_block_and_blobs_sidecar]: neverReturn, + [GossipType.attester_slashing]: neverReturn, + [GossipType.proposer_slashing]: neverReturn, + [GossipType.voluntary_exit]: neverReturn, + [GossipType.bls_to_execution_change]: neverReturn, + [GossipType.light_client_finality_update]: neverReturn, + [GossipType.light_client_optimistic_update]: neverReturn, + }; +} diff --git a/packages/beacon-node/src/network/processor/worker.ts b/packages/beacon-node/src/network/processor/worker.ts index 64edfbf37077..0ee6b2bf5a20 100644 --- a/packages/beacon-node/src/network/processor/worker.ts +++ b/packages/beacon-node/src/network/processor/worker.ts @@ -1,6 +1,6 @@ import {IBeaconChain} from "../../chain/interface.js"; import {Metrics} from "../../metrics/metrics.js"; -import {NetworkEvent, NetworkEventBus} from "../events.js"; +import {NetworkEvent, NetworkEventBus, ReprocessGossipMessageType} from "../events.js"; import {GossipHandlers, GossipValidatorFn} from "../gossip/interface.js"; import {getGossipHandlers, GossipHandlerOpts, ValidatorFnsModules} from "./gossipHandlers.js"; import {getGossipValidatorFn, ValidatorFnModules} from "./gossipValidatorFn.js"; @@ -32,21 +32,34 @@ export class NetworkWorker { const acceptance = await this.gossipValidatorFn( message.topic, message.msg, + // gossipObject is only available on 2nd validation + message.gossipObject, message.propagationSource.toString(), message.seenTimestampSec ); - if (message.startProcessUnixSec !== null) { - this.metrics?.gossipValidationQueueJobWaitTime.observe( - {topic: message.topic.type}, - message.startProcessUnixSec - message.seenTimestampSec - ); - this.metrics?.gossipValidationQueueJobTime.observe( - {topic: message.topic.type}, - Date.now() / 1000 - message.startProcessUnixSec + if (acceptance.type === "done") { + if (message.startProcessUnixSec !== null) { + this.metrics?.gossipValidationQueueJobWaitTime.observe( + {topic: message.topic.type}, + message.startProcessUnixSec - message.seenTimestampSec + ); + this.metrics?.gossipValidationQueueJobTime.observe( + {topic: message.topic.type}, + Date.now() / 1000 - message.startProcessUnixSec + ); + } + + this.events.emit( + NetworkEvent.gossipMessageValidationResult, + message.msgId, + message.propagationSource, + acceptance.result ); + } else if (acceptance.type === "retryUnknownBlock") { + // we don't have to deserialize to gossip object the next time + message.gossipObject = acceptance.gossipObject; + this.events.emit(NetworkEvent.reprocessGossipsubMessage, message, ReprocessGossipMessageType.unknownBlock); } - - this.events.emit(NetworkEvent.gossipMessageValidationResult, message.msgId, message.propagationSource, acceptance); } } diff --git a/packages/types/src/types.ts b/packages/types/src/types.ts index f6be3fbaa347..16f8c905ad5f 100644 --- a/packages/types/src/types.ts +++ b/packages/types/src/types.ts @@ -1,3 +1,5 @@ +import {Slot} from "./primitive/types.js"; + export * from "./primitive/types.js"; export {ts as phase0} from "./phase0/index.js"; export {ts as altair} from "./altair/index.js"; @@ -10,6 +12,8 @@ export {ts as allForks} from "./allForks/index.js"; /** Common non-spec type to represent roots as strings */ export type RootHex = string; +export type SlotRoot = {slot: Slot; root: RootHex}; + /** Handy enum to represent the block production source */ export enum BlockSource { builder = "builder",