fix: share in-flight payload envelope import promise to prevent sync spin loop

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

@@ -8,28 +8,24 @@ import {ImportPayloadOpts} from "./types.js";
 // TODO GLOAS: Set to be equal to DEFAULT_MAX_PENDING_UNFINALIZED_PAYLOAD_ENVELOPE_WRITES for now
 const QUEUE_MAX_LENGTH = 16;
 
-enum PayloadEnvelopeImportStatus {
-  queued = "queued",
-  importing = "importing",
-  imported = "imported",
-}
-
 /**
  * PayloadEnvelopeProcessor processes payload envelope jobs in a queued fashion, one after the other.
  *
  * Jobs are enqueued only on envelope arrival (gossip or API). The envelope may reach us before
  * the sampled data columns; importExecutionPayload awaits `verifyPayloadsDataAvailability`
  * internally, so a queued job can pend for up to `PAYLOAD_DATA_AVAILABILITY_TIMEOUT` while
- * waiting for columns. Duplicate triggers for the same payloadInput are deduped via `importStatus`.
+ * waiting for columns. Duplicate triggers for the same payloadInput are deduped by sharing the
+ * in-flight import promise: every caller observes the real import outcome. Resolving duplicates
+ * early instead would let BlockInputSync mark a still-pending payload as processed, re-queue it,
+ * and spin re-processing the same root until OOM while the import is blocked (e.g. EL syncing).
  */
 export class PayloadEnvelopeProcessor {
   readonly jobQueue: JobItemQueue<[PayloadEnvelopeInput, ImportPayloadOpts], void>;
-  private readonly importStatus = new WeakMap<PayloadEnvelopeInput, PayloadEnvelopeImportStatus>();
+  private readonly imports = new WeakMap<PayloadEnvelopeInput, Promise<void>>();
 
   constructor(chain: BeaconChain, metrics: Metrics | null, signal: AbortSignal) {
     this.jobQueue = new JobItemQueue<[PayloadEnvelopeInput, ImportPayloadOpts], void>(
       (payloadInput, opts) => {
-        this.importStatus.set(payloadInput, PayloadEnvelopeImportStatus.importing);
         return processExecutionPayload.call(chain, payloadInput, signal, opts);
       },
       {maxLength: QUEUE_MAX_LENGTH, noYieldIfOneItem: true, signal},
@@ -38,24 +34,28 @@ export class PayloadEnvelopeProcessor {
   }
 
   async processPayloadEnvelopeJob(payloadInput: PayloadEnvelopeInput, opts: ImportPayloadOpts = {}): Promise<void> {
-    if (this.importStatus.get(payloadInput) !== undefined) {
-      return;
+    const existing = this.imports.get(payloadInput);
+    if (existing) {
+      return existing;
     }
 
     await this.jobQueue.waitForSpace();
 
     // Re-check after await, as another call may have queued this payload.
-    if (this.importStatus.get(payloadInput) !== undefined) {
-      return;
+    const queued = this.imports.get(payloadInput);
+    if (queued) {
+      return queued;
     }
 
-    this.importStatus.set(payloadInput, PayloadEnvelopeImportStatus.queued);
+    const importPromise = this.jobQueue.push(payloadInput, opts);
+    this.imports.set(payloadInput, importPromise);
 
     try {
-      await this.jobQueue.push(payloadInput, opts);
-      this.importStatus.set(payloadInput, PayloadEnvelopeImportStatus.imported);
+      await importPromise;
     } catch (e) {
-      this.importStatus.delete(payloadInput);
+      // Drop the failed import so a later attempt can retry once the failure cause is resolved
+      // (e.g. BLOCK_NOT_IN_FORK_CHOICE after the block lands in fork choice)
+      this.imports.delete(payloadInput);
       throw e;
     }
   }

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

@@ -0,0 +1,91 @@
+import {afterEach, beforeEach, describe, expect, it, vi} from "vitest";
+import {processExecutionPayload} from "../../../../src/chain/blocks/importExecutionPayload.js";
+import {PayloadEnvelopeProcessor} from "../../../../src/chain/blocks/payloadEnvelopeProcessor.js";
+import type {BeaconChain} from "../../../../src/chain/chain.js";
+import {PayloadEnvelopeInput} from "../../../../src/chain/seenCache/seenPayloadEnvelopeInput.js";
+
+vi.mock("../../../../src/chain/blocks/importExecutionPayload.js");
+
+describe("chain / blocks / PayloadEnvelopeProcessor", () => {
+  let processor: PayloadEnvelopeProcessor;
+  let abortController: AbortController;
+  let payloadInput: PayloadEnvelopeInput;
+
+  beforeEach(() => {
+    abortController = new AbortController();
+    processor = new PayloadEnvelopeProcessor({} as BeaconChain, null, abortController.signal);
+    // processExecutionPayload is mocked, the input is only used as an identity key
+    payloadInput = {} as PayloadEnvelopeInput;
+  });
+
+  afterEach(() => {
+    abortController.abort();
+    vi.resetAllMocks();
+  });
+
+  it("should run the import once for duplicate calls and resolve all callers on completion", async () => {
+    let resolveImport: () => void = () => {};
+    vi.mocked(processExecutionPayload).mockReturnValue(
+      new Promise((resolve) => {
+        resolveImport = resolve;
+      })
+    );
+
+    let firstResolved = false;
+    let secondResolved = false;
+    const first = processor.processPayloadEnvelopeJob(payloadInput).then(() => {
+      firstResolved = true;
+    });
+    const second = processor.processPayloadEnvelopeJob(payloadInput).then(() => {
+      secondResolved = true;
+    });
+
+    // Duplicate calls must not resolve before the import completes, else callers like
+    // BlockInputSync treat a still-pending payload as processed and re-queue it in a tight loop
+    await new Promise((resolve) => setTimeout(resolve, 0));
+    expect(firstResolved).toBe(false);
+    expect(secondResolved).toBe(false);
+    expect(processExecutionPayload).toHaveBeenCalledTimes(1);
+
+    resolveImport();
+    await Promise.all([first, second]);
+    expect(firstResolved).toBe(true);
+    expect(secondResolved).toBe(true);
+    expect(processExecutionPayload).toHaveBeenCalledTimes(1);
+  });
+
+  it("should propagate import failure to deduped callers", async () => {
+    let rejectImport: (e: Error) => void = () => {};
+    vi.mocked(processExecutionPayload).mockReturnValue(
+      new Promise((_resolve, reject) => {
+        rejectImport = reject;
+      })
+    );
+
+    const first = processor.processPayloadEnvelopeJob(payloadInput);
+    const second = processor.processPayloadEnvelopeJob(payloadInput);
+
+    rejectImport(new Error("IMPORT_FAILED"));
+    await expect(first).rejects.toThrow("IMPORT_FAILED");
+    await expect(second).rejects.toThrow("IMPORT_FAILED");
+    expect(processExecutionPayload).toHaveBeenCalledTimes(1);
+  });
+
+  it("should retry the import after a failed attempt", async () => {
+    vi.mocked(processExecutionPayload).mockRejectedValueOnce(new Error("IMPORT_FAILED"));
+    vi.mocked(processExecutionPayload).mockResolvedValueOnce(undefined);
+
+    await expect(processor.processPayloadEnvelopeJob(payloadInput)).rejects.toThrow("IMPORT_FAILED");
+
+    await processor.processPayloadEnvelopeJob(payloadInput);
+    expect(processExecutionPayload).toHaveBeenCalledTimes(2);
+  });
+
+  it("should not re-run the import after success", async () => {
+    vi.mocked(processExecutionPayload).mockResolvedValue(undefined);
+
+    await processor.processPayloadEnvelopeJob(payloadInput);
+    await processor.processPayloadEnvelopeJob(payloadInput);
+    expect(processExecutionPayload).toHaveBeenCalledTimes(1);
+  });
+});