feat(consensus): add batched forward epoch sync

aptos-core/consensus/consensus-types/src/forward_epoch_sync.rs

@@ -0,0 +1,132 @@
+// Copyright © Aptos Foundation
+// SPDX-License-Identifier: Apache-2.0
+
+//! Versioned messages for forward, block-number anchored epoch synchronization.
+//!
+//! The block number is only a lookup/cursor hint. Block IDs, parent links, quorum certificates,
+//! and signed ledger infos remain the authenticated source of truth.
+
+use crate::{block::Block, quorum_cert::QuorumCert};
+use gaptos::{aptos_crypto::HashValue, aptos_types::ledger_info::LedgerInfoWithSignatures};
+use serde::{Deserialize, Serialize};
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+pub enum ForwardEpochSyncRequest {
+    V1(ForwardEpochSyncRequestV1),
+}
+
+impl ForwardEpochSyncRequest {
+    pub fn epoch(&self) -> u64 {
+        match self {
+            Self::V1(request) => request.epoch(),
+        }
+    }
+}
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+pub enum ForwardEpochSyncRequestV1 {
+    Prepare(ForwardEpochSyncPrepareRequest),
+    Fetch(ForwardEpochSyncFetchRequest),
+}
+
+impl ForwardEpochSyncRequestV1 {
+    pub fn epoch(&self) -> u64 {
+        match self {
+            Self::Prepare(request) => request.epoch,
+            Self::Fetch(request) => request.epoch,
+        }
+    }
+}
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+pub struct ForwardEpochSyncPrepareRequest {
+    pub epoch: u64,
+    pub anchor_block_number: u64,
+    pub anchor_block_id: HashValue,
+}
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+pub struct ForwardEpochSyncFetchRequest {
+    pub epoch: u64,
+    pub manifest_id: HashValue,
+    pub anchor_block_number: u64,
+    pub anchor_block_id: HashValue,
+    /// Last boundary durably replayed by the client. It can lag the fetch anchor.
+    pub replay_anchor_block_number: u64,
+    pub replay_anchor_block_id: HashValue,
+    pub batch_size_blocks: u64,
+}
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+pub enum ForwardEpochSyncResponse {
+    V1(ForwardEpochSyncResponseV1),
+}
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+pub enum ForwardEpochSyncResponseV1 {
+    Prepared(ForwardEpochSyncManifest),
+    Batch(ForwardEpochSyncBatch),
+    Error(ForwardEpochSyncError),
+}
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+pub struct ForwardEpochSyncManifest {
+    pub epoch: u64,
+    pub manifest_id: HashValue,
+    /// First block available for this epoch on the canonical path.
+    pub first_block_number: u64,
+    /// Tail block included in the epoch snapshot. This can be later than the committed target
+    /// when an epoch-change suffix is present.
+    pub terminal_block_number: u64,
+    pub terminal_block_id: HashValue,
+    /// Signed epoch-ending commit target.
+    pub target_block_number: u64,
+    pub target_block_id: HashValue,
+    pub target_ledger_info: LedgerInfoWithSignatures,
+}
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+pub struct ForwardEpochSyncRecord {
+    pub block: Block,
+    /// Execution block number, when this consensus block was ordered. A certifying suffix after
+    /// a non-blocking epoch boundary can legitimately have no execution block number.
+    pub block_number: Option<u64>,
+    pub randomness: Option<Vec<u8>>,
+    pub quorum_cert: QuorumCert,
+}
+
+#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
+pub enum ForwardEpochSyncBatchStatus {
+    More,
+    Complete,
+}
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+pub struct ForwardEpochSyncBatch {
+    pub epoch: u64,
+    pub manifest_id: HashValue,
+    /// Echo of the fetch cursor. The first returned block must be its direct child.
+    pub anchor_block_number: u64,
+    pub anchor_block_id: HashValue,
+    pub records: Vec<ForwardEpochSyncRecord>,
+    pub ledger_infos: Vec<LedgerInfoWithSignatures>,
+    /// Authenticated boundary that can be replayed after this batch is persisted. It can lag the
+    /// response tail because the tail QC commits an ancestor.
+    pub replay_target_block_number: u64,
+    pub replay_target_block_id: HashValue,
+    /// Cursor for the next fetch. This is always the response tail, not the replay target.
+    pub next_anchor_block_number: u64,
+    pub next_anchor_block_id: HashValue,
+    pub status: ForwardEpochSyncBatchStatus,
+}
+
+#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
+pub enum ForwardEpochSyncError {
+    EpochNotFound,
+    AnchorMismatch,
+    ManifestMismatch,
+    InvalidBatchSize,
+    BatchBoundaryNotFound,
+    Busy,
+    Internal,
+}

aptos-core/consensus/consensus-types/src/lib.rs

@@ -10,6 +10,7 @@ pub mod block_retrieval;
 pub mod common;
 pub mod delayed_qc_msg;
 pub mod epoch_retrieval;
+pub mod forward_epoch_sync;
 pub mod order_vote;
 pub mod order_vote_msg;
 pub mod order_vote_proposal;

aptos-core/consensus/src/block_storage/block_store.rs

@@ -137,6 +137,9 @@ pub struct BlockStore {
     /// Mapping from validator address to their index in the ordered validator set.
     /// Used during recovery to compute proposer_index for blocks.
     validator_indices: HashMap<AccountAddress, usize>,
+    /// Ephemeral, bounded metadata cache used by forward epoch sync. It is deliberately not
+    /// persisted: restart recovery is derived from the existing ConsensusDB schemas.
+    forward_epoch_sync_indexes: Mutex<HashMap<u64, Arc<sync_manager::ForwardEpochSyncIndex>>>,
 }
 
 impl BlockStore {
@@ -247,7 +250,7 @@ impl BlockStore {
     /// epoch change was in progress), recovery stops at the epoch change block's round.
     /// Suffix blocks beyond that point would never receive execution results from reth,
     /// so attempting to recover them would cause the pipeline to hang.
-    async fn recover_blocks(&self) {
+    async fn recover_blocks_checked(&self) -> anyhow::Result<()> {
         RECOVERY_GAUGE.set_with(&[], 1);
 
         let mut certs = self.inner.read().get_all_quorum_certs_with_commit_info();
@@ -287,16 +290,24 @@ impl BlockStore {
                 commit_round,
                 self.commit_root().round(),
             );
-            if let Err(e) = self
+            if let Err(error) = self
                 .send_for_execution(qc.into_wrapped_ledger_info(), true, epoch_change_block_number)
                 .await
             {
-                error!("recover_blocks: failed to commit blocks: {e}");
-                break;
+                RECOVERY_GAUGE.set_with(&[], 0);
+                return Err(error).context("recover_blocks: failed to commit blocks");
             }
         }
 
         RECOVERY_GAUGE.set_with(&[], 0);
+        Ok(())
+    }
+
+    async fn recover_blocks(&self) {
+        if let Err(error) = self.recover_blocks_checked().await {
+            RECOVERY_GAUGE.set_with(&[], 0);
+            error!(error = ?error, "Failed to recover consensus blocks");
+        }
     }
 
     pub(crate) async fn replay_ordered_path_if_needed(&self) -> anyhow::Result<()> {
@@ -452,6 +463,7 @@ impl BlockStore {
             enable_randomness,
             require_block_randomness,
             validator_indices,
+            forward_epoch_sync_indexes: Mutex::new(HashMap::new()),
         };
 
         // Skip ancestors of the root. They can appear in recovery data when an
@@ -871,6 +883,25 @@ impl BlockStore {
         self.recover_blocks().await;
     }
 
+    pub async fn append_blocks_for_sync_checked(
+        &self,
+        blocks: Vec<(Block, Option<u64>, Option<Vec<u8>>)>,
+        quorum_certs: Vec<QuorumCert>,
+    ) -> anyhow::Result<()> {
+        for (block, block_number, _) in blocks {
+            if let Some(num) = block_number {
+                if block.block_number().is_none() {
+                    block.set_block_number(num);
+                }
+            }
+            self.insert_block(block, true).await.context("Failed to append forward-sync block")?;
+        }
+        for qc in quorum_certs {
+            self.insert_single_quorum_cert(qc, true).context("Failed to append forward-sync QC")?;
+        }
+        self.recover_blocks_checked().await
+    }
+
     pub async fn rebuild(&self, root: RootInfo, blocks: Vec<Block>, quorum_certs: Vec<QuorumCert>) {
         info!(
             "Rebuilding block tree. root {:?}, blocks {:?}, qcs {:?}",

aptos-core/consensus/src/block_storage/sync_manager.rs

@@ -19,7 +19,7 @@ use crate::{
     payload_manager::TPayloadManager,
     persistent_liveness_storage::PersistentLivenessStorage,
 };
-use anyhow::{anyhow, bail};
+use anyhow::{anyhow, bail, ensure};
 use aptos_consensus_types::{
     block::Block,
     block_retrieval::{
@@ -61,6 +61,10 @@ use sha3::digest::generic_array::typenum::Le;
 use std::{clone::Clone, cmp::min, hash::Hash, sync::Arc, time::Duration};
 use tokio::{time, time::timeout};
 
+#[path = "sync_manager/forward_epoch_sync.rs"]
+mod forward_epoch_sync;
+pub(super) use forward_epoch_sync::ForwardEpochSyncIndex;
+
 static CUR_BLOCK_SYNC_BLOCK_SUM_GAUGE: Lazy<IntGaugeVec> = Lazy::new(|| {
     register_int_gauge_vec!(
         "aptos_current_block_sync_block_sum",
@@ -121,6 +125,33 @@ impl BlockStore {
         committed
     }
 
+    /// Replays the non-durable completion signal for an epoch whose blocks and ledger info are
+    /// already durable. This must be safe to call after restart: `send_epoch_change` only queues a
+    /// message to the local EpochManager, so a crash can lose that message after the epoch boundary
+    /// itself has been committed.
+    async fn send_committed_epoch_change(
+        &self,
+        retriever: &BlockRetriever,
+        ledger_info: &LedgerInfoWithSignatures,
+    ) -> anyhow::Result<()> {
+        ensure!(
+            ledger_info.ledger_info().ends_epoch(),
+            "Cannot complete epoch sync with a non-epoch-ending ledger info"
+        );
+        ensure!(
+            self.is_epoch_change_li_boundary_locally_committed(ledger_info),
+            "Epoch-change boundary is not locally committed"
+        );
+        retriever
+            .network
+            .send_epoch_change(EpochChangeProof::new(
+                vec![ledger_info.clone()],
+                /* more = */ false,
+            ))
+            .await;
+        Ok(())
+    }
+
     /// Check if we're far away from this ledger info and need to sync.
     /// This ensures that the block referred by the ledger info is not in buffer manager.
     pub fn need_sync_for_ledger_info(&self, li: &LedgerInfoWithSignatures) -> bool {
@@ -350,24 +381,7 @@ impl BlockStore {
         Ok(())
     }
 
-    /// Fast-forwards the local consensus state by synchronizing blocks and ledger infos for a given
-    /// epoch.
-    ///
-    /// This function retrieves all blocks, quorum certificates, and ledger infos for the specified
-    /// epoch from a remote retriever. It then prefetches payload data for each block, saves the
-    /// blocks and certificates to local storage, and updates the ledger info in the database.
-    /// After updating storage, it attempts to recover the consensus state from the latest
-    /// ledger info and rebuilds the in-memory state. If the epoch ends, it sends an epoch
-    /// change proof to the network.
-    ///
-    /// # Arguments
-    /// * `retriever` - The block retriever used to fetch blocks and related data.
-    /// * `epoch` - The epoch to fast-forward to.
-    ///
-    /// # Returns
-    /// * `Ok(())` if the synchronization and state rebuild succeed.
-    /// * `Err` if any step fails.
-    pub async fn fast_forward_sync_by_epoch(
+    async fn fast_forward_sync_by_epoch_legacy(
         &self,
         mut retriever: BlockRetriever,
         epoch: u64,
@@ -404,6 +418,17 @@ impl BlockStore {
                 "[Fast_Forward_sync] all fetched blocks at or below local HCC round {}, nothing to sync",
                 hcc_round
             );
+            let latest_li = ledger_infos
+                .iter()
+                .filter(|ledger_info| ledger_info.ledger_info().ends_epoch())
+                .max_by_key(|ledger_info| ledger_info.ledger_info().version())
+                .ok_or_else(|| {
+                    anyhow!(
+                        "No epoch-ending ledger info available to resume completed epoch {}",
+                        epoch
+                    )
+                })?;
+            self.send_committed_epoch_change(&retriever, latest_li).await?;
             return Ok(());
         }
 
@@ -463,16 +488,8 @@ impl BlockStore {
         self.rebuild(root, blocks, quorum_certs).await;
 
         let latest_li = ledger_infos.last().unwrap();
-        if latest_li.ledger_info().ends_epoch() &&
-            self.is_epoch_change_li_boundary_locally_committed(latest_li)
-        {
-            retriever
-                .network
-                .send_epoch_change(EpochChangeProof::new(
-                    vec![latest_li.clone()],
-                    /* more = */ false,
-                ))
-                .await;
+        if latest_li.ledger_info().ends_epoch() {
+            self.send_committed_epoch_change(&retriever, latest_li).await?;
         }
         Ok(())
     }