feat(pipeline): allow syncing blocks ontop of the proposed chain

yarn-project/archiver/src/archiver.ts

@@ -19,7 +19,7 @@ import {
   type L2Tips,
   type ValidateCheckpointResult,
 } from '@aztec/stdlib/block';
-import { PublishedCheckpoint } from '@aztec/stdlib/checkpoint';
+import { type PendingCheckpointData, PublishedCheckpoint } from '@aztec/stdlib/checkpoint';
 import {
   type L1RollupConstants,
   getEpochAtSlot,
@@ -209,6 +209,14 @@ export class Archiver extends ArchiverDataSourceBase implements L2BlockSink, Tra
     });
   }
 
+  public async setPendingCheckpoint(pending: PendingCheckpointData): Promise<void> {
+    await this.dataStore.blockStore.setPendingCheckpoint(pending);
+  }
+
+  public setPipeliningTreeInProgress(value: bigint): Promise<void> {
+    return this.store.setPipeliningTreeInProgress(value);
+  }
+
   /**
    * Processes all queued blocks, adding them to the store.
    * Called at the beginning of each sync iteration.

yarn-project/archiver/src/errors.ts

@@ -26,9 +26,13 @@ export class InitialCheckpointNumberNotSequentialError extends Error {
 }
 
 export class CheckpointNumberNotSequentialError extends Error {
-  constructor(newCheckpointNumber: number, previous: number | undefined) {
+  constructor(
+    newCheckpointNumber: number,
+    previous: number | undefined,
+    source: 'confirmed' | 'pending' = 'confirmed',
+  ) {
     super(
-      `Cannot insert new checkpoint ${newCheckpointNumber} given previous checkpoint number in batch is ${previous ?? 'undefined'}`,
+      `Cannot insert new checkpoint ${newCheckpointNumber} given previous ${source} checkpoint number is ${previous ?? 'undefined'}`,
     );
   }
 }

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

@@ -6,7 +6,12 @@ import { isDefined } from '@aztec/foundation/types';
 import type { FunctionSelector } from '@aztec/stdlib/abi';
 import type { AztecAddress } from '@aztec/stdlib/aztec-address';
 import { type BlockData, type BlockHash, CheckpointedL2Block, L2Block, type L2Tips } from '@aztec/stdlib/block';
-import { Checkpoint, type CheckpointData, PublishedCheckpoint } from '@aztec/stdlib/checkpoint';
+import {
+  Checkpoint,
+  type CheckpointData,
+  type PendingCheckpointData,
+  PublishedCheckpoint,
+} from '@aztec/stdlib/checkpoint';
 import type { ContractClassPublic, ContractDataSource, ContractInstanceWithAddress } from '@aztec/stdlib/contract';
 import { type L1RollupConstants, getSlotRangeForEpoch } from '@aztec/stdlib/epoch-helpers';
 import type { GetContractClassLogsResponse, GetPublicLogsResponse } from '@aztec/stdlib/interfaces/client';
@@ -157,6 +162,10 @@ export abstract class ArchiverDataSourceBase
     return this.store.getSettledTxReceipt(txHash, this.l1Constants);
   }
 
+  public getPendingCheckpoint(): Promise<PendingCheckpointData | undefined> {
+    return this.store.blockStore.getPendingCheckpoint();
+  }
+
   public isPendingChainInvalid(): Promise<boolean> {
     return this.getPendingChainValidationStatus().then(status => !status.valid);
   }

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

@@ -258,9 +258,10 @@ export class ArchiverL1Synchronizer implements Traceable {
 
   /** Prune all proposed local blocks that should have been checkpointed by now. */
   private async pruneUncheckpointedBlocks(currentL1Timestamp: bigint) {
-    const [lastCheckpointedBlockNumber, lastProposedBlockNumber] = await Promise.all([
+    const [lastCheckpointedBlockNumber, lastProposedBlockNumber, pendingCheckpoint] = await Promise.all([
       this.store.getCheckpointedL2BlockNumber(),
       this.store.getLatestBlockNumber(),
+      this.store.blockStore.getPendingCheckpoint(),
     ]);
 
     // If there are no uncheckpointed blocks, we got nothing to do
@@ -269,8 +270,17 @@ export class ArchiverL1Synchronizer implements Traceable {
       return;
     }
 
-    // What's the slot of the first uncheckpointed block?
+    // Don't prune blocks that are covered by a pending checkpoint (awaiting L1 submission from pipelining)
     const firstUncheckpointedBlockNumber = BlockNumber(lastCheckpointedBlockNumber + 1);
+    if (pendingCheckpoint) {
+      const lastPendingBlock = BlockNumber(pendingCheckpoint.startBlock + pendingCheckpoint.blockCount - 1);
+      if (lastPendingBlock >= firstUncheckpointedBlockNumber) {
+        this.log.trace(`Skipping prune: pending checkpoint covers blocks up to ${lastPendingBlock}`);
+        return;
+      }
+    }
+
+    // What's the slot of the first uncheckpointed block?
     const [firstUncheckpointedBlockHeader] = await this.store.getBlockHeaders(firstUncheckpointedBlockNumber, 1);
     const firstUncheckpointedBlockSlot = firstUncheckpointedBlockHeader?.getSlot();
 

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

@@ -9,7 +9,7 @@ import {
   getAttestationInfoFromPayload,
 } from '@aztec/stdlib/block';
 import type { PublishedCheckpoint } from '@aztec/stdlib/checkpoint';
-import { type L1RollupConstants, getEpochAtSlot } from '@aztec/stdlib/epoch-helpers';
+import { type L1RollupConstants, computeQuorum, getEpochAtSlot } from '@aztec/stdlib/epoch-helpers';
 import { ConsensusPayload } from '@aztec/stdlib/p2p';
 
 export type { ValidateCheckpointResult };
@@ -66,7 +66,7 @@ export async function validateCheckpointAttestations(
     return { valid: true };
   }
 
-  const requiredAttestationCount = Math.floor((committee.length * 2) / 3) + 1;
+  const requiredAttestationCount = computeQuorum(committee.length);
 
   const failedValidationResult = <TReason extends ValidateCheckpointNegativeResult['reason']>(reason: TReason) => ({
     valid: false as const,

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

@@ -19,7 +19,12 @@ import {
   deserializeValidateCheckpointResult,
   serializeValidateCheckpointResult,
 } from '@aztec/stdlib/block';
-import { type CheckpointData, L1PublishedData, PublishedCheckpoint } from '@aztec/stdlib/checkpoint';
+import {
+  type CheckpointData,
+  L1PublishedData,
+  type PendingCheckpointData,
+  PublishedCheckpoint,
+} from '@aztec/stdlib/checkpoint';
 import { type L1RollupConstants, getEpochAtSlot } from '@aztec/stdlib/epoch-helpers';
 import { CheckpointHeader } from '@aztec/stdlib/rollup';
 import { AppendOnlyTreeSnapshot } from '@aztec/stdlib/trees';
@@ -69,6 +74,16 @@ type CheckpointStorage = {
   attestations: Buffer[];
 };
 
+/** Storage format for a pending checkpoint (attested but not yet L1-confirmed). */
+type PendingCheckpointStore = {
+  header: Buffer;
+  checkpointNumber: number;
+  startBlock: number;
+  blockCount: number;
+  totalManaUsed: string;
+  feeAssetPriceModifier: string;
+};
+
 export type RemoveCheckpointsResult = { blocksRemoved: L2Block[] | undefined };
 
 /**
@@ -111,6 +126,8 @@ export class BlockStore {
   /** Index mapping block archive to block number */
   #blockArchiveIndex: AztecAsyncMap<string, number>;
 
+  #pendingCheckpoint: AztecAsyncSingleton<PendingCheckpointStore>;
+
   #log = createLogger('archiver:block_store');
 
   constructor(private db: AztecAsyncKVStore) {
@@ -126,6 +143,7 @@ export class BlockStore {
     this.#pendingChainValidationStatus = db.openSingleton('archiver_pending_chain_validation_status');
     this.#checkpoints = db.openMap('archiver_checkpoints');
     this.#slotToCheckpoint = db.openMap('archiver_slot_to_checkpoint');
+    this.#pendingCheckpoint = db.openSingleton('pending_checkpoint_data');
   }
 
   /**
@@ -161,6 +179,7 @@ export class BlockStore {
 
       // Extract the latest block and checkpoint numbers
       const previousBlockNumber = await this.getLatestBlockNumber();
+      const pendingCheckpointNumber = await this.getPendingCheckpointNumber();
       const previousCheckpointNumber = await this.getLatestCheckpointNumber();
 
       // Verify we're not overwriting checkpointed blocks
@@ -179,9 +198,19 @@ export class BlockStore {
         throw new BlockNumberNotSequentialError(blockNumber, previousBlockNumber);
       }
 
-      // The same check as above but for checkpoints
-      if (!opts.force && previousCheckpointNumber !== blockCheckpointNumber - 1) {
-        throw new CheckpointNumberNotSequentialError(blockCheckpointNumber, previousCheckpointNumber);
+      // The same check as above but for checkpoints. Accept the block if either the confirmed
+      // checkpoint or the pending (locally validated but not yet confirmed) checkpoint matches.
+      const expectedCheckpointNumber = blockCheckpointNumber - 1;
+      if (
+        !opts.force &&
+        previousCheckpointNumber !== expectedCheckpointNumber &&
+        pendingCheckpointNumber !== expectedCheckpointNumber
+      ) {
+        const [reported, source]: [CheckpointNumber, 'confirmed' | 'pending'] =
+          pendingCheckpointNumber > previousCheckpointNumber
+            ? [pendingCheckpointNumber, 'pending']
+            : [previousCheckpointNumber, 'confirmed'];
+        throw new CheckpointNumberNotSequentialError(blockCheckpointNumber, reported, source);
       }
 
       // Extract the previous block if there is one and see if it is for the same checkpoint or not
@@ -326,6 +355,13 @@ export class BlockStore {
         await this.#slotToCheckpoint.set(checkpoint.checkpoint.header.slotNumber, checkpoint.checkpoint.number);
       }
 
+      // Clear the pending checkpoint if any of the confirmed checkpoints match or supersede it
+      const pendingCheckpointNumber = await this.getPendingCheckpointNumber();
+      const lastConfirmedCheckpointNumber = checkpoints[checkpoints.length - 1].checkpoint.number;
+      if (pendingCheckpointNumber <= lastConfirmedCheckpointNumber) {
+        await this.#pendingCheckpoint.delete();
+      }
+
       await this.#lastSynchedL1Block.set(checkpoints[checkpoints.length - 1].l1.blockNumber);
       return true;
     });
@@ -423,6 +459,12 @@ export class BlockStore {
         this.#log.debug(`Removed checkpoint ${c}`);
       }
 
+      // Clear any pending checkpoint that was removed
+      const pendingCheckpointNumber = await this.getPendingCheckpointNumber();
+      if (pendingCheckpointNumber > checkpointNumber) {
+        await this.#pendingCheckpoint.delete();
+      }
+
       return { blocksRemoved };
     });
   }
@@ -576,6 +618,34 @@ export class BlockStore {
     return CheckpointNumber(latestCheckpointNumber);
   }
 
+  async getPendingCheckpoint(): Promise<PendingCheckpointData | undefined> {
+    const stored = await this.#pendingCheckpoint.getAsync();
+    if (!stored) {
+      return undefined;
+    }
+    return {
+      checkpointNumber: CheckpointNumber(stored.checkpointNumber),
+      header: CheckpointHeader.fromBuffer(stored.header),
+      startBlock: BlockNumber(stored.startBlock),
+      blockCount: stored.blockCount,
+      totalManaUsed: BigInt(stored.totalManaUsed ?? '0'),
+      feeAssetPriceModifier: BigInt(stored.feeAssetPriceModifier ?? '0'),
+    };
+  }
+
+  async getPendingCheckpointNumber(): Promise<CheckpointNumber> {
+    const pending = await this.getPendingCheckpoint();
+    return CheckpointNumber(pending?.checkpointNumber ?? INITIAL_CHECKPOINT_NUMBER - 1);
+  }
+
+  async getPendingCheckpointL2BlockNumber(): Promise<BlockNumber> {
+    const pending = await this.getPendingCheckpoint();
+    if (!pending) {
+      return BlockNumber(INITIAL_L2_BLOCK_NUM - 1);
+    }
+    return BlockNumber(pending.startBlock + pending.blockCount - 1);
+  }
+
   async getCheckpointedBlock(number: BlockNumber): Promise<CheckpointedL2Block | undefined> {
     const blockStorage = await this.#blocks.getAsync(number);
     if (!blockStorage) {
@@ -950,6 +1020,30 @@ export class BlockStore {
     return this.#lastSynchedL1Block.set(l1BlockNumber);
   }
 
+  /** Sets the pending checkpoint (quorum-attested but not yet L1-confirmed). Only accepts confirmed + 1. */
+  async setPendingCheckpoint(pending: PendingCheckpointData) {
+    const current = await this.getPendingCheckpointNumber();
+    if (pending.checkpointNumber <= current) {
+      this.#log.warn(`Ignoring stale pending checkpoint number ${pending.checkpointNumber} (current: ${current})`);
+      return;
+    }
+    const confirmed = await this.getLatestCheckpointNumber();
+    if (pending.checkpointNumber !== confirmed + 1) {
+      this.#log.warn(
+        `Ignoring pending checkpoint ${pending.checkpointNumber}: expected ${confirmed + 1} (confirmed + 1)`,
+      );
+      return;
+    }
+    await this.#pendingCheckpoint.set({
+      header: pending.header.toBuffer(),
+      checkpointNumber: pending.checkpointNumber,
+      startBlock: pending.startBlock,
+      blockCount: pending.blockCount,
+      totalManaUsed: pending.totalManaUsed.toString(),
+      feeAssetPriceModifier: pending.feeAssetPriceModifier.toString(),
+    });
+  }
+
   async getProvenCheckpointNumber(): Promise<CheckpointNumber> {
     const [latestCheckpointNumber, provenCheckpointNumber] = await Promise.all([
       this.getLatestCheckpointNumber(),