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

Author: Maddiaa0

yarn-project/archiver/src/archiver.ts

@@ -36,9 +36,12 @@ import { ArchiverDataSourceBase } from './modules/data_source_base.js';
 import { ArchiverDataStoreUpdater } from './modules/data_store_updater.js';
 import type { ArchiverInstrumentation } from './modules/instrumentation.js';
 import type { ArchiverL1Synchronizer } from './modules/l1_synchronizer.js';
+import type { PendingCheckpointStorage } from './store/block_store.js';
 import type { KVArchiverDataStore } from './store/kv_archiver_store.js';
 import { L2TipsCache } from './store/l2_tips_cache.js';
 
+export type { PendingCheckpointStorage };
+
 /** Export ArchiverEmitter for use in factory and tests. */
 export type { ArchiverEmitter };
 
@@ -209,6 +212,10 @@ export class Archiver extends ArchiverDataSourceBase implements L2BlockSink, Tra
     });
   }
 
+  public async setPendingCheckpoint(pending: PendingCheckpointStorage): Promise<void> {
+    await this.dataStore.blockStore.setPendingCheckpoint(pending);
+  }
+
   /**
    * 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

@@ -18,6 +18,7 @@ import type { BlockHeader, IndexedTxEffect, TxHash, TxReceipt } from '@aztec/std
 import type { UInt64 } from '@aztec/stdlib/types';
 
 import type { ArchiverDataSource } from '../interfaces.js';
+import type { PendingCheckpointStorage } from '../store/block_store.js';
 import type { KVArchiverDataStore } from '../store/kv_archiver_store.js';
 import type { ValidateCheckpointResult } from './validation.js';
 
@@ -173,6 +174,10 @@ export abstract class ArchiverDataSourceBase
     return this.store.getPrivateLogsByTags(tags, page, upToBlockNumber);
   }
 
+  public getPendingCheckpoint(): Promise<PendingCheckpointStorage | undefined> {
+    return this.store.blockStore.getPendingCheckpoint();
+  }
+
   public getPublicLogsByTagsFromContract(
     contractAddress: AztecAddress,
     tags: Tag[],

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

@@ -69,6 +69,12 @@ type CheckpointStorage = {
   attestations: Buffer[];
 };
 
+/** Quorum-attested checkpoint awaiting L1 confirmation. Stored as a single value for atomic reads/writes. */
+export type PendingCheckpointStorage = {
+  checkpointNumber: CheckpointNumber;
+  archive: Fr;
+};
+
 export type RemoveCheckpointsResult = { blocksRemoved: L2Block[] | undefined };
 
 /**
@@ -111,6 +117,8 @@ export class BlockStore {
   /** Index mapping block archive to block number */
   #blockArchiveIndex: AztecAsyncMap<string, number>;
 
+  #pendingCheckpoint: AztecAsyncSingleton<PendingCheckpointStorage>;
+
   #log = createLogger('archiver:block_store');
 
   constructor(private db: AztecAsyncKVStore) {
@@ -126,6 +134,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');
   }
 
   /**
@@ -161,6 +170,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 +189,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 +346,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 +450,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 +609,15 @@ export class BlockStore {
     return CheckpointNumber(latestCheckpointNumber);
   }
 
+  async getPendingCheckpoint(): Promise<PendingCheckpointStorage | undefined> {
+    return await this.#pendingCheckpoint.getAsync();
+  }
+
+  async getPendingCheckpointNumber(): Promise<CheckpointNumber> {
+    const pending = await this.getPendingCheckpoint();
+    return pending?.checkpointNumber ?? CheckpointNumber(INITIAL_CHECKPOINT_NUMBER - 1);
+  }
+
   async getCheckpointedBlock(number: BlockNumber): Promise<CheckpointedL2Block | undefined> {
     const blockStorage = await this.#blocks.getAsync(number);
     if (!blockStorage) {
@@ -950,6 +992,16 @@ export class BlockStore {
     return this.#lastSynchedL1Block.set(l1BlockNumber);
   }
 
+  /** Sets the pending checkpoint (quorum-attested but not yet L1-confirmed). Only updates if the new value is greater than the current one. */
+  async setPendingCheckpoint(pending: PendingCheckpointStorage) {
+    const current = await this.getPendingCheckpointNumber();
+    if (pending.checkpointNumber <= current) {
+      this.#log.warn(`Ignoring stale pending checkpoint number ${pending.checkpointNumber} (current: ${current})`);
+      return;
+    }
+    await this.#pendingCheckpoint.set(pending);
+  }
+
   async getProvenCheckpointNumber(): Promise<CheckpointNumber> {
     const [latestCheckpointNumber, provenCheckpointNumber] = await Promise.all([
       this.getLatestCheckpointNumber(),