diff --git a/packages/beacon-node/src/chain/blocks/payloadEnvelopeProcessor.ts b/packages/beacon-node/src/chain/blocks/payloadEnvelopeProcessor.ts index df9d871a0e4c..030b306c7eac 100644 --- a/packages/beacon-node/src/chain/blocks/payloadEnvelopeProcessor.ts +++ b/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(); + private readonly imports = new WeakMap>(); 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 { - 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; } } diff --git a/packages/beacon-node/test/unit/chain/blocks/payloadEnvelopeProcessor.test.ts b/packages/beacon-node/test/unit/chain/blocks/payloadEnvelopeProcessor.test.ts new file mode 100644 index 000000000000..353de17ff6a8 --- /dev/null +++ b/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); + }); +});