feat: merge-train/spartan

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

@@ -432,4 +432,116 @@ describe('Archiver Store', () => {
       expect(result).toEqual([]);
     });
   });
+
+  describe('rollbackTo', () => {
+    beforeEach(() => {
+      publicClient.getBlock.mockImplementation(
+        (args: { blockNumber?: bigint } = {}) =>
+          Promise.resolve({ number: args.blockNumber ?? 0n, hash: `0x${'0'.repeat(64)}` }) as any,
+      );
+    });
+
+    it('rejects rollback to a block that is not at a checkpoint boundary', async () => {
+      const genesisArchive = new AppendOnlyTreeSnapshot(new Fr(GENESIS_ARCHIVE_ROOT), 1);
+      // Checkpoint 1: 3 blocks (1, 2, 3). Checkpoint 2: 3 blocks (4, 5, 6).
+      const testCheckpoints = await makeChainedCheckpoints(2, {
+        previousArchive: genesisArchive,
+        blocksPerCheckpoint: 3,
+      });
+      await archiverStore.addCheckpoints(testCheckpoints);
+
+      // Block 1 is not at a checkpoint boundary (checkpoint 1 ends at block 3)
+      await expect(archiver.rollbackTo(BlockNumber(1))).rejects.toThrow(
+        /not at a checkpoint boundary.*Use block 3 to roll back to this checkpoint.*or block 0 to roll back to the previous one/,
+      );
+
+      // Block 2 is also not at a checkpoint boundary
+      await expect(archiver.rollbackTo(BlockNumber(2))).rejects.toThrow(
+        /not at a checkpoint boundary.*Use block 3 to roll back to this checkpoint.*or block 0 to roll back to the previous one/,
+      );
+    });
+
+    it('allows rollback to the last block of a checkpoint and updates sync points', async () => {
+      const genesisArchive = new AppendOnlyTreeSnapshot(new Fr(GENESIS_ARCHIVE_ROOT), 1);
+      // Checkpoint 1: 3 blocks (1, 2, 3), L1 block 10. Checkpoint 2: 3 blocks (4, 5, 6), L1 block 20.
+      const testCheckpoints = await makeChainedCheckpoints(2, {
+        previousArchive: genesisArchive,
+        blocksPerCheckpoint: 3,
+      });
+      await archiverStore.addCheckpoints(testCheckpoints);
+
+      // Block 3 is the last block of checkpoint 1 — should succeed
+      await archiver.rollbackTo(BlockNumber(3));
+
+      expect(await archiver.getSynchedCheckpointNumber()).toEqual(CheckpointNumber(1));
+
+      // Verify sync points are set to checkpoint 1's L1 block number (10)
+      const synchPoint = await archiverStore.getSynchPoint();
+      expect(synchPoint.blocksSynchedTo).toEqual(10n);
+      expect(synchPoint.messagesSynchedTo?.l1BlockNumber).toEqual(10n);
+    });
+
+    it('includes correct boundary info in error for mid-checkpoint rollback', async () => {
+      const genesisArchive = new AppendOnlyTreeSnapshot(new Fr(GENESIS_ARCHIVE_ROOT), 1);
+      // Checkpoint 1: 2 blocks (1, 2). Checkpoint 2: 3 blocks (3, 4, 5).
+      const checkpoints1 = await makeChainedCheckpoints(1, {
+        previousArchive: genesisArchive,
+        blocksPerCheckpoint: 2,
+      });
+      const checkpoints2 = await makeChainedCheckpoints(1, {
+        previousArchive: checkpoints1[0].checkpoint.blocks.at(-1)!.archive,
+        startCheckpointNumber: CheckpointNumber(2),
+        startBlockNumber: 3,
+        startL1BlockNumber: 20,
+        blocksPerCheckpoint: 3,
+      });
+      await archiverStore.addCheckpoints([...checkpoints1, ...checkpoints2]);
+
+      // Block 3 is the first of checkpoint 2 (spans 3-5)
+      // Should suggest block 5 (end of this checkpoint) or block 2 (end of previous)
+      await expect(archiver.rollbackTo(BlockNumber(3))).rejects.toThrow(
+        /Checkpoint 2 spans blocks 3 to 5.*Use block 5 to roll back to this checkpoint.*or block 2 to roll back to the previous one/,
+      );
+    });
+
+    it('rolls back proven checkpoint number when target is before proven block', async () => {
+      const genesisArchive = new AppendOnlyTreeSnapshot(new Fr(GENESIS_ARCHIVE_ROOT), 1);
+      // Checkpoint 1: blocks 1-2, Checkpoint 2: blocks 3-4, Checkpoint 3: blocks 5-6
+      const testCheckpoints = await makeChainedCheckpoints(3, {
+        previousArchive: genesisArchive,
+        blocksPerCheckpoint: 2,
+      });
+      await archiverStore.addCheckpoints(testCheckpoints);
+
+      // Mark checkpoint 2 as proven
+      await archiverStore.setProvenCheckpointNumber(CheckpointNumber(2));
+      expect(await archiver.getProvenCheckpointNumber()).toEqual(CheckpointNumber(2));
+
+      // Roll back to block 2 (end of checkpoint 1), which is before proven block 4
+      await archiver.rollbackTo(BlockNumber(2));
+
+      expect(await archiver.getSynchedCheckpointNumber()).toEqual(CheckpointNumber(1));
+      expect(await archiver.getProvenCheckpointNumber()).toEqual(CheckpointNumber(1));
+    });
+
+    it('preserves proven checkpoint number when target is after proven block', async () => {
+      const genesisArchive = new AppendOnlyTreeSnapshot(new Fr(GENESIS_ARCHIVE_ROOT), 1);
+      // Checkpoint 1: blocks 1-2, Checkpoint 2: blocks 3-4, Checkpoint 3: blocks 5-6
+      const testCheckpoints = await makeChainedCheckpoints(3, {
+        previousArchive: genesisArchive,
+        blocksPerCheckpoint: 2,
+      });
+      await archiverStore.addCheckpoints(testCheckpoints);
+
+      // Mark checkpoint 1 as proven
+      await archiverStore.setProvenCheckpointNumber(CheckpointNumber(1));
+      expect(await archiver.getProvenCheckpointNumber()).toEqual(CheckpointNumber(1));
+
+      // Roll back to block 4 (end of checkpoint 2), which is after proven block 2
+      await archiver.rollbackTo(BlockNumber(4));
+
+      expect(await archiver.getSynchedCheckpointNumber()).toEqual(CheckpointNumber(2));
+      expect(await archiver.getProvenCheckpointNumber()).toEqual(CheckpointNumber(1));
+    });
+  });
 });

yarn-project/archiver/src/archiver.ts

@@ -399,7 +399,6 @@ export class Archiver extends ArchiverDataSourceBase implements L2BlockSink, Tra
   }
 
   public async rollbackTo(targetL2BlockNumber: BlockNumber): Promise<void> {
-    // TODO(pw/mbps): This still assumes 1 block per checkpoint
     const currentBlocks = await this.getL2Tips();
     const currentL2Block = currentBlocks.proposed.number;
     const currentProvenBlock = currentBlocks.proven.block.number;
@@ -411,8 +410,25 @@ export class Archiver extends ArchiverDataSourceBase implements L2BlockSink, Tra
     if (!targetL2Block) {
       throw new Error(`Target L2 block ${targetL2BlockNumber} not found`);
     }
-    const targetL1BlockNumber = targetL2Block.l1.blockNumber;
     const targetCheckpointNumber = targetL2Block.checkpointNumber;
+
+    // Rollback operates at checkpoint granularity: the target block must be the last block of its checkpoint.
+    const checkpointData = await this.store.getCheckpointData(targetCheckpointNumber);
+    if (checkpointData) {
+      const lastBlockInCheckpoint = BlockNumber(checkpointData.startBlock + checkpointData.blockCount - 1);
+      if (targetL2BlockNumber !== lastBlockInCheckpoint) {
+        const previousCheckpointBoundary =
+          checkpointData.startBlock > 1 ? BlockNumber(checkpointData.startBlock - 1) : BlockNumber(0);
+        throw new Error(
+          `Target L2 block ${targetL2BlockNumber} is not at a checkpoint boundary. ` +
+            `Checkpoint ${targetCheckpointNumber} spans blocks ${checkpointData.startBlock} to ${lastBlockInCheckpoint}. ` +
+            `Use block ${lastBlockInCheckpoint} to roll back to this checkpoint, ` +
+            `or block ${previousCheckpointBoundary} to roll back to the previous one.`,
+        );
+      }
+    }
+
+    const targetL1BlockNumber = targetL2Block.l1.blockNumber;
     const targetL1Block = await this.publicClient.getBlock({
       blockNumber: targetL1BlockNumber,
       includeTransactions: false,
@@ -431,13 +447,14 @@ export class Archiver extends ArchiverDataSourceBase implements L2BlockSink, Tra
     await this.store.setCheckpointSynchedL1BlockNumber(targetL1BlockNumber);
     await this.store.setMessageSynchedL1Block({ l1BlockNumber: targetL1BlockNumber, l1BlockHash: targetL1BlockHash });
     if (targetL2BlockNumber < currentProvenBlock) {
-      this.log.info(`Clearing proven L2 block number`);
-      await this.updater.setProvenCheckpointNumber(CheckpointNumber.ZERO);
+      this.log.info(`Rolling back proven L2 checkpoint to ${targetCheckpointNumber}`);
+      await this.updater.setProvenCheckpointNumber(targetCheckpointNumber);
     }
     // TODO(palla/reorg): Set the finalized block when we add support for it.
+    // const currentFinalizedBlock = currentBlocks.finalized.block.number;
     // if (targetL2BlockNumber < currentFinalizedBlock) {
-    //   this.log.info(`Clearing finalized L2 block number`);
-    //   await this.store.setFinalizedL2BlockNumber(0);
+    //   this.log.info(`Rolling back finalized L2 checkpoint to ${targetCheckpointNumber}`);
+    //   await this.updater.setFinalizedCheckpointNumber(targetCheckpointNumber);
     // }
   }
 }

yarn-project/end-to-end/src/spartan/n_tps.test.ts

@@ -90,10 +90,10 @@ const mempoolTxMinedDelayQuery = (perc: string) =>
 const mempoolAttestationMinedDelayQuery = (perc: string) =>
   `histogram_quantile(${perc}, sum(rate(aztec_mempool_attestations_mined_delay_milliseconds_bucket{k8s_namespace_name="${config.NAMESPACE}"}[1m])) by (le))`;
 
-const peerCountQuery = () => `avg(aztec_peer_manager_peer_count{k8s_namespace_name="${config.NAMESPACE}"})`;
+const peerCountQuery = () => `avg(aztec_peer_manager_peer_count_peers{k8s_namespace_name="${config.NAMESPACE}"})`;
 
-const peerConnectionDurationQuery = (perc: string) =>
-  `histogram_quantile(${perc}, sum(rate(aztec_peer_manager_peer_connection_duration_milliseconds_bucket{k8s_namespace_name="${config.NAMESPACE}"}[1m])) by (le))`;
+const peerConnectionDurationQuery = (perc: string, windowSeconds: number) =>
+  `histogram_quantile(${perc}, sum(rate(aztec_peer_manager_peer_connection_duration_milliseconds_bucket{k8s_namespace_name="${config.NAMESPACE}"}[${windowSeconds}s])) by (le))`;
 
 describe('sustained N TPS test', () => {
   jest.setTimeout(60 * 60 * 1000 * 10); // 10 hours
@@ -168,8 +168,8 @@ describe('sustained N TPS test', () => {
       try {
         const [avgCount, durationP50, durationP95] = await Promise.all([
           prometheusClient.querySingleValue(peerCountQuery()),
-          prometheusClient.querySingleValue(peerConnectionDurationQuery('0.50')),
-          prometheusClient.querySingleValue(peerConnectionDurationQuery('0.95')),
+          prometheusClient.querySingleValue(peerConnectionDurationQuery('0.50', TEST_DURATION_SECONDS + 60)),
+          prometheusClient.querySingleValue(peerConnectionDurationQuery('0.95', TEST_DURATION_SECONDS + 60)),
         ]);
         metrics.recordPeerStats(avgCount, durationP50, durationP95);
         logger.debug('Scraped peer stats', { avgCount, durationP50, durationP95 });
@@ -384,7 +384,7 @@ describe('sustained N TPS test', () => {
       const tx = await (config.REAL_VERIFIER ? submitProven(wallet, fee) : submitUnproven(wallet, fee));
       const t1 = performance.now();
 
-      metrics.recordSentTx(tx, `high_value_${highValueTps}tps`);
+      metrics.recordSentTx(tx, 'tx_inclusion_time');
 
       const txHash = await tx.send({ wait: NO_WAIT });
       const t2 = performance.now();
@@ -461,8 +461,8 @@ describe('sustained N TPS test', () => {
         logger.warn(`Failed transaction ${idx + 1}: ${result.error}`);
       });
 
-    const highValueGroup = `high_value_${highValueTps}tps`;
-    const inclusionStats = metrics.inclusionTimeInSeconds(highValueGroup);
+    const txInclusionGroup = 'tx_inclusion_time';
+    const inclusionStats = metrics.inclusionTimeInSeconds(txInclusionGroup);
     logger.info(`Transaction inclusion summary: ${successCount} succeeded, ${failureCount} failed`);
     logger.info('Inclusion time stats', inclusionStats);
   });

yarn-project/end-to-end/src/spartan/tx_metrics.ts

@@ -296,7 +296,7 @@ export class TxInclusionMetrics {
           value: stats.mean,
         },
         {
-          name: `${group}/median_inclusion`,
+          name: `${group}/p50_inclusion`,
           unit: 's',
           value: stats.median,
         },

yarn-project/ethereum/src/l1_tx_utils/l1_tx_utils.test.ts

@@ -147,7 +147,7 @@ describe('L1TxUtils', () => {
         address: l1Client.account.address,
       });
 
-      // Next send fails at sendRawTransaction (e.g. network error)
+      // Next send fails at sendRawTransaction (e.g. network error / 429)
       const originalSendRawTransaction = l1Client.sendRawTransaction.bind(l1Client);
       using _sendSpy = jest
         .spyOn(l1Client, 'sendRawTransaction')
@@ -163,6 +163,29 @@ describe('L1TxUtils', () => {
       expect((await l1Client.getTransaction({ hash: txHash })).nonce).toBe(expectedNonce);
     }, 30_000);
 
+    it('bumps nonce when getTransactionCount returns a stale value after a successful send', async () => {
+      // Send a successful tx first to advance the chain nonce
+      await gasUtils.sendAndMonitorTransaction(request);
+
+      const expectedNonce = await l1Client.getTransactionCount({
+        blockTag: 'pending',
+        address: l1Client.account.address,
+      });
+
+      // Simulate a stale fallback RPC node that returns the pre-send nonce
+      const originalGetTransactionCount = l1Client.getTransactionCount.bind(l1Client);
+      using _spy = jest
+        .spyOn(l1Client, 'getTransactionCount')
+        .mockImplementationOnce(() => Promise.resolve(expectedNonce - 1)) // stale: one behind
+        .mockImplementation(originalGetTransactionCount);
+
+      // Despite the stale count, the send should use lastSentNonce+1 = expectedNonce
+      const { txHash, state } = await gasUtils.sendTransaction(request);
+
+      expect(state.nonce).toBe(expectedNonce);
+      expect((await l1Client.getTransaction({ hash: txHash })).nonce).toBe(expectedNonce);
+    }, 30_000);
+
     // Regression for TMNT-312
     it('speed-up of blob tx sets non-zero maxFeePerBlobGas', async () => {
       await cheatCodes.setAutomine(false);

yarn-project/ethereum/src/l1_tx_utils/l1_tx_utils.ts

@@ -45,6 +45,8 @@ const MAX_L1_TX_STATES = 32;
 
 export class L1TxUtils extends ReadOnlyL1TxUtils {
   protected txs: L1TxState[] = [];
+  /** Last nonce successfully sent to the chain. Used as a lower bound when a fallback RPC node returns a stale count. */
+  private lastSentNonce: number | undefined;
   /** Tx delayer for testing. Only set when enableDelayer config is true. */
   public delayer?: Delayer;
   /** KZG instance for blob operations. */
@@ -105,6 +107,11 @@ export class L1TxUtils extends ReadOnlyL1TxUtils {
       this.metrics?.recordMinedTx(l1TxState, new Date(l1Timestamp));
     } else if (newState === TxUtilsState.NOT_MINED) {
       this.metrics?.recordDroppedTx(l1TxState);
+      // The tx was dropped: the chain nonce reverted to l1TxState.nonce, so our lower bound is
+      // no longer valid. Clear it so the next send fetches the real nonce from the chain.
+      if (this.lastSentNonce === l1TxState.nonce) {
+        this.lastSentNonce = undefined;
+      }
     }
 
     // Update state in the store
@@ -246,14 +253,20 @@ export class L1TxUtils extends ReadOnlyL1TxUtils {
         );
       }
 
-      const nonce = await this.client.getTransactionCount({ address: account, blockTag: 'pending' });
+      const chainNonce = await this.client.getTransactionCount({ address: account, blockTag: 'pending' });
+      // If a fallback RPC node returns a stale count (lower than what we last sent), use our
+      // local lower bound to avoid sending a duplicate of an already-pending transaction.
+      const nonce =
+        this.lastSentNonce !== undefined && chainNonce <= this.lastSentNonce ? this.lastSentNonce + 1 : chainNonce;
 
       const baseState = { request, gasLimit, blobInputs, gasPrice, nonce };
       const txData = this.makeTxData(baseState, { isCancelTx: false });
 
       // Send the new tx
       const signedRequest = await this.prepareSignedTransaction(txData);
       const txHash = await this.client.sendRawTransaction({ serializedTransaction: signedRequest });
+      // Update after tx is sent successfully
+      this.lastSentNonce = nonce;
 
       // Create the new state for monitoring
       const l1TxState: L1TxState = {

yarn-project/p2p/src/config.ts

@@ -38,7 +38,7 @@ export interface P2PConfig
     ChainConfig,
     TxCollectionConfig,
     TxFileStoreConfig,
-    Pick<SequencerConfig, 'blockDurationMs' | 'expectedBlockProposalsPerSlot'> {
+    Pick<SequencerConfig, 'blockDurationMs' | 'expectedBlockProposalsPerSlot' | 'maxTxsPerBlock'> {
   /** A flag dictating whether the P2P subsystem should be enabled. */
   p2pEnabled: boolean;
 

yarn-project/p2p/src/mem_pools/attestation_pool/attestation_pool.ts

@@ -359,11 +359,10 @@ export class AttestationPool {
         }
 
         const address = sender.toString();
+        const ownKey = this.getAttestationKey(slotNumber, proposalId, address);
 
-        await this.checkpointAttestations.set(
-          this.getAttestationKey(slotNumber, proposalId, address),
-          attestation.toBuffer(),
-        );
+        await this.checkpointAttestations.set(ownKey, attestation.toBuffer());
+        this.metrics.trackMempoolItemAdded(ownKey);
 
         this.log.debug(`Added own checkpoint attestation for slot ${slotNumber} from ${address}`, {
           signature: attestation.signature.toString(),
@@ -429,6 +428,7 @@ export class AttestationPool {
       const attestationEndKey = new Fr(oldestSlot).toString();
       for await (const key of this.checkpointAttestations.keysAsync({ end: attestationEndKey })) {
         await this.checkpointAttestations.delete(key);
+        this.metrics.trackMempoolItemRemoved(key);
         numberOfAttestations++;
       }
 
@@ -526,6 +526,7 @@ export class AttestationPool {
 
       // Add the attestation
       await this.checkpointAttestations.set(key, attestation.toBuffer());
+      this.metrics.trackMempoolItemAdded(key);
 
       // Track this attestation in the per-signer-per-slot index for duplicate detection
       const slotSignerKey = this.getSlotSignerKey(slotNumber, signerAddress);

yarn-project/p2p/src/mem_pools/instrumentation.ts

@@ -73,7 +73,7 @@ export class PoolInstrumentation<PoolObject extends Gossipable> {
   private defaultAttributes;
   private meter: Meter;
 
-  private txAddedTimestamp: Map<bigint, number> = new Map<bigint, number>();
+  private mempoolItemAddedTimestamp: Map<bigint | string, number> = new Map<bigint | string, number>();
 
   constructor(
     telemetry: TelemetryClient,
@@ -114,22 +114,26 @@ export class PoolInstrumentation<PoolObject extends Gossipable> {
   }
 
   public transactionsAdded(transactions: Tx[]) {
-    const timestamp = Date.now();
-    for (const transaction of transactions) {
-      this.txAddedTimestamp.set(transaction.txHash.toBigInt(), timestamp);
-    }
+    transactions.forEach(tx => this.trackMempoolItemAdded(tx.txHash.toBigInt()));
   }
 
   public transactionsRemoved(hashes: Iterable<bigint> | Iterable<string>) {
-    const timestamp = Date.now();
     for (const hash of hashes) {
-      const key = BigInt(hash);
-      const addedAt = this.txAddedTimestamp.get(key);
-      if (addedAt !== undefined) {
-        this.txAddedTimestamp.delete(key);
-        if (addedAt < timestamp) {
-          this.minedDelay.record(timestamp - addedAt);
-        }
+      this.trackMempoolItemRemoved(BigInt(hash));
+    }
+  }
+
+  public trackMempoolItemAdded(key: bigint | string): void {
+    this.mempoolItemAddedTimestamp.set(key, Date.now());
+  }
+
+  public trackMempoolItemRemoved(key: bigint | string): void {
+    const timestamp = Date.now();
+    const addedAt = this.mempoolItemAddedTimestamp.get(key);
+    if (addedAt !== undefined) {
+      this.mempoolItemAddedTimestamp.delete(key);
+      if (addedAt < timestamp) {
+        this.minedDelay.record(timestamp - addedAt);
       }
     }
   }