feat(builder): observe beacon node blocks

packages/builder/src/builder.ts

@@ -7,12 +7,14 @@ import {waitForGenesis} from "./genesis.js";
 import {resolveBuilderIdentity} from "./identity.js";
 import {Metrics} from "./metrics.js";
 import {logNodeVersion, waitForNodeReady} from "./readiness.js";
+import {BlockObserver} from "./services/blockObserver.js";
 import {BuilderSigner, Keypair} from "./services/builderSigner.js";
 import {BuilderStatusTracker} from "./services/builderStatusTracker.js";
 
 export type BuilderModules = {
   opts: BuilderOptions;
   builderSigner: BuilderSigner;
+  blockObserver: BlockObserver;
   builderStatusTracker: BuilderStatusTracker;
   clock: IClock;
   index: BuilderIndex;
@@ -34,15 +36,17 @@ export type BuilderOptions = {
  */
 export class Builder {
   readonly builderSigner: BuilderSigner;
+  private readonly blockObserver: BlockObserver;
   private readonly builderStatusTracker: BuilderStatusTracker;
   private readonly controller: AbortController;
   private readonly clock: IClock;
   private readonly index: BuilderIndex;
   private readonly logger: Logger;
   private readonly executionFeeRecipient: ExecutionAddress;
 
-  constructor({opts, builderSigner, builderStatusTracker, clock, index}: BuilderModules) {
+  constructor({opts, builderSigner, blockObserver, builderStatusTracker, clock, index}: BuilderModules) {
     this.builderSigner = builderSigner;
+    this.blockObserver = blockObserver;
     this.builderStatusTracker = builderStatusTracker;
     this.clock = clock;
     this.controller = opts.abortController;
@@ -53,6 +57,7 @@ export class Builder {
 
     this.clock.runEveryEpoch((epoch) => this.builderStatusTracker.poll(epoch));
     this.clock.start(this.controller.signal);
+    this.blockObserver.start(this.controller.signal);
 
     this.logger.info("Builder client initialized", {
       index: this.index,
@@ -89,8 +94,9 @@ export class Builder {
     );
 
     const builderStatusTracker = new BuilderStatusTracker(api, logger, index, opts.metrics);
+    const blockObserver = new BlockObserver(config, logger, api);
 
-    return new Builder({opts, builderSigner, builderStatusTracker, clock, index});
+    return new Builder({opts, builderSigner, blockObserver, builderStatusTracker, clock, index});
   }
 
   async close(): Promise<void> {

packages/builder/src/services/blockObserver.ts

@@ -0,0 +1,226 @@
+import {ApiClient, ApiError, routes} from "@lodestar/api";
+import {ChainForkConfig} from "@lodestar/config";
+import {ForkPostGloas, isForkPostGloas} from "@lodestar/params";
+import {RootHex, SignedBeaconBlock, Slot, isGloasBeaconBlock} from "@lodestar/types";
+import {Logger, TimeoutError, isErrorAborted, isFetchError, pruneSetToMax, retry, toRootHex} from "@lodestar/utils";
+
+const {EventType} = routes.events;
+
+type BlockEvent = routes.events.EventData[typeof EventType.block];
+
+type BlockObserverOptions = {
+  retries?: number;
+  retryDelay?: number;
+  maxSeenBlockRoots?: number;
+};
+
+export type ObservedBlock = {
+  blockRoot: RootHex;
+  slot: Slot;
+  executionOptimistic: boolean;
+  version: ForkPostGloas;
+  block: SignedBeaconBlock<ForkPostGloas>;
+  signedBid: SignedBeaconBlock<ForkPostGloas>["message"]["body"]["signedExecutionPayloadBid"];
+};
+
+type RunOnBlockFn = (block: ObservedBlock) => Promise<void>;
+
+/**
+ * Observes imported beacon blocks through the source beacon node API.
+ *
+ * Each block root is evaluated at most once while retained in the bounded seen set. Terminal failures remain consumed
+ * until normal FIFO eviction; reconnect, replay, and recovery policies are handled separately.
+ */
+export class BlockObserver {
+  private readonly fns: RunOnBlockFn[] = [];
+  private readonly seenBlockRoots = new Set<RootHex>();
+  private readonly retries: number;
+  private readonly retryDelay: number;
+  private readonly maxSeenBlockRoots: number;
+
+  constructor(
+    private readonly config: ChainForkConfig,
+    private readonly logger: Logger,
+    private readonly api: ApiClient,
+    {retries = 5, retryDelay = 200, maxSeenBlockRoots = 256}: BlockObserverOptions = {}
+  ) {
+    this.retries = retries;
+    this.retryDelay = retryDelay;
+    this.maxSeenBlockRoots = maxSeenBlockRoots;
+  }
+
+  /**
+   * Register a block consumer. Consumers that require complete observation must register before {@link start}, because
+   * roots observed before registration remain consumed until normal FIFO eviction.
+   */
+  runOnBlock(fn: RunOnBlockFn): void {
+    this.fns.push(fn);
+  }
+
+  start(signal: AbortSignal): void {
+    this.logger.info("Subscribing to block events");
+
+    this.api.events
+      .eventstream({
+        topics: [EventType.block],
+        signal,
+        onEvent: (event) => {
+          if (event.type !== EventType.block) {
+            this.logger.debug("Ignoring unexpected beacon event", {eventType: event.type});
+            return;
+          }
+
+          void this.processBlockEvent(event.message, signal);
+        },
+        onError: (error) => {
+          this.logger.error("Failed to receive block event", {}, error);
+        },
+        onClose: () => {
+          if (signal.aborted) {
+            this.logger.debug("Closed stream for block events");
+          } else {
+            this.logger.error("Block event stream closed unexpectedly", {});
+          }
+        },
+      })
+      .catch((error: unknown) => {
+        this.logger.error(
+          "Failed to subscribe to block events",
+          {},
+          error instanceof Error ? error : Error(String(error))
+        );
+      });
+  }
+
+  async processBlockEvent(event: BlockEvent, signal: AbortSignal): Promise<void> {
+    try {
+      const {slot, block: blockRoot, executionOptimistic} = event;
+
+      if (this.seenBlockRoots.has(blockRoot)) {
+        this.logger.debug("Ignoring duplicate block event", {slot, blockRoot});
+        return;
+      }
+
+      this.seenBlockRoots.add(blockRoot);
+      pruneSetToMax(this.seenBlockRoots, this.maxSeenBlockRoots);
+
+      if (!isForkPostGloas(this.config.getForkName(slot))) {
+        this.logger.debug("Ignoring pre-Gloas block event", {slot, blockRoot});
+        return;
+      }
+
+      const response = await retry(
+        async () => {
+          const result = await this.api.beacon.getBlockV2({blockId: blockRoot}, {signal});
+          result.assertOk();
+          return result;
+        },
+        {
+          retries: this.retries,
+          retryDelay: this.retryDelay,
+          signal,
+          shouldRetry: isRetryableBlockRetrievalError,
+          onRetry: (error, attempt) => {
+            this.logger.debug("Retrying block retrieval", {
+              slot,
+              blockRoot,
+              attempt,
+              error: error.message,
+            });
+          },
+        }
+      ).catch((error: unknown) => {
+        if (!isErrorAborted(error)) {
+          this.logger.error(
+            "Failed to retrieve block referenced by block event",
+            {slot, blockRoot},
+            error instanceof Error ? error : Error(String(error))
+          );
+        }
+        return null;
+      });
+
+      if (response === null) {
+        return;
+      }
+
+      const {version} = response.meta();
+      if (!isForkPostGloas(version)) {
+        this.logger.warn("Ignoring block with unsupported fork version", {slot, blockRoot, fork: version});
+        return;
+      }
+
+      const block = response.value();
+      if (!isGloasBeaconBlock(block.message)) {
+        this.logger.error("Block response version and body do not agree", {slot, blockRoot, fork: version});
+        return;
+      }
+
+      if (block.message.slot !== slot) {
+        this.logger.error("Block response slot does not match block event", {
+          slot,
+          blockRoot,
+          blockSlot: block.message.slot,
+        });
+        return;
+      }
+
+      const postGloasBlock = block as SignedBeaconBlock<ForkPostGloas>;
+      const signedBid = postGloasBlock.message.body.signedExecutionPayloadBid;
+      const observedBlock: ObservedBlock = {
+        blockRoot,
+        slot,
+        executionOptimistic,
+        version,
+        block: postGloasBlock,
+        signedBid,
+      };
+
+      this.logger.debug("Observed post-Gloas block", {
+        slot,
+        blockRoot,
+        fork: version,
+        builderIndex: signedBid.message.builderIndex,
+        value: signedBid.message.value,
+        blockHash: toRootHex(signedBid.message.blockHash),
+        parentBlockHash: toRootHex(signedBid.message.parentBlockHash),
+      });
+
+      await Promise.all(
+        this.fns.map(async (fn) => {
+          try {
+            await fn(observedBlock);
+          } catch (error) {
+            if (isErrorAborted(error)) {
+              return;
+            }
+
+            this.logger.error(
+              "Failed to process observed block",
+              {slot, blockRoot},
+              error instanceof Error ? error : Error(String(error))
+            );
+          }
+        })
+      );
+    } catch (error) {
+      if (isErrorAborted(error)) {
+        return;
+      }
+
+      this.logger.error(
+        "Failed to process block event",
+        {slot: event.slot, blockRoot: event.block},
+        error instanceof Error ? error : Error(String(error))
+      );
+    }
+  }
+}
+
+export function isRetryableBlockRetrievalError(error: Error): boolean {
+  if (error instanceof ApiError) {
+    return error.status === 404 || error.status >= 500;
+  }
+
+  return error instanceof TimeoutError || (isFetchError(error) && error.type !== "input");
+}

packages/builder/test/unit/builder.test.ts

@@ -0,0 +1,50 @@
+import {describe, expect, it, vi} from "vitest";
+import {SecretKey} from "@chainsafe/lodestar-z/blst";
+import {createBeaconConfig} from "@lodestar/config";
+import {getConfig} from "@lodestar/config/test-utils";
+import {ForkName} from "@lodestar/params";
+import {Builder, BuilderOptions} from "../../src/builder.js";
+import {BlockObserver} from "../../src/services/blockObserver.js";
+import {BuilderSigner} from "../../src/services/builderSigner.js";
+import {BuilderStatusTracker} from "../../src/services/builderStatusTracker.js";
+import {getApiClientStub} from "./utils/apiStub.js";
+import {ClockMock} from "./utils/clock.js";
+import {getMockedLogger} from "./utils/logger.js";
+
+describe("Builder", () => {
+  it("starts long-lived duties with the shared signal and aborts them on close", async () => {
+    const config = getConfig(ForkName.gloas);
+    const logger = getMockedLogger();
+    const api = getApiClientStub();
+    const controller = new AbortController();
+    const clock = new ClockMock();
+    const secretKey = SecretKey.fromBytes(Buffer.alloc(32, 1));
+    const keypair = {secretKey, publicKey: secretKey.toPublicKey()};
+    const builderSigner = new BuilderSigner(createBeaconConfig(config, Buffer.alloc(32)), keypair);
+    const builderStatusTracker = new BuilderStatusTracker(api, logger, 1, null);
+    const blockObserver = new BlockObserver(config, logger, api);
+    const clockStart = vi.spyOn(clock, "start");
+    const observerStart = vi.spyOn(blockObserver, "start").mockImplementation(() => {});
+
+    const opts: BuilderOptions = {
+      logger,
+      config,
+      keypair,
+      abortController: controller,
+      api,
+      executionFeeRecipient: Buffer.alloc(20),
+      metrics: null,
+    };
+
+    const builder = new Builder({opts, builderSigner, blockObserver, builderStatusTracker, clock, index: 1});
+
+    expect(clockStart).toHaveBeenCalledWith(controller.signal);
+    expect(observerStart).toHaveBeenCalledWith(controller.signal);
+    expect(clockStart.mock.invocationCallOrder[0]).toBeLessThan(observerStart.mock.invocationCallOrder[0]);
+    expect(controller.signal.aborted).toBe(false);
+
+    await builder.close();
+
+    expect(controller.signal.aborted).toBe(true);
+  });
+});