feat: enhance NetworkProcessor for gloas #9025

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged

nflaig merged 21 commits into unstable from te/gloas_network_processor

Apr 2, 2026

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

            
                      Original file line number
                      Diff line number
                      Diff line change
                  
    @@ -217,7 +217,7 @@ export function getBeaconBlockApi({
  
            if (!blockLocallyProduced) {

              const parentBlock = chain.forkChoice.getBlockDefaultStatus(signedBlock.message.parentRoot);

              if (parentBlock === null) {

                chain.emitter.emit(ChainEvent.unknownParent, {

                chain.emitter.emit(ChainEvent.blockUnknownParent, {

                  blockInput: blockForImport,

                  peer: IDENTITY_PEER_ID,

                  source: BlockInputSource.api,

    @@ -312,7 +312,7 @@ export function getBeaconBlockApi({
  
              .processBlock(blockForImport, opts)

              .catch((e) => {

                if (e instanceof BlockError && e.type.code === BlockErrorCode.PARENT_UNKNOWN) {

                  chain.emitter.emit(ChainEvent.unknownParent, {

                  chain.emitter.emit(ChainEvent.blockUnknownParent, {

                    blockInput: blockForImport,

                    peer: IDENTITY_PEER_ID,

                    source: BlockInputSource.api,

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

-Original file line number
+Diff line change
@@ Expand Up / @@ -505,7 +505,11 @@ export class BeaconChain implements IBeaconChain { @@
       }
       seenBlock(blockRoot: RootHex): boolean {
-        return this.seenBlockInputCache.has(blockRoot) || this.forkChoice.hasBlockHex(blockRoot);
+        return this.seenBlockInputCache.hasBlock(blockRoot) || this.forkChoice.hasBlockHexUnsafe(blockRoot);
+      }
+      seenPayloadEnvelope(blockRoot: RootHex): boolean {
+        return this.seenPayloadEnvelopeInputCache.hasPayload(blockRoot) || this.forkChoice.hasPayloadHexUnsafe(blockRoot);
       }
       regenCanAcceptWork(): boolean {
@@ Expand Down @@

packages/beacon-node/src/chain/emitter.ts

            
                      Original file line number
                      Diff line number
                      Diff line change
                  
    @@ -4,6 +4,7 @@ import {routes} from "@lodestar/api";
  
    import {CheckpointWithPayloadStatus} from "@lodestar/fork-choice";

    import {IBeaconStateView} from "@lodestar/state-transition";

    import {DataColumnSidecar, RootHex, deneb, phase0} from "@lodestar/types";

    import {SignedExecutionPayloadEnvelope} from "@lodestar/types/gloas";

    import {PeerIdStr} from "../util/peerId.js";

    import {BlockInputSource, IBlockInput} from "./blocks/blockInput/types.js";

    @@ -54,13 +55,22 @@ export enum ChainEvent {
  
       */

      updateStatus = "updateStatus",

      /**

       * Trigger a BlockInputSync for blocks where the parentRoot is not known to fork choice

       * Trigger BlockInputSync to find parent of a SignedBeaconBlock received

       * Post-gloas, missing parent could be a SignedBeaconBlock and/or a SignedExecutionPayloadEnvelope

       */

      unknownParent = "unknownParent",

      blockUnknownParent = "blockUnknownParent",

      /**

       * Trigger BlockInputSync for objects that correspond to a block that is not known to fork choice

       * Trigger BlockInputSync to find a SignedBeaconBlock given a SignedExecutionPayloadEnvelop received

       */

      envelopeUnknownBlock = "envelopeUnknownBlock",

      /**

       * Trigger BlockInputSync to find a SignedBeaconBlock with specified block root.

       */

      unknownBlockRoot = "unknownBlockRoot",

      /**

       * Trigger BlockInputSync to find a SignedExecutionPayloadEnvelope with specified block root.

       */

      unknownEnvelopeBlockRoot = "unknownEnvelopeBlockRoot",

      /**

       * Trigger BlockInputSync for blocks that are partially received via gossip but are not complete by time the

       * cut-off window passes for waiting on gossip

    @@ -75,9 +85,15 @@ export type ReorgEventData = routes.events.EventData[routes.events.EventType.cha
  
    type ApiEvents = {[K in routes.events.EventType]: (data: routes.events.EventData[K]) => void};

    export type ChainEventData = {

      [ChainEvent.unknownParent]: {blockInput: IBlockInput; peer: PeerIdStr; source: BlockInputSource};

      [ChainEvent.blockUnknownParent]: {blockInput: IBlockInput; peer: PeerIdStr; source: BlockInputSource};

      [ChainEvent.envelopeUnknownBlock]: {

        envelope: SignedExecutionPayloadEnvelope;

        peer?: PeerIdStr;

        source: BlockInputSource;

      };

      [ChainEvent.unknownBlockRoot]: {rootHex: RootHex; peer?: PeerIdStr; source: BlockInputSource};

      [ChainEvent.incompleteBlockInput]: {blockInput: IBlockInput; peer: PeerIdStr; source: BlockInputSource};

      [ChainEvent.unknownEnvelopeBlockRoot]: {rootHex: RootHex; peer?: PeerIdStr; source: BlockInputSource};

    };

    export type IChainEvents = ApiEvents & {

    @@ -96,9 +112,11 @@ export type IChainEvents = ApiEvents & {
  
      // Sync events that are chain->chain. Initiated from network requests but do not cross the network

      // barrier so are considered ChainEvent(s).

      [ChainEvent.unknownParent]: (data: ChainEventData[ChainEvent.unknownParent]) => void;

      [ChainEvent.blockUnknownParent]: (data: ChainEventData[ChainEvent.blockUnknownParent]) => void;

      [ChainEvent.envelopeUnknownBlock]: (data: ChainEventData[ChainEvent.envelopeUnknownBlock]) => void;

      [ChainEvent.unknownBlockRoot]: (data: ChainEventData[ChainEvent.unknownBlockRoot]) => void;

      [ChainEvent.incompleteBlockInput]: (data: ChainEventData[ChainEvent.incompleteBlockInput]) => void;

      [ChainEvent.unknownEnvelopeBlockRoot]: (data: ChainEventData[ChainEvent.unknownEnvelopeBlockRoot]) => void;

    };

    /**

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

-Original file line number
+Diff line change
@@ Expand Up / @@ -159,6 +159,8 @@ export interface IBeaconChain { @@
       close(): Promise<void>;
       /** Chain has seen the specified block root or not. The block may not be processed yet, use forkchoice.hasBlock to check it  */
       seenBlock(blockRoot: RootHex): boolean;
+      /** Chain has seen a SignedExecutionPayloadEnvelope for this block root (via seenCache or fork choice FULL variant) */
+      seenPayloadEnvelope(blockRoot: RootHex): boolean;
       /** Populate in-memory caches with persisted data. Call at least once on startup */
       loadFromDisk(): Promise<void>;
       /** Persist in-memory data to the DB. Call at least once before stopping the process */
@@ Expand Down @@

packages/beacon-node/src/chain/seenCache/seenGossipBlockInput.ts

            
                      Original file line number
                      Diff line number
                      Diff line change
                  
    @@ -149,8 +149,8 @@ export class SeenBlockInput {
  
        });

      }

      has(rootHex: RootHex): boolean {

        return this.blockInputs.has(rootHex);

      hasBlock(rootHex: RootHex): boolean {

        return this.blockInputs.get(rootHex)?.hasBlock() ?? false;

      }

      get(rootHex: RootHex): IBlockInput | undefined {

packages/beacon-node/src/chain/seenCache/seenPayloadEnvelopeInput.ts

            
                      Original file line number
                      Diff line number
                      Diff line change
                  
    @@ -84,8 +84,8 @@ export class SeenPayloadEnvelopeInput {
  
        return this.payloadInputs.get(blockRootHex);

      }

      has(blockRootHex: RootHex): boolean {

        return this.payloadInputs.has(blockRootHex);

      hasPayload(blockRootHex: RootHex): boolean {

        return this.payloadInputs.get(blockRootHex)?.hasPayloadEnvelope() ?? false;

      }

      prune(blockRootHex: RootHex): void {

packages/beacon-node/src/chain/validation/attestation.ts

            
                      Original file line number
                      Diff line number
                      Diff line change
                  
    @@ -44,7 +44,7 @@ import {
  
      getAttDataFromSignedAggregateAndProofElectra,

      getAttDataFromSignedAggregateAndProofPhase0,

      getAttesterIndexFromSingleAttestationSerialized,

      getCommitteeIndexFromSingleAttestationSerialized,

      getIndexFromSingleAttestationSerialized,

      getSignatureFromAttestationSerialized,

      getSignatureFromSingleAttestationSerialized,

    } from "../../util/sszBytes.js";

    @@ -882,12 +882,12 @@ export function getCommitteeIndexFromAttestationOrBytes(
  
      if (isForkPostElectra(fork)) {

        if (isGossipAttestation) {

          return getCommitteeIndexFromSingleAttestationSerialized(ForkName.electra, attestationOrBytes.serializedData);

          return getIndexFromSingleAttestationSerialized(ForkName.electra, attestationOrBytes.serializedData);

        }

        return (attestationOrBytes.attestation as SingleAttestation<ForkPostElectra>).committeeIndex;

      }

      if (isGossipAttestation) {

        return getCommitteeIndexFromSingleAttestationSerialized(ForkName.phase0, attestationOrBytes.serializedData);

        return getIndexFromSingleAttestationSerialized(ForkName.phase0, attestationOrBytes.serializedData);

      }

      return (attestationOrBytes.attestation as SingleAttestation<ForkPreElectra>).data.index;

    }

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

-Original file line number
+Diff line change
@@ Expand Up / @@ -1746,6 +1746,40 @@ export function createLodestarMetrics( @@
           }),
         },
+        // some gossip messages need to wait for payload to be processed before they can be processed
+        awaitingPayloadGossipMessages: {
+          queue: register.gauge<{topic: GossipType}>({
+            name: "lodestar_awaiting_payload_gossip_messages_total",
+            help: "Total number of gossip messages waiting for payload to be processed",
+            labelNames: ["topic"],
+          }),
+          countPerSlot: register.gauge({
+            name: "lodestar_awaiting_payload_gossip_messages_per_slot_total",
+            help: "Total number of gossip messages waiting for payload to be processed per slot",
+          }),
+          resolve: register.gauge<{topic: GossipType}>({
+            name: "lodestar_awaiting_payload_gossip_messages_resolve_total",
+            help: "Total number of gossip messages are reprocessed",
+            labelNames: ["topic"],
+          }),
+          waitSecBeforeResolve: register.gauge<{topic: GossipType}>({
+            name: "lodestar_awaiting_payload_gossip_messages_wait_time_resolve_seconds",
+            help: "Time to wait for unknown payload in seconds",
+            labelNames: ["topic"],
+          }),
+          // 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_payload_gossip_messages_reject_total",
+            help: "Total number of gossip messages are rejected to reprocess",
+            labelNames: ["reason", "topic"],
+          }),
+          waitSecBeforeReject: register.gauge<{reason: ReprocessRejectReason; topic: GossipType}>({
+            name: "lodestar_awaiting_payload_gossip_messages_wait_time_reject_seconds",
+            help: "Time to wait for unknown payload before being rejected",
+            labelNames: ["reason", "topic"],
+          }),
+        },
         lightclientServer: {
           onSyncAggregate: register.gauge<{event: string}>({
             name: "lodestar_lightclient_server_on_sync_aggregate_event_total",
@@ Expand Down @@

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

-Original file line number
+Diff line change
@@ Expand Up / @@ -74,6 +74,7 @@ export interface INetwork extends INetworkCorePublic { @@
       shouldAggregate(subnet: SubnetID, slot: Slot): boolean;
       reStatusPeers(peers: PeerIdStr[]): Promise<void>;
       searchUnknownBlock(slotRoot: SlotRootHex, source: BlockInputSource, peer?: PeerIdStr): void;
+      searchUnknownEnvelope(slotRoot: SlotRootHex, source: BlockInputSource, peer?: PeerIdStr): void;
       // ReqResp
       sendBeaconBlocksByRange(peerId: PeerIdStr, request: phase0.BeaconBlocksByRangeRequest): Promise<SignedBeaconBlock[]>;
       sendBeaconBlocksByRoot(peerId: PeerIdStr, request: BeaconBlocksByRootRequest): Promise<SignedBeaconBlock[]>;
@@ Expand Down @@

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

-Original file line number
+Diff line change
@@ Expand Up / @@ -285,6 +285,10 @@ export class Network implements INetwork { @@
         this.networkProcessor.searchUnknownBlock(slotRoot, source, peer);
       }
+      searchUnknownEnvelope(slotRoot: SlotRootHex, source: BlockInputSource, peer?: PeerIdStr): void {
+        this.networkProcessor.searchUnknownEnvelope(slotRoot, source, peer);
+      }
       async reportPeer(peer: PeerIdStr, action: PeerAction, actionName: string): Promise<void> {
         return this.core.reportPeer(peer, action, actionName);
       }
@@ Expand Down @@

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

-Original file line number
+Diff line change
@@ -1,23 +1,25 @@
-    import {ForkName, isForkPostGloas} from "@lodestar/params";
+    import {ForkName, ForkSeq} from "@lodestar/params";
     import {SlotOptionalRoot, SlotRootHex} from "@lodestar/types";
     import {
       getBeaconBlockRootFromDataColumnSidecarSerialized,
-      getBeaconBlockRootFromExecutionPayloadEnvelopeSerialized,
       getBlockRootFromBeaconAttestationSerialized,
+      getBlockRootFromPayloadAttestationMessageSerialized,
       getBlockRootFromSignedAggregateAndProofSerialized,
       getSlotFromBeaconAttestationSerialized,
       getSlotFromBlobSidecarSerialized,
       getSlotFromDataColumnSidecarSerialized,
       getSlotFromExecutionPayloadEnvelopeSerialized,
+      getSlotFromPayloadAttestationMessageSerialized,
       getSlotFromSignedAggregateAndProofSerialized,
       getSlotFromSignedBeaconBlockSerialized,
+      getSlotFromSignedExecutionPayloadBidSerialized,
     } from "../../util/sszBytes.js";
     import {GossipType} from "../gossip/index.js";
     import {ExtractSlotRootFns} from "./types.js";
     /**
      * Extract the slot and block root of a gossip message form serialized data.
-     * Not applicable for all topics.
+     * Only do it for messages that have a slot and block root, and we want to await the block if the block root is not known.
      */
     export function createExtractBlockSlotRootFns(): ExtractSlotRootFns {
       return {
@@ Expand Down Expand Up @@
         },
         [GossipType.data_column_sidecar]: (data: Uint8Array, fork: ForkName): SlotOptionalRoot | null => {
           const slot = getSlotFromDataColumnSidecarSerialized(data, fork);
           if (slot === null) {
             return null;
           }
-          const root = isForkPostGloas(fork) ? getBeaconBlockRootFromDataColumnSidecarSerialized(data) : null;
+          if (ForkSeq[fork] < ForkSeq.gloas) {
+            return {slot};
+          }
+          const root = getBeaconBlockRootFromDataColumnSidecarSerialized(data);
+          // null root means the message is invalid here and will be ignored in gossip handler later
+          // returning the slot here helps check the earliest permissable slot in the network processor
           return root !== null ? {slot, root} : {slot};
         },
-        [GossipType.execution_payload]: (data: Uint8Array): SlotRootHex | null => {
+        [GossipType.execution_payload]: (data: Uint8Array): SlotOptionalRoot | null => {
           const slot = getSlotFromExecutionPayloadEnvelopeSerialized(data);
-          const root = getBeaconBlockRootFromExecutionPayloadEnvelopeSerialized(data);
+          // Do not extract the root here; the network processor will extract it in the 2nd round to trigger block search without awaiting.
+          if (slot === null) {
+            return null;
+          }
+          return {slot};
+        },
+        [GossipType.payload_attestation_message]: (data: Uint8Array): SlotRootHex | null => {
+          const slot = getSlotFromPayloadAttestationMessageSerialized(data);
+          const root = getBlockRootFromPayloadAttestationMessageSerialized(data);
           if (slot === null || root === null) {
             return null;
           }
           return {slot, root};
         },
+        [GossipType.execution_payload_bid]: (data: Uint8Array): SlotOptionalRoot | null => {
+          const slot = getSlotFromSignedExecutionPayloadBidSerialized(data);
+          if (slot === null) {
+            return null;
+          }
+          return {slot};
+        },
       };
     }

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

-Original file line number
+Diff line change
@@ Expand Up @@
         logger.debug("Received gossip block", {...logCtx});
-        let blockInput: IBlockInput | undefined;
+        // optimistically add gossip block to the seen cache
+        // if validation fails, we will NOT forward this gossip block to peers
+        //   - if PARENT_UNKNOWN error, blockInput will then be queued inside BlockInputSync. If the gossip block is really invalid, it will be pruned there
+        //   - if other validator errors, blockInput will stay in the seen cache and will be pruned on finalization
+        const blockInput = chain.seenBlockInputCache.getByBlock({
+          block: signedBlock,
+          blockRootHex,
+          source: BlockInputSource.gossip,
+          seenTimestampSec,
+          peerIdStr,
+        });
         try {
           await validateGossipBlock(config, chain, signedBlock, fork);
-          blockInput = chain.seenBlockInputCache.getByBlock({
-            block: signedBlock,
-            blockRootHex,
-            source: BlockInputSource.gossip,
-            seenTimestampSec,
-            peerIdStr,
-          });
           const blockInputMeta = blockInput.getLogMeta();
           const recvToValidation = Date.now() / 1000 - seenTimestampSec;
@@ Expand All @@
           return blockInput;
         } catch (e) {
           if (e instanceof BlockGossipError) {
+            logger.debug("Gossip block has error", {slot, root: blockShortHex, code: e.type.code});
             if (e.type.code === BlockErrorCode.PARENT_UNKNOWN && blockInput) {
-              logger.debug("Gossip block has error", {slot, root: blockShortHex, code: e.type.code});
-              chain.emitter.emit(ChainEvent.unknownParent, {
+              chain.emitter.emit(ChainEvent.blockUnknownParent, {
                 blockInput,
                 peer: peerIdStr,
                 source: BlockInputSource.gossip,
@@ Expand Down Expand Up @@
           const {serializedData} = gossipData;
           const signedEnvelope = sszDeserialize(topic, serializedData);
           const envelope = signedEnvelope.message;
-          // TODO GLOAS: handle BLOCK_ROOT_UNKNOWN error to trigger sync
-          await validateGossipExecutionPayloadEnvelope(chain, signedEnvelope);
+          // TODO GLOAS: consider optimistically create PayloadEnvelopeInput here similar to how we do that for beacon_block
+          // so that UnknownBlockSync can handle backward sync
+          // the problem now is we cannot create a PayloadEnvelopeInput without the beacon block being known, we need at least the proposer index
+          // we can achieve that by looking into the EpochCache
+          try {
+            await validateGossipExecutionPayloadEnvelope(chain, signedEnvelope);
+          } catch (e) {
+            if (e instanceof ExecutionPayloadEnvelopeError) {
+              const {slot, beaconBlockRoot} = signedEnvelope.message;
+              logger.debug("Gossip envelope has error", {slot, root: toRootHex(beaconBlockRoot), code: e.type.code});
+              if (e.type.code === ExecutionPayloadEnvelopeErrorCode.BLOCK_ROOT_UNKNOWN) {
+                // TODO GLOAS: UnknownBlockSync to handle this
+                chain.emitter.emit(ChainEvent.envelopeUnknownBlock, {
+                  envelope: signedEnvelope,
+                  peer: peerIdStr,
+                  source: BlockInputSource.gossip,
+                });
+              }
+              if (e.action === GossipAction.REJECT) {
+                chain.persistInvalidSszValue(
+                  ssz.gloas.SignedExecutionPayloadEnvelope,
+                  signedEnvelope,
+                  `gossip_reject_slot_${slot}`
+                );
+              }
+            }
+            throw e;
+          }
           const slot = envelope.slot;
           const delaySec = seenTimestampSec - computeTimeAtSlot(config, slot, chain.genesisTime);
@@ Expand Down @@

Navigation Menu

Uh oh!

Uh oh!

feat: enhance NetworkProcessor for gloas #9025

Uh oh!

Diff view

Diff view

There are no files selected for viewing

Uh oh!

Uh oh!

twoeths Mar 26, 2026

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

feat: enhance NetworkProcessor for gloas #9025

Uh oh!

feat: enhance NetworkProcessor for gloas #9025

Uh oh!

Uh oh!

Diff view

Diff view

There are no files selected for viewing

Uh oh!

Uh oh!

twoeths Mar 26, 2026

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!