fix: Restart proving job if epoch content changes (#12655)

As another fix for #12625, adds a recurring check on the epoch's block
headers to the proving job. If a mismatch is found, the job is aborted
and restarted with fresh data.

Also removes the unused `prove` call on the prover-node.

See also #12638

Author: spalladino

yarn-project/archiver/src/archiver/archiver.ts

@@ -612,6 +612,24 @@ export class Archiver extends EventEmitter implements ArchiveSource, Traceable {
     return blocks.reverse();
   }
 
+  public async getBlockHeadersForEpoch(epochNumber: bigint): Promise<BlockHeader[]> {
+    const [start, end] = getSlotRangeForEpoch(epochNumber, this.l1constants);
+    const blocks: BlockHeader[] = [];
+
+    // Walk the list of blocks backwards and filter by slots matching the requested epoch.
+    // We'll typically ask for blocks for a very recent epoch, so we shouldn't need an index here.
+    let number = await this.store.getSynchedL2BlockNumber();
+    let header = await this.getBlockHeader(number);
+    const slot = (b: BlockHeader) => b.globalVariables.slotNumber.toBigInt();
+    while (header && slot(header) >= start) {
+      if (slot(header) <= end) {
+        blocks.push(header);
+      }
+      header = await this.getBlockHeader(--number);
+    }
+    return blocks.reverse();
+  }
+
   public async isEpochComplete(epochNumber: bigint): Promise<boolean> {
     // The epoch is complete if the current L2 block is the last one in the epoch (or later)
     const header = await this.getBlockHeader('latest');

yarn-project/archiver/src/test/mock_l2_block_source.ts

@@ -118,6 +118,10 @@ export class MockL2BlockSource implements L2BlockSource {
     return Promise.resolve(blocks);
   }
 
+  getBlockHeadersForEpoch(epochNumber: bigint): Promise<BlockHeader[]> {
+    return this.getBlocksForEpoch(epochNumber).then(blocks => blocks.map(b => b.header));
+  }
+
   /**
    * Gets a tx effect.
    * @param txHash - The hash of a transaction which resulted in the returned tx effect.

yarn-project/foundation/src/promise/running-promise.ts

@@ -73,6 +73,7 @@ export class RunningPromise {
       }
     };
     this.runningPromise = poll();
+    return this;
   }
 
   /**

yarn-project/foundation/src/retry/index.ts

@@ -83,7 +83,12 @@ export async function retry<Result>(
  * @param interval - The optional interval, in seconds, between retry attempts. Defaults to 1 second.
  * @returns A Promise that resolves with the successful (truthy) result of the provided function, or rejects if timeout is exceeded.
  */
-export async function retryUntil<T>(fn: () => Promise<T | undefined>, name = '', timeout = 0, interval = 1) {
+export async function retryUntil<T>(
+  fn: () => (T | undefined) | Promise<T | undefined>,
+  name = '',
+  timeout = 0,
+  interval = 1,
+) {
   const timer = new Timer();
   while (true) {
     const result = await fn();

yarn-project/prover-node/src/job/epoch-proving-job.test.ts

@@ -3,6 +3,7 @@ import { toArray } from '@aztec/foundation/iterable';
 import { sleep } from '@aztec/foundation/sleep';
 import type { PublicProcessor, PublicProcessorFactory } from '@aztec/simulator/server';
 import { L2Block, type L2BlockSource } from '@aztec/stdlib/block';
+import type { L1RollupConstants } from '@aztec/stdlib/epoch-helpers';
 import type { EpochProver, MerkleTreeWriteOperations, WorldStateSynchronizer } from '@aztec/stdlib/interfaces/server';
 import type { L1ToL2MessageSource } from '@aztec/stdlib/messaging';
 import { Proof } from '@aztec/stdlib/proofs';
@@ -85,9 +86,12 @@ describe('epoch-proving-job', () => {
 
     l1ToL2MessageSource.getL1ToL2Messages.mockResolvedValue([]);
     l2BlockSource.getBlockHeader.mockResolvedValue(initialHeader);
+    l2BlockSource.getL1Constants.mockResolvedValue({ ethereumSlotDuration: 0.1 } as L1RollupConstants);
+    l2BlockSource.getBlockHeadersForEpoch.mockResolvedValue(blocks.map(b => b.header));
     publicProcessorFactory.create.mockReturnValue(publicProcessor);
     db.getInitialHeader.mockReturnValue(initialHeader);
     worldState.fork.mockResolvedValue(db);
+    prover.startNewBlock.mockImplementation(() => sleep(200));
     prover.finaliseEpoch.mockResolvedValue({ publicInputs, proof });
     publisher.submitEpochProof.mockResolvedValue(true);
     publicProcessor.process.mockImplementation(async txs => {
@@ -146,7 +150,6 @@ describe('epoch-proving-job', () => {
   });
 
   it('halts if stopped externally', async () => {
-    prover.startNewBlock.mockImplementation(() => sleep(200));
     const job = createJob();
     void job.run();
     await sleep(100);
@@ -155,4 +158,16 @@ describe('epoch-proving-job', () => {
     expect(job.getState()).toEqual('stopped');
     expect(publisher.submitEpochProof).not.toHaveBeenCalled();
   });
+
+  it('halts if a new block for the epoch is found', async () => {
+    const newBlocks = await timesParallel(NUM_BLOCKS + 1, i => L2Block.random(i + 1, TXS_PER_BLOCK));
+    l2BlockSource.getBlockHeadersForEpoch.mockResolvedValue(newBlocks.map(b => b.header));
+
+    const job = createJob();
+    await job.run();
+
+    expect(job.getState()).toEqual('reorg');
+    expect(publisher.submitEpochProof).not.toHaveBeenCalled();
+    expect(prover.cancel).toHaveBeenCalled();
+  });
 });

yarn-project/prover-node/src/job/epoch-proving-job.ts

@@ -1,6 +1,6 @@
 import { asyncPool } from '@aztec/foundation/async-pool';
 import { createLogger } from '@aztec/foundation/log';
-import { promiseWithResolvers } from '@aztec/foundation/promise';
+import { RunningPromise, promiseWithResolvers } from '@aztec/foundation/promise';
 import { Timer } from '@aztec/foundation/timer';
 import type { PublicProcessor, PublicProcessorFactory } from '@aztec/simulator/server';
 import type { L2Block, L2BlockSource } from '@aztec/stdlib/block';
@@ -30,6 +30,7 @@ export class EpochProvingJob implements Traceable {
   private uuid: string;
 
   private runPromise: Promise<void> | undefined;
+  private epochCheckPromise: RunningPromise | undefined;
   private deadlineTimeoutHandler: NodeJS.Timeout | undefined;
 
   public readonly tracer: Tracer;
@@ -47,7 +48,6 @@ export class EpochProvingJob implements Traceable {
     private metrics: ProverNodeMetrics,
     private deadline: Date | undefined,
     private config: { parallelBlockLimit: number } = { parallelBlockLimit: 32 },
-    private cleanUp: (job: EpochProvingJob) => Promise<void> = () => Promise.resolve(),
   ) {
     this.uuid = crypto.randomUUID();
     this.tracer = metrics.client.getTracer('EpochProvingJob');
@@ -73,6 +73,7 @@ export class EpochProvingJob implements Traceable {
   })
   public async run() {
     this.scheduleDeadlineStop();
+    await this.scheduleEpochCheck();
 
     const epochNumber = Number(this.epochNumber);
     const epochSizeBlocks = this.blocks.length;
@@ -154,14 +155,18 @@ export class EpochProvingJob implements Traceable {
       this.metrics.recordProvingJob(executionTime, timer.ms(), epochSizeBlocks, epochSizeTxs);
     } catch (err: any) {
       if (err && err.name === 'HaltExecutionError') {
-        this.log.warn(`Halted execution of epoch ${epochNumber} prover job`, { uuid: this.uuid, epochNumber });
+        this.log.warn(`Halted execution of epoch ${epochNumber} prover job`, {
+          uuid: this.uuid,
+          epochNumber,
+          details: err.message,
+        });
         return;
       }
       this.log.error(`Error running epoch ${epochNumber} prover job`, err, { uuid: this.uuid, epochNumber });
       this.state = 'failed';
     } finally {
       clearTimeout(this.deadlineTimeoutHandler);
-      await this.cleanUp(this);
+      await this.epochCheckPromise?.stop();
       await this.prover.stop();
       resolve();
     }
@@ -173,12 +178,12 @@ export class EpochProvingJob implements Traceable {
   }
 
   private checkState() {
-    if (this.state === 'timed-out' || this.state === 'stopped' || this.state === 'failed') {
+    if (this.state === 'timed-out' || this.state === 'stopped' || this.state === 'failed' || this.state === 'reorg') {
       throw new HaltExecutionError(this.state);
     }
   }
 
-  public async stop(state: EpochProvingJobState = 'stopped') {
+  public async stop(state: EpochProvingJobTerminalState = 'stopped') {
     this.state = state;
     this.prover.cancel();
     // TODO(palla/prover): Stop the publisher as well
@@ -207,6 +212,36 @@ export class EpochProvingJob implements Traceable {
     }
   }
 
+  /**
+   * Kicks off a running promise that queries the archiver for the set of L2 blocks of the current epoch.
+   * If those changed, stops the proving job with a `rerun` state, so the node re-enqueues it.
+   */
+  private async scheduleEpochCheck() {
+    const intervalMs = Math.ceil((await this.l2BlockSource.getL1Constants()).ethereumSlotDuration / 2) * 1000;
+    this.epochCheckPromise = new RunningPromise(
+      async () => {
+        const blocks = await this.l2BlockSource.getBlockHeadersForEpoch(this.epochNumber);
+        const blockHashes = await Promise.all(blocks.map(block => block.hash()));
+        const thisBlockHashes = await Promise.all(this.blocks.map(block => block.hash()));
+        if (
+          blocks.length !== this.blocks.length ||
+          !blockHashes.every((block, i) => block.equals(thisBlockHashes[i]))
+        ) {
+          this.log.warn('Epoch blocks changed underfoot', {
+            uuid: this.uuid,
+            epochNumber: this.epochNumber,
+            oldBlockHashes: thisBlockHashes,
+            newBlockHashes: blockHashes,
+          });
+          void this.stop('reorg');
+        }
+      },
+      this.log,
+      intervalMs,
+    ).start();
+    this.log.verbose(`Scheduled epoch check for epoch ${this.epochNumber} every ${intervalMs}ms`);
+  }
+
   /* Returns the header for the given block number, or the genesis block for block zero. */
   private async getBlockHeader(blockNumber: number) {
     if (blockNumber === 0) {
@@ -249,7 +284,7 @@ export class EpochProvingJob implements Traceable {
 }
 
 class HaltExecutionError extends Error {
-  constructor(state: EpochProvingJobState) {
+  constructor(public readonly state: EpochProvingJobState) {
     super(`Halted execution due to state ${state}`);
     this.name = 'HaltExecutionError';
   }

yarn-project/prover-node/src/prover-node.test.ts

@@ -1,5 +1,8 @@
 import { timesParallel } from '@aztec/foundation/collection';
 import { EthAddress } from '@aztec/foundation/eth-address';
+import { promiseWithResolvers } from '@aztec/foundation/promise';
+import { retryUntil } from '@aztec/foundation/retry';
+import { sleep } from '@aztec/foundation/sleep';
 import type { PublicProcessorFactory } from '@aztec/simulator/server';
 import { L2Block, type L2BlockSource } from '@aztec/stdlib/block';
 import type { ContractDataSource } from '@aztec/stdlib/contract';
@@ -45,11 +48,7 @@ describe('prover-node', () => {
   let address: EthAddress;
 
   // List of all jobs ever created by the test prover node and their dependencies
-  let jobs: {
-    job: MockProxy<EpochProvingJob>;
-    cleanUp: (job: EpochProvingJob) => Promise<void>;
-    epochNumber: bigint;
-  }[];
+  let jobs: { job: MockProxy<EpochProvingJob>; epochNumber: bigint }[];
 
   const createProverNode = () =>
     new TestProverNode(
@@ -140,39 +139,70 @@ describe('prover-node', () => {
   it('starts a proof on a finished epoch', async () => {
     await proverNode.handleEpochReadyToProve(10n);
     expect(jobs[0].epochNumber).toEqual(10n);
+    expect(proverNode.totalJobCount).toEqual(1);
   });
 
   it('does not start a proof if there are no blocks in the epoch', async () => {
     l2BlockSource.getBlocksForEpoch.mockResolvedValue([]);
     await proverNode.handleEpochReadyToProve(10n);
-    expect(jobs.length).toEqual(0);
+    expect(proverNode.totalJobCount).toEqual(0);
   });
 
   it('does not start a proof if there is a tx missing from coordinator', async () => {
     mockCoordination.getTxsByHash.mockResolvedValue([]);
     await proverNode.handleEpochReadyToProve(10n);
-    expect(jobs.length).toEqual(0);
+    expect(proverNode.totalJobCount).toEqual(0);
   });
 
   it('does not prove the same epoch twice', async () => {
+    const firstJob = promiseWithResolvers<void>();
+    proverNode.nextJobRun = () => firstJob.promise;
+    proverNode.nextJobState = 'processing';
     await proverNode.handleEpochReadyToProve(10n);
     await proverNode.handleEpochReadyToProve(10n);
 
-    expect(jobs.length).toEqual(1);
+    firstJob.resolve();
+    expect(proverNode.totalJobCount).toEqual(1);
+  });
+
+  it('restarts a proof on a reorg', async () => {
+    proverNode.nextJobState = 'reorg';
+    await proverNode.handleEpochReadyToProve(10n);
+    await retryUntil(() => proverNode.totalJobCount === 2, 'job retried', 5);
+    expect(proverNode.totalJobCount).toEqual(2);
+  });
+
+  it('does not restart a proof on an error', async () => {
+    proverNode.nextJobState = 'failed';
+    await proverNode.handleEpochReadyToProve(10n);
+    await sleep(1000);
+    expect(proverNode.totalJobCount).toEqual(1);
   });
 
   class TestProverNode extends ProverNode {
+    public totalJobCount = 0;
+    public nextJobState: EpochProvingJobState = 'completed';
+    public nextJobRun: () => Promise<void> = () => Promise.resolve();
+
     protected override doCreateEpochProvingJob(
       epochNumber: bigint,
       _deadline: Date | undefined,
       _blocks: L2Block[],
       _txs: Tx[],
       _publicProcessorFactory: PublicProcessorFactory,
-      cleanUp: (job: EpochProvingJob) => Promise<void>,
     ): EpochProvingJob {
-      const job = mock<EpochProvingJob>({ getState: () => 'processing', run: () => Promise.resolve() });
+      const state = this.nextJobState;
+      this.nextJobState = 'completed';
+      const run = this.nextJobRun;
+      this.nextJobRun = () => Promise.resolve();
+      const job = mock<EpochProvingJob>({
+        getState: () => state,
+        run,
+        getEpochNumber: () => epochNumber,
+      });
       job.getId.mockReturnValue(jobs.length.toString());
-      jobs.push({ epochNumber, job, cleanUp });
+      jobs.push({ epochNumber, job });
+      this.totalJobCount++;
       return job;
     }