feat: implement head_v2 event

packages/api/src/beacon/routes/events.ts

@@ -56,9 +56,23 @@ const gloasDataColumnSidecarSSE = new ContainerType(
   },
   {typeName: "DataColumnSidecarSSE", jsonCase: "eth2"}
 );
+const headV2 = new ContainerType(
+  {
+    slot: ssz.Slot,
+    block: stringType,
+    state: stringType,
+    payloadStatus: stringType,
+    epochTransition: ssz.Boolean,
+    currentEpochDependentRoot: stringType,
+    nextEpochDependentRoot: stringType,
+    executionOptimistic: ssz.Boolean,
+  },
+  {typeName: "HeadV2", jsonCase: "eth2"}
+);
 type FuluDataColumnSidecarSSE = ValueOf<typeof fuluDataColumnSidecarSSE>;
 type GloasDataColumnSidecarSSE = ValueOf<typeof gloasDataColumnSidecarSSE>;
 type DataColumnSidecarSSE = FuluDataColumnSidecarSSE | GloasDataColumnSidecarSSE;
+type HeadV2 = ValueOf<typeof headV2>;
 
 export enum EventType {
   /**
@@ -68,6 +82,7 @@ export enum EventType {
    * Both dependent roots use the genesis block root in the case of underflow.
    */
   head = "head",
+  headV2 = "head_v2",
   /** The node has received a block (from P2P or API) that is successfully imported on the fork-choice `on_block` handler */
   block = "block",
   /** The node has received a block (from P2P or API) that passes validation rules of the `beacon_block` topic */
@@ -116,6 +131,7 @@ export enum EventType {
 
 export const eventTypes: {[K in EventType]: K} = {
   [EventType.head]: EventType.head,
+  [EventType.headV2]: EventType.headV2,
   [EventType.block]: EventType.block,
   [EventType.blockGossip]: EventType.blockGossip,
   [EventType.attestation]: EventType.attestation,
@@ -150,6 +166,10 @@ export type EventData = {
     currentDutyDependentRoot: RootHex;
     executionOptimistic: boolean;
   };
+  [EventType.headV2]: {
+    version: ForkName;
+    data: HeadV2;
+  };
   [EventType.block]: {
     slot: Slot;
     block: RootHex;
@@ -295,7 +315,7 @@ export function getTypeByEvent(config: ChainForkConfig): {[K in EventType]: Type
       },
       {jsonCase: "eth2"}
     ),
-
+    [EventType.headV2]: WithVersion(() => headV2),
     [EventType.block]: new ContainerType(
       {
         slot: ssz.Slot,

packages/api/test/unit/beacon/testData/events.ts

@@ -30,6 +30,19 @@ export const eventTestData: EventData = {
     currentDutyDependentRoot: "0x5e0043f107cb57913498fbf2f99ff55e730bf1e151f02f221e977c91a90a0e91",
     executionOptimistic: false,
   },
+  [EventType.headV2]: {
+    version: ForkName.bellatrix,
+    data: {
+      slot: 10,
+      block: "0x9a2fefd2fdb57f74993c7780ea5b9030d2897b615b89f808011ca5aebed54eaf",
+      state: "0x600e852a08c1200654ddf11025f1ceacb3c2e74bdd5c630cde0838b2591b69f9",
+      payloadStatus: "full",
+      epochTransition: false,
+      currentEpochDependentRoot: "0x5e0043f107cb57913498fbf2f99ff55e730bf1e151f02f221e977c91a90a0e91",
+      nextEpochDependentRoot: "0x5e0043f107cb57913498fbf2f99ff55e730bf1e151f02f221e977c91a90a0e91",
+      executionOptimistic: false,
+    },
+  },
   [EventType.block]: {
     slot: 10,
     block: "0x9a2fefd2fdb57f74993c7780ea5b9030d2897b615b89f808011ca5aebed54eaf",

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

@@ -0,0 +1,39 @@
+import {routes} from "@lodestar/api";
+import {EpochDifference, PayloadStatus, ProtoBlock} from "@lodestar/fork-choice";
+import {computeEpochAtSlot, computeStartSlotAtEpoch} from "@lodestar/state-transition";
+import {isOptimisticBlock} from "../../util/forkChoice.js";
+import {BeaconChain} from "../chain.js";
+
+export function emitHeadV2(this: BeaconChain, head: ProtoBlock, headChanged: boolean) {
+  if (
+    headChanged ||
+    !this.headV2PayloadStatusCache.has(head.blockRoot) ||
+    (this.headV2PayloadStatusCache.get(head.blockRoot)?.status !== PayloadStatus.FULL &&
+      head.payloadStatus === PayloadStatus.FULL)
+  ) {
+    this.emitter.emit(routes.events.EventType.headV2, {
+      version: this.config.getForkName(head.slot),
+      data: {
+        slot: head.slot,
+        block: head.blockRoot,
+        state: head.stateRoot,
+        payloadStatus: toApiPayloadStatus(head.payloadStatus),
+        epochTransition: computeStartSlotAtEpoch(computeEpochAtSlot(head.slot)) === head.slot,
+        currentEpochDependentRoot: this.forkChoice.getDependentRoot(head, EpochDifference.previous),
+        nextEpochDependentRoot: this.forkChoice.getDependentRoot(head, EpochDifference.current),
+        executionOptimistic: isOptimisticBlock(head),
+      },
+    });
+    this.headV2PayloadStatusCache.set(head.blockRoot, {status: head.payloadStatus, slot: head.slot});
+    this.metrics?.headV2PayloadStatusCacheSize.set(this.headV2PayloadStatusCache.size);
+  }
+}
+
+function toApiPayloadStatus(status?: PayloadStatus): "empty" | "full" {
+  switch (status) {
+    case PayloadStatus.FULL:
+      return "full";
+    default:
+      return "empty";
+  }
+}

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

@@ -39,6 +39,7 @@ import {ForkchoiceCaller} from "../forkChoice/index.js";
 import {REPROCESS_MIN_TIME_TO_NEXT_SLOT_SEC} from "../reprocess.js";
 import {toCheckpointHex} from "../stateCache/persistentCheckpointsCache.js";
 import {isBlockInputBlobs, isBlockInputColumns} from "./blockInput/blockInput.js";
+import {emitHeadV2} from "./headV2Event.js";
 import {AttestationImportOpt, FullyVerifiedBlock, ImportBlockOpts} from "./types.js";
 import {getCheckpointFromState} from "./utils/checkpoint.js";
 
@@ -316,9 +317,10 @@ export async function importBlock(
         currentDutyDependentRoot: this.forkChoice.getDependentRoot(newHead, EpochDifference.current),
         executionOptimistic: isOptimisticBlock(newHead),
       });
+      emitHeadV2.call(this, newHead, true);
     } catch (e) {
       // getDependentRoot() may fail with error: "No block for root" as we can see in holesky non-finality issue
-      this.logger.debug("Error emitting head event", {slot: newHead.slot, root: newHead.blockRoot}, e as Error);
+      this.logger.debug("Error emitting head/head_v2 event", {slot: newHead.slot, root: newHead.blockRoot}, e as Error);
     }
 
     const delaySec = this.clock.secFromSlot(newHead.slot);

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

@@ -8,6 +8,7 @@ import {isQueueErrorAborted} from "../../util/queue/index.js";
 import {BeaconChain} from "../chain.js";
 import {RegenCaller} from "../regen/interface.js";
 import {PayloadEnvelopeInput} from "../seenCache/seenPayloadEnvelopeInput.js";
+import {emitHeadV2} from "./headV2Event.js";
 import {ImportPayloadOpts} from "./types.js";
 import {
   verifyExecutionPayloadEnvelope,
@@ -254,6 +255,12 @@ export async function importExecutionPayload(
     });
   }
 
+  try {
+    emitHeadV2.call(this, head, false);
+  } catch (e) {
+    this.logger.debug("Error emitting head_v2 event", {slot: head.slot, root: head.blockRoot}, e as Error);
+  }
+
   // 8. Record metrics for payload envelope and column sources
   const delaySec = this.clock.secFromSlot(slot);
   this.metrics?.importPayload.elapsedTimeTillImported.observe(

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

@@ -2,7 +2,14 @@ import path from "node:path";
 import {PrivateKey} from "@libp2p/interface";
 import {Type} from "@chainsafe/ssz";
 import {BeaconConfig} from "@lodestar/config";
-import {CheckpointWithHex, ForkChoiceStateGetter, IForkChoice, ProtoBlock, UpdateHeadOpt} from "@lodestar/fork-choice";
+import {
+  CheckpointWithHex,
+  ForkChoiceStateGetter,
+  IForkChoice,
+  PayloadStatus,
+  ProtoBlock,
+  UpdateHeadOpt,
+} from "@lodestar/fork-choice";
 import {LoggerNode} from "@lodestar/logger/node";
 import {
   EFFECTIVE_BALANCE_INCREMENT,
@@ -222,6 +229,12 @@ export class BeaconChain implements IBeaconChain {
 
   readonly opts: IChainOptions;
 
+  /**
+   * Cache of head block payload status at time of head_v2 emission, used to detect
+   * empty -> full transitions and emit a second head_v2 event accordingly.
+   */
+  readonly headV2PayloadStatusCache = new Map<RootHex, {status: PayloadStatus; slot: Slot}>();
+
   protected readonly blockProcessor: BlockProcessor;
   protected readonly payloadEnvelopeProcessor: PayloadEnvelopeProcessor;
   protected readonly db: IBeaconDb;
@@ -1482,6 +1495,14 @@ export class BeaconChain implements IBeaconChain {
     pruneSetToMax(this.blockProductionCache, this.opts.maxCachedProducedRoots ?? DEFAULT_MAX_CACHED_PRODUCED_RESULTS);
     this.metrics?.blockProductionCacheSize.set(this.blockProductionCache.size);
 
+    // Prune outdated head v2 payload statuses
+    for (const [blockRoot, value] of this.headV2PayloadStatusCache.entries()) {
+      if (value.slot + SLOTS_PER_EPOCH < slot) {
+        this.headV2PayloadStatusCache.delete(blockRoot);
+      }
+    }
+    this.metrics?.headV2PayloadStatusCacheSize.set(this.headV2PayloadStatusCache.size);
+
     const metrics = this.metrics;
     if (metrics && (slot + 1) % SLOTS_PER_EPOCH === 0) {
       // On the last slot of the epoch

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

@@ -137,6 +137,10 @@ export function createBeaconMetrics(register: RegistryMetricCreator) {
       name: "beacon_block_production_cache_size",
       help: "Count of cached produced results",
     }),
+    headV2PayloadStatusCacheSize: register.gauge({
+      name: "head_v2_payload_status_cache_size",
+      help: "Count of cached head v2 payload statuses",
+    }),
 
     blockPayload: {
       payloadAdvancePrepTime: register.histogram({

packages/beacon-node/test/mocks/mockedBeaconChain.ts

@@ -186,6 +186,7 @@ vi.mock("../../src/chain/chain.js", async (importActual) => {
         canAcceptWork: vi.fn().mockReturnValue(true),
       },
       emitter: new ChainEventEmitter(),
+      headV2PayloadStatusCache: new Map(),
     };
   });
 

packages/beacon-node/test/unit/chain/blocks/headV2Event.test.ts

@@ -0,0 +1,101 @@
+import {beforeEach, describe, expect, it} from "vitest";
+import {routes} from "@lodestar/api";
+import {ExecutionStatus, PayloadStatus, ProtoBlock} from "@lodestar/fork-choice";
+import {DataAvailabilityStatus} from "@lodestar/state-transition";
+import {emitHeadV2} from "../../../../src/chain/blocks/headV2Event.js";
+import {MockedBeaconChain, getMockedBeaconChain} from "../../../mocks/mockedBeaconChain.js";
+
+const {EventType} = routes.events;
+
+const blockRoot1 = "0xaabb";
+const blockRoot2 = "0xbbcc";
+
+function protoBlock(blockRoot: string, slot: number, payloadStatus: PayloadStatus): ProtoBlock {
+  return {
+    slot,
+    blockRoot,
+    parentRoot: blockRoot,
+    stateRoot: blockRoot,
+    targetRoot: blockRoot,
+    justifiedEpoch: 0,
+    justifiedRoot: blockRoot,
+    finalizedEpoch: 0,
+    finalizedRoot: blockRoot,
+    unrealizedJustifiedEpoch: 0,
+    unrealizedJustifiedRoot: blockRoot,
+    unrealizedFinalizedEpoch: 0,
+    unrealizedFinalizedRoot: blockRoot,
+    timeliness: false,
+    executionPayloadBlockHash: null,
+    executionStatus: ExecutionStatus.PreMerge,
+    dataAvailabilityStatus: DataAvailabilityStatus.PreData,
+    payloadStatus,
+    parentBlockHash: null,
+  };
+}
+
+describe("head_v2 event emission", () => {
+  let chain: MockedBeaconChain;
+  const slot = 10;
+
+  beforeEach(() => {
+    chain = getMockedBeaconChain();
+    chain.forkChoice.getDependentRoot.mockReturnValue("0x1234");
+  });
+
+  it("emits head_v2 with empty on first import", () => {
+    const events: routes.events.EventData[typeof EventType.headV2][] = [];
+    chain.emitter.on(EventType.headV2, (data) => events.push(data));
+
+    emitHeadV2.call(chain, protoBlock(blockRoot1, slot, PayloadStatus.EMPTY), true);
+
+    expect(events).toHaveLength(1);
+    expect(events[0].data.payloadStatus).toBe("empty");
+  });
+
+  it("emits head_v2 with full on empty→full transition", () => {
+    const events: routes.events.EventData[typeof EventType.headV2][] = [];
+    chain.emitter.on(EventType.headV2, (data) => events.push(data));
+
+    emitHeadV2.call(chain, protoBlock(blockRoot1, slot, PayloadStatus.EMPTY), true);
+    emitHeadV2.call(chain, protoBlock(blockRoot1, slot, PayloadStatus.FULL), false);
+
+    expect(events).toHaveLength(2);
+    expect(events[0].data.payloadStatus).toBe("empty");
+    expect(events[1].data.payloadStatus).toBe("full");
+  });
+
+  it("does not re-emit on empty→empty", () => {
+    const events: routes.events.EventData[typeof EventType.headV2][] = [];
+    chain.emitter.on(EventType.headV2, (data) => events.push(data));
+
+    emitHeadV2.call(chain, protoBlock(blockRoot1, slot, PayloadStatus.EMPTY), true);
+    emitHeadV2.call(chain, protoBlock(blockRoot1, slot, PayloadStatus.EMPTY), false);
+
+    expect(events).toHaveLength(1);
+    expect(events[0].data.payloadStatus).toBe("empty");
+  });
+
+  it("does not re-emit on full→full", () => {
+    const events: routes.events.EventData[typeof EventType.headV2][] = [];
+    chain.emitter.on(EventType.headV2, (data) => events.push(data));
+
+    emitHeadV2.call(chain, protoBlock(blockRoot1, slot, PayloadStatus.FULL), true);
+    emitHeadV2.call(chain, protoBlock(blockRoot1, slot, PayloadStatus.FULL), false);
+
+    expect(events).toHaveLength(1);
+    expect(events[0].data.payloadStatus).toBe("full");
+  });
+
+  it("re-emits on head change to already-cached root (reorg)", () => {
+    const events: routes.events.EventData[typeof EventType.headV2][] = [];
+    chain.emitter.on(EventType.headV2, (data) => events.push(data));
+
+    emitHeadV2.call(chain, protoBlock(blockRoot1, slot, PayloadStatus.FULL), true);
+    emitHeadV2.call(chain, protoBlock(blockRoot2, slot, PayloadStatus.FULL), true);
+    emitHeadV2.call(chain, protoBlock(blockRoot1, slot, PayloadStatus.FULL), true);
+
+    expect(events).toHaveLength(3);
+    expect(events[0].data.payloadStatus).toBe("full");
+  });
+});

packages/validator/src/services/chainHeaderTracker.ts

@@ -39,12 +39,14 @@ export class ChainHeaderTracker {
   start(signal: AbortSignal): void {
     this.logger.verbose("Subscribing to validator events");
 
-    const topics = [EventType.head];
     // We wait until the gloas fork is configured to avoid breaking
     // connections with pre-gloas beacon nodes
-    if (this.config.GLOAS_FORK_EPOCH !== Infinity) {
-      topics.push(EventType.executionPayloadAvailable);
-    }
+    // Post-gloas BNs are required to support head_v2 per beacon-APIs#590,
+    // so we assume any gloas-configured BN will accept this topic.
+    const topics =
+      this.config.GLOAS_FORK_EPOCH === Infinity
+        ? [EventType.head]
+        : [EventType.headV2, EventType.executionPayloadAvailable];
 
     this.api.events
       .eventstream({
@@ -74,6 +76,7 @@ export class ChainHeaderTracker {
   }
 
   private onEvent = (event: routes.events.BeaconEvent): void => {
+    // Use head_v2 instead when gloas is configured, as head is deprecated post-gloas
     if (event.type === EventType.head) {
       const {message} = event;
       const {slot, block, previousDutyDependentRoot, currentDutyDependentRoot} = message;
@@ -101,6 +104,33 @@ export class ChainHeaderTracker {
       });
     }
 
+    if (event.type === EventType.headV2) {
+      const {data} = event.message;
+      const {slot, block, currentEpochDependentRoot, nextEpochDependentRoot} = data;
+      this.headBlockSlot = slot;
+      this.headBlockRoot = fromHex(block);
+
+      const headEventData = {
+        slot: this.headBlockSlot,
+        head: block,
+        previousDutyDependentRoot: currentEpochDependentRoot,
+        currentDutyDependentRoot: nextEpochDependentRoot,
+      };
+
+      for (const fn of this.fns) {
+        fn(headEventData).catch((e) => this.logger.error("Error calling head event handler", e));
+      }
+
+      this.emitter.emit(ValidatorEvent.chainHead, headEventData);
+
+      this.logger.verbose("Found new chain head", {
+        slot: slot,
+        head: block,
+        previousDuty: currentEpochDependentRoot,
+        currentDuty: nextEpochDependentRoot,
+      });
+    }
+
     if (event.type === EventType.executionPayloadAvailable) {
       this.emitter.emit(ValidatorEvent.executionPayloadAvailable, event.message);