fix(slasher): handle multiple blocks per slot in epoch prune watcher

Update to e2e test pending.

Author: spalladino

yarn-project/end-to-end/src/e2e_p2p/valid_epoch_pruned_slash.test.ts

@@ -29,6 +29,8 @@ const DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), 'valid-epoch-pruned-slash
  * We don't need to do anything special for this test other than to run it without a prover node
  * (which is the default), and this will produce pruned epochs that could have been proven. But we do
  * need to send a tx to make sure that the slash is due to valid epoch prune and not data withholding.
+ *
+ * TODO(palla/mbps): Add tests for 1) out messages and 2) partial epoch prunes
  */
 describe('e2e_p2p_valid_epoch_pruned_slash', () => {
   let t: P2PNetworkTest;

yarn-project/slasher/src/watchers/epoch_prune_watcher.test.ts

@@ -1,5 +1,5 @@
 import type { EpochCache } from '@aztec/epoch-cache';
-import { BlockNumber, EpochNumber, SlotNumber } from '@aztec/foundation/branded-types';
+import { BlockNumber, CheckpointNumber, EpochNumber, SlotNumber } from '@aztec/foundation/branded-types';
 import { EthAddress } from '@aztec/foundation/eth-address';
 import { sleep } from '@aztec/foundation/sleep';
 import { L2Block, type L2BlockSourceEventEmitter, L2BlockSourceEvents } from '@aztec/stdlib/block';
@@ -76,12 +76,14 @@ describe('EpochPruneWatcher', () => {
   it('should emit WANT_TO_SLASH_EVENT when a validator is in a pruned epoch when data is unavailable', async () => {
     const emitSpy = jest.spyOn(watcher, 'emit');
     const epochNumber = EpochNumber(1);
+    const checkpointNumber = CheckpointNumber(1);
 
     const block = await L2Block.random(
       BlockNumber(12), // block number
       {
         txsPerBlock: 4,
         slotNumber: SlotNumber(10),
+        checkpointNumber,
       },
     );
     txProvider.getAvailableTxs.mockResolvedValue({ txs: [], missingTxs: [block.body.txEffects[0].txHash] });
@@ -124,12 +126,14 @@ describe('EpochPruneWatcher', () => {
 
   it('should slash if the data is available and the epoch could have been proven', async () => {
     const emitSpy = jest.spyOn(watcher, 'emit');
+    const checkpointNumber = CheckpointNumber(1);
 
     const block = await L2Block.random(
       BlockNumber(12), // block number
       {
         txsPerBlock: 4,
         slotNumber: SlotNumber(10),
+        checkpointNumber,
       },
     );
     const tx = Tx.random();
@@ -186,12 +190,14 @@ describe('EpochPruneWatcher', () => {
 
   it('should not slash if the data is available but the epoch could not have been proven', async () => {
     const emitSpy = jest.spyOn(watcher, 'emit');
+    const checkpointNumber = CheckpointNumber(1);
 
     const blockFromL1 = await L2Block.random(
       BlockNumber(12), // block number
       {
         txsPerBlock: 1,
         slotNumber: SlotNumber(10),
+        checkpointNumber,
       },
     );
 
@@ -200,6 +206,7 @@ describe('EpochPruneWatcher', () => {
       {
         txsPerBlock: 1,
         slotNumber: SlotNumber(10),
+        checkpointNumber,
       },
     );
     const tx = Tx.random();
@@ -244,6 +251,7 @@ describe('EpochPruneWatcher', () => {
 
 class MockL2BlockSource {
   public readonly events = new EventEmitter();
+  public getCheckpointsForEpoch = () => [];
 
   constructor() {}
 }

yarn-project/slasher/src/watchers/epoch_prune_watcher.ts

@@ -1,6 +1,6 @@
 import { EpochCache } from '@aztec/epoch-cache';
-import { BlockNumber, CheckpointNumber, EpochNumber } from '@aztec/foundation/branded-types';
-import { merge, pick } from '@aztec/foundation/collection';
+import { BlockNumber, EpochNumber } from '@aztec/foundation/branded-types';
+import { chunkBy, merge, pick } from '@aztec/foundation/collection';
 import type { Fr } from '@aztec/foundation/curves/bn254';
 import { type Logger, createLogger } from '@aztec/foundation/log';
 import {
@@ -12,6 +12,7 @@ import {
 } from '@aztec/stdlib/block';
 import { getEpochAtSlot } from '@aztec/stdlib/epoch-helpers';
 import type {
+  ICheckpointBlockBuilder,
   ICheckpointsBuilder,
   ITxProvider,
   MerkleTreeWriteOperations,
@@ -106,7 +107,7 @@ export class EpochPruneWatcher extends (EventEmitter as new () => WatcherEmitter
         { blocks: epochBlocks.map(b => b.toBlockInfo()) },
       );
 
-      await this.validateBlocks(epochBlocks);
+      await this.validateBlocks(epochBlocks, epochNumber);
       this.log.info(`Pruned epoch ${epochNumber} was valid. Want to slash committee for not having it proven.`);
       await this.emitSlashForEpoch(OffenseType.VALID_EPOCH_PRUNED, epochNumber);
     } catch (error) {
@@ -121,45 +122,52 @@ export class EpochPruneWatcher extends (EventEmitter as new () => WatcherEmitter
     }
   }
 
-  public async validateBlocks(blocks: L2Block[]): Promise<void> {
+  public async validateBlocks(blocks: L2Block[], epochNumber: EpochNumber): Promise<void> {
     if (blocks.length === 0) {
       return;
     }
 
-    let previousCheckpointOutHashes: Fr[] = [];
-    const fork = await this.checkpointsBuilder.getFork(BlockNumber(blocks[0].header.globalVariables.blockNumber - 1));
+    // Sort blocks by block number and group by checkpoint
+    const sortedBlocks = [...blocks].sort((a, b) => a.number - b.number);
+    const blocksByCheckpoint = chunkBy(sortedBlocks, b => b.checkpointNumber);
+
+    // Get prior checkpoints in the epoch (in case this was a partial prune) to extract the out hashes
+    const priorCheckpointOutHashes = (await this.l2BlockSource.getCheckpointsForEpoch(epochNumber))
+      .filter(c => c.number < sortedBlocks[0].checkpointNumber)
+      .map(c => c.getCheckpointOutHash());
+    let previousCheckpointOutHashes: Fr[] = [...priorCheckpointOutHashes];
+
+    const fork = await this.checkpointsBuilder.getFork(
+      BlockNumber(sortedBlocks[0].header.globalVariables.blockNumber - 1),
+    );
     try {
-      for (const block of blocks) {
-        await this.validateBlock(block, previousCheckpointOutHashes, fork);
+      for (const checkpointBlocks of blocksByCheckpoint) {
+        await this.validateCheckpoint(checkpointBlocks, previousCheckpointOutHashes, fork);
 
-        // TODO(mbps): This assumes one block per checkpoint, which is only true for now.
-        const checkpointOutHash = computeCheckpointOutHash([block.body.txEffects.map(tx => tx.l2ToL1Msgs)]);
+        // Compute checkpoint out hash from all blocks in this checkpoint
+        const checkpointOutHash = computeCheckpointOutHash(
+          checkpointBlocks.map(b => b.body.txEffects.map(tx => tx.l2ToL1Msgs)),
+        );
         previousCheckpointOutHashes = [...previousCheckpointOutHashes, checkpointOutHash];
       }
     } finally {
       await fork.close();
     }
   }
 
-  public async validateBlock(
-    blockFromL1: L2Block,
+  private async validateCheckpoint(
+    checkpointBlocks: L2Block[],
     previousCheckpointOutHashes: Fr[],
     fork: MerkleTreeWriteOperations,
   ): Promise<void> {
-    this.log.debug(`Validating pruned block ${blockFromL1.header.globalVariables.blockNumber}`);
-    const txHashes = blockFromL1.body.txEffects.map(txEffect => txEffect.txHash);
-    // We load txs from the mempool directly, since the TxCollector running in the background has already been
-    // trying to fetch them from nodes or via reqresp. If we haven't managed to collect them by now,
-    // it's likely that they are not available in the network at all.
-    const { txs, missingTxs } = await this.txProvider.getAvailableTxs(txHashes);
-
-    if (missingTxs && missingTxs.length > 0) {
-      throw new TransactionsNotAvailableError(missingTxs);
-    }
+    const checkpointNumber = checkpointBlocks[0].checkpointNumber;
+    this.log.debug(`Validating pruned checkpoint ${checkpointNumber} with ${checkpointBlocks.length} blocks`);
 
-    const checkpointNumber = CheckpointNumber.fromBlockNumber(blockFromL1.number);
+    // Get L1ToL2Messages once for the entire checkpoint
     const l1ToL2Messages = await this.l1ToL2MessageSource.getL1ToL2Messages(checkpointNumber);
-    const gv = blockFromL1.header.globalVariables;
+
+    // Build checkpoint constants from first block's global variables
+    const gv = checkpointBlocks[0].header.globalVariables;
     const constants: CheckpointGlobalVariables = {
       chainId: gv.chainId,
       version: gv.version,
@@ -169,7 +177,7 @@ export class EpochPruneWatcher extends (EventEmitter as new () => WatcherEmitter
       gasFees: gv.gasFees,
     };
 
-    // Use checkpoint builder to validate the block
+    // Start checkpoint builder once for all blocks in this checkpoint
     const checkpointBuilder = await this.checkpointsBuilder.startCheckpoint(
       checkpointNumber,
       constants,
@@ -179,6 +187,28 @@ export class EpochPruneWatcher extends (EventEmitter as new () => WatcherEmitter
       this.log.getBindings(),
     );
 
+    // Validate all blocks in the checkpoint sequentially
+    for (const block of checkpointBlocks) {
+      await this.validateBlockInCheckpoint(block, checkpointBuilder);
+    }
+  }
+
+  private async validateBlockInCheckpoint(
+    blockFromL1: L2Block,
+    checkpointBuilder: ICheckpointBlockBuilder,
+  ): Promise<void> {
+    this.log.debug(`Validating pruned block ${blockFromL1.header.globalVariables.blockNumber}`);
+    const txHashes = blockFromL1.body.txEffects.map(txEffect => txEffect.txHash);
+    // We load txs from the mempool directly, since the TxCollector running in the background has already been
+    // trying to fetch them from nodes or via reqresp. If we haven't managed to collect them by now,
+    // it's likely that they are not available in the network at all.
+    const { txs, missingTxs } = await this.txProvider.getAvailableTxs(txHashes);
+
+    if (missingTxs && missingTxs.length > 0) {
+      throw new TransactionsNotAvailableError(missingTxs);
+    }
+
+    const gv = blockFromL1.header.globalVariables;
     const { block, failedTxs, numTxs } = await checkpointBuilder.buildBlock(txs, gv.blockNumber, gv.timestamp, {});
 
     if (numTxs !== txs.length) {