fix: prune serialized cache by block input keys

dashboards/lodestar_beacon_chain.json

@@ -388,6 +388,18 @@
           "legendFormat": "created_by_blob",
           "range": true,
           "refId": "F"
+        },
+        {
+          "datasource": {
+            "type": "prometheus",
+            "uid": "${DS_PROMETHEUS}"
+          },
+          "editorMode": "code",
+          "expr": "lodestar_seen_block_input_cache_serialized_object_count",
+          "instant": false,
+          "legendFormat": "serialized_objects",
+          "range": true,
+          "refId": "G"
         }
       ],
       "title": "Block Input",

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

@@ -105,6 +105,7 @@ abstract class AbstractBlockInput<F extends ForkName = ForkName, TData extends D
   }
 
   abstract addBlock(props: AddBlock<F>): void;
+  abstract getSerializedCacheKeys(): object[];
 
   hasBlock(): boolean {
     return this.state.hasBlock;
@@ -243,6 +244,10 @@ export class BlockInputPreData extends AbstractBlockInput<ForkPreDeneb, null> {
       );
     }
   }
+
+  getSerializedCacheKeys(): object[] {
+    return [this.state.block];
+  }
 }
 
 // Blobs DA
@@ -526,6 +531,20 @@ export class BlockInputBlobs extends AbstractBlockInput<ForkBlobsDA, deneb.BlobS
   getBlobs(): deneb.BlobSidecars {
     return this.getAllBlobsWithSource().map(({blobSidecar}) => blobSidecar);
   }
+
+  getSerializedCacheKeys(): object[] {
+    const objects: object[] = [];
+
+    if (this.state.hasBlock) {
+      objects.push(this.state.block);
+    }
+
+    for (const {blobSidecar} of this.blobsCache.values()) {
+      objects.push(blobSidecar);
+    }
+
+    return objects;
+  }
 }
 
 function blockAndBlobArePaired(block: SignedBeaconBlock<ForkBlobsDA>, blobSidecar: deneb.BlobSidecar): boolean {
@@ -906,6 +925,18 @@ export class BlockInputColumns extends AbstractBlockInput<ForkColumnsDA, fulu.Da
     }
     return Promise.resolve(this.getSampledColumns());
   }
+
+  getSerializedCacheKeys(): object[] {
+    const objects: object[] = [];
+
+    if (this.state.hasBlock) {
+      objects.push(this.state.block);
+    }
+
+    objects.push(...this.getAllColumns());
+
+    return objects;
+  }
 }
 
 type BlockInputNoDataState = {
@@ -967,4 +998,8 @@ export class BlockInputNoData extends AbstractBlockInput<ForkPostGloas, null> {
     return (this.state.block.message.body as gloas.BeaconBlockBody).signedExecutionPayloadBid.message
       .blobKzgCommitments;
   }
+
+  getSerializedCacheKeys(): object[] {
+    return [this.state.block];
+  }
 }

packages/beacon-node/src/chain/blocks/blockInput/types.ts

@@ -151,6 +151,11 @@ export interface IBlockInput<F extends ForkName = ForkName, TData extends DAData
 
   /** Only safe to call when `hasBlockAndAllData` is true */
   getTimeComplete(): number;
+  /**
+   * Return object references used as keys in `SerializedCache` that can be safely removed
+   * once this block input lifecycle has completed.
+   */
+  getSerializedCacheKeys(): object[];
 
   waitForBlock(timeout: number, signal?: AbortSignal): Promise<SignedBeaconBlock<F>>;
   waitForAllData(timeout: number, signal?: AbortSignal): Promise<TData>;

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

@@ -3,13 +3,7 @@ import {SignedBeaconBlock} from "@lodestar/types";
 import {fromHex, toRootHex} from "@lodestar/utils";
 import {getBlobKzgCommitments} from "../../util/dataColumns.js";
 import {BeaconChain} from "../chain.js";
-import {
-  IBlockInput,
-  IDataColumnsInput,
-  isBlockInputBlobs,
-  isBlockInputColumns,
-  isBlockInputNoData,
-} from "./blockInput/index.js";
+import {IBlockInput, IDataColumnsInput, isBlockInputBlobs, isBlockInputColumns} from "./blockInput/index.js";
 import {BLOB_AVAILABILITY_TIMEOUT} from "./verifyBlocksDataAvailability.js";
 
 /**
@@ -137,17 +131,6 @@ export async function persistBlockInput(this: BeaconChain, blockInput: IBlockInp
     })
     .finally(() => {
       this.seenBlockInputCache.prune(blockInput.blockRootHex);
-      // Without forcefully clearing this cache, we would rely on WeakMap to evict memory which is not reliable.
-      // Clear here (after the DB write) so that writeBlockInputToDb can still use the cached serialized bytes.
-      //
-      // For Gloas (BlockInputNoData), the execution payload and columns arrive separately after the beacon block.
-      // Do NOT clear the cache here — it must remain available for writeDataColumnsToDb when the payload arrives.
-      // The cache is cleared in the Gloas payload persistence path instead.
-      if (!isBlockInputNoData(blockInput)) {
-        // TODO: enhance this SerializedCache for Gloas because payload may not come
-        // see https://github.com/ChainSafe/lodestar/pull/8974#discussion_r2885598229
-        this.serializedCache.clear();
-      }
       this.logger.debug("Pruned block input", {
         slot: blockInput.slot,
         root: blockInput.blockRootHex,

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

@@ -320,12 +320,14 @@ export class BeaconChain implements IBeaconChain {
 
     this.beaconProposerCache = new BeaconProposerCache(opts);
     this.checkpointBalancesCache = new CheckpointBalancesCache();
+    this.serializedCache = new SerializedCache();
     this.seenBlockInputCache = new SeenBlockInput({
       config,
       custodyConfig: this.custodyConfig,
       clock,
       chainEvents: emitter,
       signal,
+      serializedCache: this.serializedCache,
       metrics,
       logger,
     });
@@ -412,8 +414,6 @@ export class BeaconChain implements IBeaconChain {
     this.bls = bls;
     this.emitter = emitter;
 
-    this.serializedCache = new SerializedCache();
-
     this.getBlobsTracker = new GetBlobsTracker({
       logger,
       executionEngine: this.executionEngine,

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

@@ -17,6 +17,7 @@ import {Metrics} from "../../metrics/metrics.js";
 import {MAX_LOOK_AHEAD_EPOCHS} from "../../sync/constants.js";
 import {IClock} from "../../util/clock.js";
 import {CustodyConfig} from "../../util/dataColumns.js";
+import {SerializedCache} from "../../util/serializedCache.js";
 import {
   BlockInput,
   BlockInputBlobs,
@@ -55,6 +56,7 @@ export type SeenBlockInputCacheModules = {
   chainEvents: ChainEventEmitter;
   signal: AbortSignal;
   custodyConfig: CustodyConfig;
+  serializedCache: SerializedCache;
   metrics: Metrics | null;
   logger?: Logger;
 };
@@ -101,6 +103,7 @@ export class SeenBlockInput {
   private readonly clock: IClock;
   private readonly chainEvents: ChainEventEmitter;
   private readonly signal: AbortSignal;
+  private readonly serializedCache: SerializedCache;
   private readonly metrics: Metrics | null;
   private readonly logger?: Logger;
   private blockInputs = new Map<RootHex, IBlockInput>();
@@ -109,19 +112,35 @@ export class SeenBlockInput {
   // and the signature to ensure we only skip verification if both match
   private verifiedProposerSignatures = new Map<Slot, Map<RootHex, BLSSignature>>();
 
-  constructor({config, custodyConfig, clock, chainEvents, signal, metrics, logger}: SeenBlockInputCacheModules) {
+  constructor({
+    config,
+    custodyConfig,
+    clock,
+    chainEvents,
+    signal,
+    serializedCache,
+    metrics,
+    logger,
+  }: SeenBlockInputCacheModules) {
     this.config = config;
     this.custodyConfig = custodyConfig;
     this.clock = clock;
     this.chainEvents = chainEvents;
     this.signal = signal;
+    this.serializedCache = serializedCache;
     this.metrics = metrics;
     this.logger = logger;
 
     if (metrics) {
-      metrics.seenCache.blockInput.blockInputCount.addCollect(() =>
-        metrics.seenCache.blockInput.blockInputCount.set(this.blockInputs.size)
-      );
+      metrics.seenCache.blockInput.blockInputCount.addCollect(() => {
+        metrics.seenCache.blockInput.blockInputCount.set(this.blockInputs.size);
+        metrics.seenCache.blockInput.serializedObjectCount.set(
+          Array.from(this.blockInputs.values()).reduce(
+            (count, blockInput) => count + blockInput.getSerializedCacheKeys().length,
+            0
+          )
+        );
+      });
     }
 
     this.chainEvents.on(ChainEvent.forkChoiceFinalized, this.onFinalized);
@@ -142,7 +161,10 @@ export class SeenBlockInput {
    * Removes the single BlockInput from the cache
    */
   remove(rootHex: RootHex): void {
-    this.blockInputs.delete(rootHex);
+    const blockInput = this.blockInputs.get(rootHex);
+    if (blockInput) {
+      this.evictBlockInput(blockInput);
+    }
   }
 
   /**
@@ -154,7 +176,7 @@ export class SeenBlockInput {
     let deletedCount = 0;
     while (blockInput) {
       deletedCount++;
-      this.blockInputs.delete(blockInput.blockRootHex);
+      this.evictBlockInput(blockInput);
       blockInput = this.blockInputs.get(parentRootHex ?? "");
       parentRootHex = blockInput?.parentRootHex;
     }
@@ -165,10 +187,10 @@ export class SeenBlockInput {
   onFinalized = (checkpoint: CheckpointWithHex) => {
     let deletedCount = 0;
     const cutoffSlot = computeStartSlotAtEpoch(checkpoint.epoch);
-    for (const [rootHex, blockInput] of this.blockInputs) {
+    for (const [, blockInput] of this.blockInputs) {
       if (blockInput.slot < cutoffSlot) {
         deletedCount++;
-        this.blockInputs.delete(rootHex);
+        this.evictBlockInput(blockInput);
       }
     }
     this.logger?.debug(`BlockInputCache.onFinalized deleted ${deletedCount} cached BlockInputs`);
@@ -408,14 +430,20 @@ export class SeenBlockInput {
 
     if (itemsToDelete > 0) {
       const sorted = [...this.blockInputs.entries()].sort((a, b) => a[1].slot - b[1].slot);
-      for (const [rootHex] of sorted) {
-        this.blockInputs.delete(rootHex);
+      for (const [, blockInput] of sorted) {
+        this.evictBlockInput(blockInput);
         itemsToDelete--;
         if (itemsToDelete <= 0) return;
       }
     }
     pruneSetToMax(this.verifiedProposerSignatures, MAX_BLOCK_INPUT_CACHE_SIZE);
   }
+
+  private evictBlockInput(blockInput: IBlockInput): void {
+    // Without forcefully clearing this cache, we would rely on WeakMap to evict memory which is not reliable
+    this.serializedCache.delete(blockInput.getSerializedCacheKeys());
+    this.blockInputs.delete(blockInput.blockRootHex);
+  }
 }
 
 enum SeenBlockInputCacheErrorCode {

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

@@ -1461,6 +1461,10 @@ export function createLodestarMetrics(
           name: "lodestar_seen_block_input_cache_size",
           help: "Number of cached BlockInputs",
         }),
+        serializedObjectCount: register.gauge({
+          name: "lodestar_seen_block_input_cache_serialized_object_count",
+          help: "Number of serialized objects retained by cached BlockInputs",
+        }),
         duplicateBlockCount: register.gauge<{source: BlockInputSource}>({
           name: "lodestar_seen_block_input_cache_duplicate_block_count",
           help: "Total number of duplicate blocks that pass validation and attempt to be cached but are known",

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

@@ -539,7 +539,8 @@ export class Network implements INetwork {
         this.config.getForkSeq(this.clock.currentSlot) >= ForkSeq.altair ? [Version.V2] : [Version.V2, Version.V1],
         request
       ),
-      request
+      request,
+      this.chain.serializedCache
     );
   }
 

packages/beacon-node/src/util/serializedCache.ts

@@ -4,7 +4,7 @@
  * This is a thin wrapper around WeakMap
  */
 export class SerializedCache {
-  map: WeakMap<object, Uint8Array> = new WeakMap();
+  private map: WeakMap<object, Uint8Array> = new WeakMap();
 
   get(obj: object): Uint8Array | undefined {
     return this.map.get(obj);
@@ -15,11 +15,13 @@ export class SerializedCache {
   }
 
   /**
-   * Replace the internal WeakMap to force GC of all cached entries.
-   * Must only be called after all DB writes that may read from this cache have completed,
+   * Delete cached serialized entries for the provided object references.
+   * Must only be called after all DB writes that read from this cache for these objects have completed,
    * otherwise cached serialized bytes will be unavailable and data will be re-serialized unnecessarily.
    */
-  clear(): void {
-    this.map = new WeakMap();
+  delete(objs: object[]): void {
+    for (const obj of objs) {
+      this.map.delete(obj);
+    }
   }
 }