ChainSafe · matthewkeil · Feb 11, 2026 · Feb 9, 2026 · Feb 9, 2026 · Feb 9, 2026
@@ -91,6 +91,10 @@ export async function importBlock(
   }
 
   // 1. Persist block to hot DB (performed asynchronously to avoid blocking head selection)
+  // Wait for space in the write queue to apply backpressure during sync.
+  // Without this, a supernode syncing from behind can accumulate many blocks worth of column
+  // data in memory (up to 128 columns per block) causing OOM before persistence catches up.
+  await this.unfinalizedBlockWrites.waitForSpace();
   this.unfinalizedBlockWrites.push([blockInput]).catch((e) => {
     if (!isQueueErrorAborted(e)) {
       this.logger.error("Error pushing block to unfinalized write queue", {slot: blockSlot}, e as Error);

@@ -126,8 +126,11 @@ const DEFAULT_MAX_CACHED_PRODUCED_RESULTS = 4;
 
 /**
  * The maximum number of pending unfinalized block writes to the database before backpressure is applied.
+ * Write queue entries hold references to block inputs, keeping them in memory even after cache eviction.
+ * This is especially important for supernodes which store all 128 columns per block — each pending
+ * write can hold significant memory. Keep moderate to avoid OOM during sync.
  */
-const DEFAULT_MAX_PENDING_UNFINALIZED_BLOCK_WRITES = 32;
+const DEFAULT_MAX_PENDING_UNFINALIZED_BLOCK_WRITES = 16;
 
 export class BeaconChain implements IBeaconChain {
   readonly genesisTime: UintNum64;

@@ -1,10 +1,19 @@
 import {ChainForkConfig} from "@lodestar/config";
 import {CheckpointWithHex} from "@lodestar/fork-choice";
-import {ForkName, ForkPostFulu, ForkPreGloas, isForkPostDeneb, isForkPostFulu, isForkPostGloas} from "@lodestar/params";
+import {
+  ForkName,
+  ForkPostFulu,
+  ForkPreGloas,
+  SLOTS_PER_EPOCH,
+  isForkPostDeneb,
+  isForkPostFulu,
+  isForkPostGloas,
+} from "@lodestar/params";
 import {computeStartSlotAtEpoch} from "@lodestar/state-transition";
 import {BLSSignature, RootHex, SignedBeaconBlock, Slot, deneb, fulu} from "@lodestar/types";
 import {LodestarError, Logger, byteArrayEquals, pruneSetToMax} from "@lodestar/utils";
 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 {
@@ -26,7 +35,17 @@ import {
 } from "../blocks/blockInput/index.js";
 import {ChainEvent, ChainEventEmitter} from "../emitter.js";
 
-const MAX_BLOCK_INPUT_CACHE_SIZE = 5;
+// Target size for the block input cache, enforced by pruneToMaxSize() which runs after prune()
+// and onFinalized() — NOT on insertion. The cache can temporarily exceed this during range sync
+// (e.g. 32 blocks inserted per batch) but is trimmed back after blocks are processed.
+//
+// Must be large enough to hold blocks from all concurrently downloaded range sync batches.
+// Range sync downloads up to MAX_LOOK_AHEAD_EPOCHS batches ahead of the processing head,
+// so up to (MAX_LOOK_AHEAD_EPOCHS + 1) batches (current + look-ahead) of SLOTS_PER_EPOCH
+// blocks can be in the cache simultaneously. If this value is too small, pruneToMaxSize()
+// will evict blocks from the batch being processed before they are persisted to the database,
+// causing errors when async handlers like onForkChoiceFinalized run.
+const MAX_BLOCK_INPUT_CACHE_SIZE = (MAX_LOOK_AHEAD_EPOCHS + 1) * SLOTS_PER_EPOCH;
 
 export type SeenBlockInputCacheModules = {
   config: ChainForkConfig;
@@ -64,14 +83,14 @@ export type GetByBlobOptions = {
  * - onFinalized event handler will help to prune any non-canonical forks once the chain finalizes. Any block-slots that
  *   are before the finalized checkpoint will be pruned.
  * - Range-sync periods.  The range process uses this cache to store and sync blocks with DA data as the chain is pulled
- *   from peers.  We pull batches, by epoch, so 32 slots are pulled at a time and several batches are pulled concurrently.
- *   It is important to set the MAX_BLOCK_INPUT_CACHE_SIZE high enough to support range sync activities.  Currently the
- *   value is set for 5 batches of 32 slots.  As process block is called (similar to following head) the BlockInput and
- *   its ancestors will be pruned.
+ *   from peers.  We pull batches, by epoch, so 32 slots are pulled at a time and several batches are downloaded
+ *   concurrently (up to MAX_LOOK_AHEAD_EPOCHS ahead).  All downloaded blocks are added to this shared cache, so it
+ *   must be large enough to hold blocks from all concurrent batches.  If pruneToMaxSize() evicts blocks from the batch
+ *   currently being processed, those blocks may not yet be persisted to the database, causing getBlockByRoot() to fail
+ *   when async event handlers (e.g. onForkChoiceFinalized) try to look them up.
  * - Non-Finality times.  This is a bit more tricky.  There can be long periods of non-finality and storing everything
- *   will cause OOM.  The pruneToMax will help ensure a hard limit on the number of stored blocks (with DA) that are held
- *   in memory at any one time.  The value for MAX_BLOCK_INPUT_CACHE_SIZE is set to accommodate range-sync but in
- *   practice this value may need to be massaged in the future if we find issues when debugging non-finality
+ *   will cause OOM.  The pruneToMaxSize will help ensure the number of stored blocks (with DA) is trimmed back to
+ *   MAX_BLOCK_INPUT_CACHE_SIZE after each prune() or onFinalized() call
  */
 
 export class SeenBlockInput {

@@ -24,6 +24,8 @@ export class JobItemQueue<Args extends any[], R> {
   private readonly metrics?: QueueMetrics;
   private runningJobs = 0;
   private lastYield = 0;
+  /** Resolvers waiting for space in the queue */
+  private spaceWaiters: (() => void)[] = [];
 
   constructor(
     private readonly itemProcessor: (...args: Args) => Promise<R>,
@@ -72,12 +74,57 @@ export class JobItemQueue<Args extends any[], R> {
     });
   }
 
+  /**
+   * Returns a promise that resolves when there is space in the queue.
+   * If the queue already has space, resolves immediately (noop).
+   * Use this to apply backpressure when the caller should wait rather than
+   * have push() throw QUEUE_MAX_LENGTH.
+   */
+  async waitForSpace(): Promise<void> {
+    if (this.opts.signal.aborted) {
+      throw new QueueError({code: QueueErrorCode.QUEUE_ABORTED});
+    }
+
+    if (this.jobs.length < this.opts.maxLength) {
+      return;
+    }
+
+    return new Promise<void>((resolve, reject) => {
+      let settled = false;
+
+      const onAbort = (): void => {
+        if (settled) return;
+        settled = true;
+        const index = this.spaceWaiters.indexOf(wrappedResolve);
+        if (index >= 0) {
+          this.spaceWaiters.splice(index, 1);
+        }
+        reject(new QueueError({code: QueueErrorCode.QUEUE_ABORTED}));
+      };
+
+      const wrappedResolve = (): void => {
+        if (settled) return;
+        settled = true;
+        this.opts.signal.removeEventListener("abort", onAbort);
+        resolve();
+      };
+
+      this.spaceWaiters.push(wrappedResolve);
+      this.opts.signal.addEventListener("abort", onAbort, {once: true});
+
+      // Re-check after attaching listener to close the race window where
+      // signal.abort() fires between the initial check and addEventListener
+      if (this.opts.signal.aborted) onAbort();
+    });
+  }
+
   getItems(): {args: Args; addedTimeMs: number}[] {
     return this.jobs.map((job) => ({args: job.args, addedTimeMs: job.addedTimeMs}));
   }
 
   dropAllJobs = (): void => {
     this.jobs.clear();
+    this.notifySpaceWaiters();
   };
 
   private runJob = async (): Promise<void> => {
@@ -115,10 +162,25 @@ export class JobItemQueue<Args extends any[], R> {
 
     this.runningJobs = Math.max(0, this.runningJobs - 1);
 
+    // Notify any waiters that space is available
+    this.notifySpaceWaiters();
+
     // Potentially run a new job
     void this.runJob();
   };
 
+  private notifySpaceWaiters(): void {
+    // Compute available slots once to avoid thundering herd: resolved waiters
+    // won't push() until the next microtask, so jobs.length doesn't change
+    // inside this loop. Without the cap we'd wake ALL waiters on a single slot.
+    let available = this.opts.maxLength - this.jobs.length;
+    while (available > 0 && this.spaceWaiters.length > 0) {
+      const resolve = this.spaceWaiters.shift();
+      if (resolve) resolve();
+      available--;
+    }
+  }
+
   private abortAllJobs = (): void => {
     while (this.jobs.length > 0) {
       const job = this.jobs.pop();

@@ -1,6 +1,6 @@
 import {describe, expect, it} from "vitest";
 import {sleep} from "@lodestar/utils";
-import {JobFnQueue, QueueError, QueueErrorCode, QueueType} from "../../../src/util/queue/index.js";
+import {JobFnQueue, JobItemQueue, QueueError, QueueErrorCode, QueueType} from "../../../src/util/queue/index.js";
 import {expectLodestarError, expectRejectedWithLodestarError} from "../../utils/errors.js";
 
 describe("Job queue", () => {
@@ -109,6 +109,108 @@ describe("Job queue", () => {
       });
     }
   });
+
+  describe("waitForSpace", () => {
+    const maxLength = 2;
+    const jobDuration = 50;
+
+    it("should resolve immediately when queue has space", async () => {
+      const controller = new AbortController();
+      const jobQueue = new JobItemQueue<[number], number>(async (n) => n, {maxLength, signal: controller.signal});
+
+      // Queue is empty, waitForSpace should resolve immediately
+      await jobQueue.waitForSpace();
+      controller.abort();
+    });
+
+    it("should wait until space is available when queue is full", async () => {
+      const controller = new AbortController();
+      const jobQueue = new JobItemQueue<[number], number>(
+        async (n) => {
+          await sleep(jobDuration);
+          return n;
+        },
+        {maxLength, signal: controller.signal}
+      );
+
+      // Fill the queue
+      const jobs = Array.from({length: maxLength}, (_, i) => jobQueue.push(i));
+
+      // Queue is full, waitForSpace should block
+      let spaceAvailable = false;
+      const waitPromise = jobQueue.waitForSpace().then(() => {
+        spaceAvailable = true;
+      });
+
+      // Give a tick for the wait to register
+      await sleep(5);
+      expect(spaceAvailable).toBe(false);
+
+      // Wait for a job to complete, which should free space
+      await Promise.all(jobs);
+      await waitPromise;
+      expect(spaceAvailable).toBe(true);
+
+      controller.abort();
+    });
+
+    it("should reject when aborted while waiting", async () => {
+      const controller = new AbortController();
+      const jobQueue = new JobItemQueue<[number], number>(
+        async (n) => {
+          await sleep(jobDuration);
+          return n;
+        },
+        {maxLength, signal: controller.signal}
+      );
+
+      // Fill the queue (catch rejections from abort to avoid unhandled rejection errors)
+      const jobs = Array.from({length: maxLength}, (_, i) => jobQueue.push(i).catch(() => 0));
+
+      // Wait for space, then abort
+      const waitPromise = jobQueue.waitForSpace();
+      controller.abort();
+
+      await expectRejectedWithLodestarError(waitPromise, new QueueError({code: QueueErrorCode.QUEUE_ABORTED}));
+      await Promise.allSettled(jobs);
+    });
+
+    it("should only wake one waiter per available slot (no thundering herd)", async () => {
+      const controller = new AbortController();
+      const jobQueue = new JobItemQueue<[number], number>(
+        async (n) => {
+          await sleep(jobDuration);
+          return n;
+        },
+        {maxLength, signal: controller.signal}
+      );
+
+      // Fill the queue
+      const jobs = Array.from({length: maxLength}, (_, i) => jobQueue.push(i));
+
+      // Register multiple waiters
+      const resolved: number[] = [];
+      const waiter1 = jobQueue.waitForSpace().then(() => resolved.push(1));
+      const waiter2 = jobQueue.waitForSpace().then(() => resolved.push(2));
+
+      // Wait for one job to complete (frees 1 slot)
+      await sleep(jobDuration + 10);
+
+      // Give microtasks time to settle
+      await sleep(5);
+
+      // Only one waiter should have been resolved (1 slot freed)
+      expect(resolved).toHaveLength(1);
+      expect(resolved[0]).toBe(1);
+
+      // Wait for the rest to complete
+      await Promise.all(jobs);
+      await Promise.all([waiter1, waiter2]);
+      expect(resolved).toHaveLength(2);
+
+      controller.abort();
+    });
+  });
 });
 
 async function wrapFn(fn: () => Promise<unknown>): Promise<unknown> {