ChainSafe · twoeths · Mar 17, 2023
@@ -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<T extends {code: string}> extends LodestarError<T> {

@@ -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<boolean> {
-    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

@@ -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),
     });

@@ -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",

@@ -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,

@@ -404,6 +404,7 @@ export class Eth2Gossipsub extends GossipSub implements GossipBeaconNode {
       propagationSource,
       seenTimestampSec,
       startProcessUnixSec: null,
+      gossipObject: null,
     });
   }
 

@@ -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<TopicValidatorResult>;
+) => 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<void>;
+
 export type GossipHandlers = {
   [K in GossipType]: (
     object: GossipTypeMap[K],
@@ -176,6 +180,11 @@ export type GossipHandlers = {
   ) => Promise<void>;
 };
 
+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<F extends (...args: any) => Promise<any>> = F extends (...args: any) => Promise<infer T>
   ? T

@@ -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");

@@ -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};
       }
     }
   };