diff --git a/aptos-core/consensus/consensus-types/src/forward_epoch_sync.rs b/aptos-core/consensus/consensus-types/src/forward_epoch_sync.rs new file mode 100644 index 00000000..d11c57db --- /dev/null +++ b/aptos-core/consensus/consensus-types/src/forward_epoch_sync.rs @@ -0,0 +1,117 @@ +// 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, + 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(Box), + 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, + /// The block-number target for this epoch sync. Fetch batches are ordinary pages; the client + /// stops once this block has a verified ledger info and is durably committed. + 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, + pub randomness: Option>, + pub quorum_cert: QuorumCert, +} + +#[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, + /// Zero or more proofs whose certifying QC is present in this batch. A proof target may have + /// been persisted by an earlier batch; batch boundaries have no consensus meaning. + pub ledger_infos: Vec, + /// Cursor for the next ordinary page. This is always the response tail. + pub next_anchor_block_number: u64, + pub next_anchor_block_id: HashValue, +} + +#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub enum ForwardEpochSyncError { + EpochNotFound, + AnchorMismatch, + ManifestMismatch, + InvalidBatchSize, + BatchBoundaryNotFound, + Busy, + Internal, +} diff --git a/aptos-core/consensus/consensus-types/src/lib.rs b/aptos-core/consensus/consensus-types/src/lib.rs index 5470935c..96b4694c 100644 --- a/aptos-core/consensus/consensus-types/src/lib.rs +++ b/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; diff --git a/aptos-core/consensus/src/block_storage/block_store.rs b/aptos-core/consensus/src/block_storage/block_store.rs index 46952b64..92a49d94 100644 --- a/aptos-core/consensus/src/block_storage/block_store.rs +++ b/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, + /// 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>>, } 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, Option>)>, + quorum_certs: Vec, + ) -> 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, quorum_certs: Vec) { info!( "Rebuilding block tree. root {:?}, blocks {:?}, qcs {:?}", diff --git a/aptos-core/consensus/src/block_storage/sync_manager.rs b/aptos-core/consensus/src/block_storage/sync_manager.rs index 96941f98..2790655d 100644 --- a/aptos-core/consensus/src/block_storage/sync_manager.rs +++ b/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 = 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(()) } diff --git a/aptos-core/consensus/src/block_storage/sync_manager/forward_epoch_sync.rs b/aptos-core/consensus/src/block_storage/sync_manager/forward_epoch_sync.rs new file mode 100644 index 00000000..50779bba --- /dev/null +++ b/aptos-core/consensus/src/block_storage/sync_manager/forward_epoch_sync.rs @@ -0,0 +1,1081 @@ +// Copyright © Aptos Foundation +// SPDX-License-Identifier: Apache-2.0 + +//! Block-number anchored, forward epoch synchronization. +//! +//! This module owns the ephemeral server index, versioned RPC handling, authenticated client +//! verification, and batch persistence/replay. The legacy reverse-sync implementation remains in +//! the parent module as the rolling-upgrade fallback. + +use super::{BlockReader, BlockRetriever, BlockStore}; +use crate::{ + consensusdb::schema::{ + block::BlockNumberSchema, epoch_by_block_number::EpochByBlockNumberSchema, + ledger_info::LedgerInfoSchema, + }, + network::IncomingForwardEpochSyncRequest, + network_interface::ConsensusMsg, +}; +use anyhow::{anyhow, bail, ensure}; +use aptos_consensus_types::{ + block_retrieval::{NUM_RETRIES, RETRY_INTERVAL_MSEC, RPC_TIMEOUT_MSEC}, + forward_epoch_sync::{ + ForwardEpochSyncBatch, ForwardEpochSyncError, ForwardEpochSyncFetchRequest, + ForwardEpochSyncManifest, ForwardEpochSyncPrepareRequest, ForwardEpochSyncRecord, + ForwardEpochSyncRequest, ForwardEpochSyncRequestV1, ForwardEpochSyncResponse, + ForwardEpochSyncResponseV1, + }, +}; +use gaptos::{ + aptos_config::network_id::PeerNetworkId, + aptos_consensus::counters::BLOCKS_FETCHED_FROM_NETWORK_WHILE_FAST_FORWARD_SYNC, + aptos_crypto::{hash::CryptoHash, HashValue}, + aptos_logger::prelude::*, + aptos_schemadb::batch::SchemaBatch, + aptos_types::{account_address::AccountAddress, ledger_info::LedgerInfoWithSignatures}, +}; +use std::{ + collections::{HashMap, HashSet}, + sync::Arc, + time::Duration, +}; +use tokio::time; + +const FORWARD_EPOCH_SYNC_PREPARE_TIMEOUT_MSEC: u64 = 5_000; + +#[derive(Clone)] +struct ForwardEpochSyncIndexEntry { + block_number: Option, + /// Highest execution block number reached at this consensus block. Unnumbered certifying + /// suffix blocks retain the preceding value so they can still be used as resumable cursors. + anchor_block_number: u64, + block_id: HashValue, + parent_id: HashValue, +} + +#[derive(Clone)] +struct ForwardEpochSyncBoundary { + certifying_position: usize, + target_block_number: u64, + ledger_info: LedgerInfoWithSignatures, +} + +/// Immutable metadata snapshot for one epoch. Blocks, payloads, QCs, and randomness stay in the +/// existing databases and are loaded only for the requested batch. +pub(in crate::block_storage::block_store) struct ForwardEpochSyncIndex { + manifest: ForwardEpochSyncManifest, + entries: Vec, + positions: HashMap, + boundaries: Vec, +} + +fn select_forward_batch_end(start: usize, requested: usize, total: usize) -> Option { + let end = start.saturating_add(requested).min(total); + (end > start).then_some(end) +} + +fn certifying_position_in_batch(position: usize, start: usize, end: usize) -> bool { + position >= start && position < end +} + +/// A validated fetch cursor at the end of the server index has no page to return. Keep that +/// condition distinct from protocol and verification failures so the caller can wait for the +/// already-fetched epoch target to commit instead of failing the whole forward-sync attempt. +fn decode_forward_epoch_sync_fetch_response( + response: ForwardEpochSyncResponseV1, +) -> anyhow::Result> { + match response { + ForwardEpochSyncResponseV1::Batch(batch) => Ok(Some(batch)), + ForwardEpochSyncResponseV1::Error(ForwardEpochSyncError::BatchBoundaryNotFound) => Ok(None), + ForwardEpochSyncResponseV1::Error(error) => { + bail!("Forward epoch sync fetch rejected: {error:?}") + } + ForwardEpochSyncResponseV1::Prepared(_) => { + bail!("Forward epoch sync fetch returned a manifest") + } + } +} + +impl BlockStore { + fn build_forward_epoch_sync_index( + &self, + epoch: u64, + ) -> Result { + let db = self.storage.consensus_db(); + let epoch_end_block_number = db + .get_all::() + .map_err(|error| { + error!(epoch = epoch, error = ?error, "Failed to scan epoch boundaries for forward sync"); + ForwardEpochSyncError::Internal + })? + .into_iter() + .filter_map(|(block_number, stored_epoch)| { + (stored_epoch == epoch).then_some(block_number) + }) + .max() + .ok_or(ForwardEpochSyncError::EpochNotFound)?; + let target_ledger_info = db + .get::(&epoch_end_block_number) + .map_err(|error| { + error!(epoch = epoch, error = ?error, "Failed to read epoch-ending ledger info"); + ForwardEpochSyncError::Internal + })? + .ok_or(ForwardEpochSyncError::EpochNotFound)?; + + let start_key = (epoch, HashValue::zero()); + let end_key = (epoch, HashValue::new([u8::MAX; HashValue::LENGTH])); + let quorum_certs = db.get_qc_range(&start_key, &end_key).map_err(|error| { + error!(epoch = epoch, error = ?error, "Failed to read QCs for forward sync"); + ForwardEpochSyncError::Internal + })?; + let qcs_by_certified_id = quorum_certs + .into_iter() + .map(|qc| (qc.certified_block().id(), qc)) + .collect::>(); + let terminal_qc = qcs_by_certified_id + .values() + .filter(|qc| { + qc.commit_info().id() == target_ledger_info.ledger_info().consensus_block_id() + }) + .max_by_key(|qc| qc.certified_block().round()) + .ok_or_else(|| { + error!( + epoch = epoch, + target = %target_ledger_info.ledger_info().consensus_block_id(), + "Epoch-ending commit has no certifying QC" + ); + ForwardEpochSyncError::Internal + })?; + + let mut reverse_entries = Vec::new(); + let mut visited = HashSet::new(); + let mut cursor = terminal_qc.certified_block().id(); + loop { + if !visited.insert(cursor) { + error!(epoch = epoch, block_id = %cursor, "Cycle in persisted consensus block chain"); + return Err(ForwardEpochSyncError::Internal); + } + let block = db.get_block(epoch, cursor).map_err(|error| { + error!(epoch = epoch, block_id = %cursor, error = ?error, "Failed to read block"); + ForwardEpochSyncError::Internal + })?; + let Some(block) = block else { break }; + let block_number = match block.block_number() { + Some(block_number) => Some(block_number), + None => db.get::(&(epoch, block.id())).map_err(|error| { + error!( + epoch = epoch, + block_id = %block.id(), + error = ?error, + "Failed to read forward-sync block number" + ); + ForwardEpochSyncError::Internal + })?, + }; + if !qcs_by_certified_id.contains_key(&block.id()) { + error!(epoch = epoch, block_id = %block.id(), "Forward-sync block has no QC"); + return Err(ForwardEpochSyncError::Internal); + } + reverse_entries.push(ForwardEpochSyncIndexEntry { + block_number, + anchor_block_number: 0, + block_id: block.id(), + parent_id: block.parent_id(), + }); + cursor = block.parent_id(); + } + reverse_entries.reverse(); + if reverse_entries.is_empty() { + return Err(ForwardEpochSyncError::EpochNotFound); + } + for pair in reverse_entries.windows(2) { + if pair[1].parent_id != pair[0].block_id { + error!( + epoch = epoch, + parent_id = %pair[0].block_id, + child_id = %pair[1].block_id, + "Persisted epoch path is not contiguous" + ); + return Err(ForwardEpochSyncError::Internal); + } + } + let first_block_number = + reverse_entries.iter().find_map(|entry| entry.block_number).ok_or_else(|| { + error!(epoch = epoch, "Forward-sync epoch path has no numbered blocks"); + ForwardEpochSyncError::Internal + })?; + let mut anchor_block_number = first_block_number.checked_sub(1).ok_or_else(|| { + error!(epoch = epoch, "Forward-sync epoch path starts at block number zero"); + ForwardEpochSyncError::Internal + })?; + for entry in &mut reverse_entries { + if let Some(block_number) = entry.block_number { + if block_number != anchor_block_number.saturating_add(1) { + error!( + epoch = epoch, + block_id = %entry.block_id, + previous_number = anchor_block_number, + block_number = block_number, + "Persisted numbered epoch path is not contiguous" + ); + return Err(ForwardEpochSyncError::Internal); + } + anchor_block_number = block_number; + } + entry.anchor_block_number = anchor_block_number; + } + + let positions = reverse_entries + .iter() + .enumerate() + .map(|(position, entry)| (entry.block_id, position)) + .collect::>(); + let target_epoch_info = target_ledger_info.ledger_info().commit_info().epoch_block_info(); + let target_block_id = target_epoch_info + .map(|info| info.block_id) + .unwrap_or_else(|| target_ledger_info.ledger_info().consensus_block_id()); + let target_block_number = target_epoch_info + .map(|info| info.block_number) + .or_else(|| { + positions + .get(&target_block_id) + .and_then(|pos| reverse_entries[*pos].block_number) + }) + .ok_or_else(|| { + error!(epoch = epoch, target = %target_block_id, "Epoch target is not on canonical path"); + ForwardEpochSyncError::Internal + })?; + + let persisted_ledger_infos = db.get_all::().map_err(|error| { + error!(epoch = epoch, error = ?error, "Failed to scan ledger infos for forward sync"); + ForwardEpochSyncError::Internal + })?; + let mut boundaries = Vec::new(); + for (stored_block_number, ledger_info) in persisted_ledger_infos { + if ledger_info.ledger_info().epoch() != epoch { + continue; + } + let Some((_, certifying_position)) = qcs_by_certified_id + .values() + .filter(|qc| { + qc.commit_info().id() == ledger_info.ledger_info().consensus_block_id() + }) + .filter_map(|qc| { + positions + .get(&qc.certified_block().id()) + .copied() + .map(|position| (qc, position)) + }) + .min_by_key(|(qc, _)| qc.certified_block().round()) + else { + continue; + }; + let epoch_info = ledger_info.ledger_info().commit_info().epoch_block_info(); + let boundary_id = epoch_info + .map(|info| info.block_id) + .unwrap_or_else(|| ledger_info.ledger_info().consensus_block_id()); + let Some(target_position) = positions.get(&boundary_id).copied() else { + continue; + }; + if target_position > certifying_position { + continue; + } + let boundary_number = + epoch_info.map(|info| info.block_number).unwrap_or(stored_block_number); + boundaries.push(ForwardEpochSyncBoundary { + certifying_position, + target_block_number: boundary_number, + ledger_info, + }); + } + boundaries.sort_unstable_by_key(|boundary| { + (boundary.certifying_position, boundary.target_block_number) + }); + + let terminal = reverse_entries.last().expect("non-empty checked above"); + let manifest_bytes = bcs::to_bytes(&( + epoch, + first_block_number, + terminal.anchor_block_number, + terminal.block_id, + target_block_number, + target_block_id, + &target_ledger_info, + )) + .map_err(|error| { + error!(epoch = epoch, error = ?error, "Failed to hash forward-sync manifest"); + ForwardEpochSyncError::Internal + })?; + let manifest = ForwardEpochSyncManifest { + epoch, + manifest_id: HashValue::sha3_256_of(&manifest_bytes), + first_block_number, + target_block_number, + target_block_id, + target_ledger_info, + }; + Ok(ForwardEpochSyncIndex { manifest, entries: reverse_entries, positions, boundaries }) + } + + fn forward_epoch_sync_index( + &self, + epoch: u64, + ) -> Result, ForwardEpochSyncError> { + let mut indexes = self.forward_epoch_sync_indexes.lock(); + if let Some(index) = indexes.get(&epoch) { + return Ok(index.clone()); + } + let index = Arc::new(self.build_forward_epoch_sync_index(epoch)?); + // A BlockStore only needs to serve the epoch it currently owns. Bounding this map avoids + // retaining historical path metadata after unusual cross-epoch requests. + indexes.clear(); + indexes.insert(epoch, index.clone()); + Ok(index) + } + + fn validate_forward_anchor( + index: &ForwardEpochSyncIndex, + block_number: u64, + block_id: HashValue, + ) -> Result { + if let Some(position) = index.positions.get(&block_id).copied() { + return (index.entries[position].anchor_block_number == block_number) + .then_some(position.saturating_add(1)) + .ok_or(ForwardEpochSyncError::AnchorMismatch); + } + let first = index.entries.first().ok_or(ForwardEpochSyncError::EpochNotFound)?; + let first_follows_anchor = match first.block_number { + Some(first_number) => first_number == block_number.saturating_add(1), + None => first.anchor_block_number == block_number, + }; + if first.parent_id == block_id && first_follows_anchor { + Ok(0) + } else { + Err(ForwardEpochSyncError::AnchorMismatch) + } + } + + fn prepare_forward_epoch_sync( + &self, + request: ForwardEpochSyncPrepareRequest, + ) -> ForwardEpochSyncResponseV1 { + let index = match self.forward_epoch_sync_index(request.epoch) { + Ok(index) => index, + Err(error) => return ForwardEpochSyncResponseV1::Error(error), + }; + match Self::validate_forward_anchor( + &index, + request.anchor_block_number, + request.anchor_block_id, + ) { + Ok(_) => ForwardEpochSyncResponseV1::Prepared(Box::new(index.manifest.clone())), + Err(error) => ForwardEpochSyncResponseV1::Error(error), + } + } + + fn fetch_forward_epoch_sync( + &self, + request: ForwardEpochSyncFetchRequest, + max_blocks_allowed: u64, + ) -> ForwardEpochSyncResponseV1 { + if request.batch_size_blocks == 0 || request.batch_size_blocks > max_blocks_allowed { + return ForwardEpochSyncResponseV1::Error(ForwardEpochSyncError::InvalidBatchSize); + } + let index = match self.forward_epoch_sync_index(request.epoch) { + Ok(index) => index, + Err(error) => return ForwardEpochSyncResponseV1::Error(error), + }; + if request.manifest_id != index.manifest.manifest_id { + return ForwardEpochSyncResponseV1::Error(ForwardEpochSyncError::ManifestMismatch); + } + let start = match Self::validate_forward_anchor( + &index, + request.anchor_block_number, + request.anchor_block_id, + ) { + Ok(start) => start, + Err(error) => return ForwardEpochSyncResponseV1::Error(error), + }; + if start >= index.entries.len() { + return ForwardEpochSyncResponseV1::Error(ForwardEpochSyncError::BatchBoundaryNotFound); + } + let requested = usize::try_from(request.batch_size_blocks).unwrap_or(usize::MAX); + let Some(end) = select_forward_batch_end(start, requested, index.entries.len()) else { + return ForwardEpochSyncResponseV1::Error(ForwardEpochSyncError::BatchBoundaryNotFound); + }; + let ledger_infos = index + .boundaries + .iter() + .filter(|boundary| { + certifying_position_in_batch(boundary.certifying_position, start, end) + }) + .map(|boundary| boundary.ledger_info.clone()) + .collect::>(); + + let db = self.storage.consensus_db(); + let mut records = Vec::with_capacity(end - start); + for entry in &index.entries[start..end] { + let block = match db.get_block(request.epoch, entry.block_id) { + Ok(Some(block)) => block, + Ok(None) => { + return ForwardEpochSyncResponseV1::Error(ForwardEpochSyncError::Internal) + } + Err(error) => { + error!(epoch = request.epoch, block_id = %entry.block_id, error = ?error, "Failed to read forward-sync block"); + return ForwardEpochSyncResponseV1::Error(ForwardEpochSyncError::Internal); + } + }; + let quorum_cert = match db.get_qc(request.epoch, entry.block_id) { + Ok(Some(qc)) => qc, + Ok(None) => { + return ForwardEpochSyncResponseV1::Error(ForwardEpochSyncError::Internal) + } + Err(error) => { + error!(epoch = request.epoch, block_id = %entry.block_id, error = ?error, "Failed to read forward-sync QC"); + return ForwardEpochSyncResponseV1::Error(ForwardEpochSyncError::Internal); + } + }; + let randomness = match entry.block_number { + Some(block_number) => match db.get_randomness(block_number) { + Ok(randomness) => randomness, + Err(error) => { + error!(epoch = request.epoch, block_number = block_number, error = ?error, "Failed to read forward-sync randomness"); + return ForwardEpochSyncResponseV1::Error(ForwardEpochSyncError::Internal); + } + }, + None => None, + }; + records.push(ForwardEpochSyncRecord { + block, + block_number: entry.block_number, + randomness, + quorum_cert, + }); + } + let tail = index.entries.get(end - 1).expect("non-empty batch"); + ForwardEpochSyncResponseV1::Batch(ForwardEpochSyncBatch { + epoch: request.epoch, + manifest_id: request.manifest_id, + anchor_block_number: request.anchor_block_number, + anchor_block_id: request.anchor_block_id, + records, + ledger_infos, + next_anchor_block_number: tail.anchor_block_number, + next_anchor_block_id: tail.block_id, + }) + } + + pub async fn process_forward_epoch_sync( + &self, + request: IncomingForwardEpochSyncRequest, + max_blocks_allowed: u64, + ) -> anyhow::Result<()> { + let response = match request.req { + ForwardEpochSyncRequest::V1(ForwardEpochSyncRequestV1::Prepare(prepare)) => { + self.prepare_forward_epoch_sync(prepare) + } + ForwardEpochSyncRequest::V1(ForwardEpochSyncRequestV1::Fetch(fetch)) => { + self.fetch_forward_epoch_sync(fetch, max_blocks_allowed) + } + }; + let response = ConsensusMsg::ForwardEpochSyncResponse(Box::new( + ForwardEpochSyncResponse::V1(response), + )); + let response_bytes = request.protocol.to_bytes(&response)?; + request + .response_sender + .send(Ok(response_bytes.into())) + .map_err(|_| anyhow::anyhow!("Failed to send forward epoch sync response")) + } +} + +impl BlockStore { + /// 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( + &self, + mut retriever: BlockRetriever, + epoch: u64, + batch_size_blocks: u64, + ) -> anyhow::Result<()> { + if !crate::forward_epoch_sync_enabled() { + info!( + epoch = epoch, + "Forward epoch sync is not enabled; using legacy reverse epoch sync" + ); + return self.fast_forward_sync_by_epoch_legacy(retriever, epoch).await; + } + ensure!(batch_size_blocks > 0, "Forward epoch sync batch size must be positive"); + match self + .fast_forward_sync_by_epoch_forward(&mut retriever, epoch, batch_size_blocks) + .await + { + Ok(true) => Ok(()), + Ok(false) => { + info!(epoch = epoch, "Falling back to legacy reverse epoch sync"); + self.fast_forward_sync_by_epoch_legacy(retriever, epoch).await + } + Err(error) => Err(error), + } + } + + async fn fast_forward_sync_by_epoch_forward( + &self, + retriever: &mut BlockRetriever, + epoch: u64, + batch_size_blocks: u64, + ) -> anyhow::Result { + let fetch_root = self.ordered_root(); + let mut fetch_anchor_block_number = fetch_root + .block() + .block_number() + .ok_or_else(|| anyhow!("Ordered root has no block number"))?; + let mut fetch_anchor_block_id = fetch_root.id(); + + let Some((manifest, serving_peer)) = retriever + .try_prepare_forward_epoch_sync(epoch, fetch_anchor_block_number, fetch_anchor_block_id) + .await? + else { + return Ok(false); + }; + info!( + epoch = epoch, + manifest_id = manifest.manifest_id, + first_block_number = manifest.first_block_number, + target_block_number = manifest.target_block_number, + batch_size_blocks = batch_size_blocks, + "Prepared forward epoch sync" + ); + + if self.forward_epoch_sync_target_committed(&manifest)? { + // The blocks and epoch-ending LI are durable, but the self-directed epoch-change + // message is not. Re-send it after restart before reporting the sync as complete. + self.send_committed_epoch_change(retriever, &manifest.target_ledger_info).await?; + return Ok(true); + } + + // The live ordered pipeline can already contain the block targeted by the epoch-ending + // commit decision when sync starts. In that case Prepare supplied the missing signed + // decision, so commit the existing local path instead of fetching past the snapshot tail. + if self.forward_epoch_sync_commit_proof_target_is_ordered(&manifest) { + self.persist_forward_epoch_sync_ledger_infos(std::slice::from_ref( + &manifest.target_ledger_info, + ))?; + retriever.network.send_commit_proof(manifest.target_ledger_info.clone()).await; + if self.wait_for_forward_epoch_sync_target(&manifest).await? { + self.send_committed_epoch_change(retriever, &manifest.target_ledger_info).await?; + return Ok(true); + } + info!( + epoch = epoch, + target_block_number = manifest.target_block_number, + "Local ordered epoch target did not commit in time; use legacy fallback" + ); + return Ok(false); + } + + loop { + let request = ForwardEpochSyncFetchRequest { + epoch, + manifest_id: manifest.manifest_id, + anchor_block_number: fetch_anchor_block_number, + anchor_block_id: fetch_anchor_block_id, + batch_size_blocks, + }; + let Some(batch) = + retriever.fetch_forward_epoch_sync_batch(request.clone(), serving_peer).await? + else { + if self.forward_epoch_sync_target_committed(&manifest)? { + self.send_committed_epoch_change(retriever, &manifest.target_ledger_info) + .await?; + return Ok(true); + } + self.ensure_forward_epoch_sync_target_fetched(&manifest)?; + // All server pages have been consumed and the authenticated epoch target is + // local. Re-submit its decision to unblock an ordered-but-not-committed pipeline, + // then give execution a bounded window to advance the commit root. + self.persist_forward_epoch_sync_ledger_infos(std::slice::from_ref( + &manifest.target_ledger_info, + ))?; + retriever.network.send_commit_proof(manifest.target_ledger_info.clone()).await; + if self.wait_for_forward_epoch_sync_target(&manifest).await? { + self.send_committed_epoch_change(retriever, &manifest.target_ledger_info) + .await?; + return Ok(true); + } + info!( + epoch = epoch, + target_block_number = manifest.target_block_number, + fetch_anchor_block_number = fetch_anchor_block_number, + "Forward epoch sync reached end of data before target committed; use legacy fallback" + ); + return Ok(false); + }; + + BLOCKS_FETCHED_FROM_NETWORK_WHILE_FAST_FORWARD_SYNC.inc_by(batch.records.len() as u64); + self.persist_and_process_forward_epoch_sync_batch(&batch).await?; + + let committed = self.commit_root(); + let committed_number = committed + .block() + .block_number() + .ok_or_else(|| anyhow!("Commit root has no block number after forward replay"))?; + + fetch_anchor_block_number = batch.next_anchor_block_number; + fetch_anchor_block_id = batch.next_anchor_block_id; + + info!( + epoch = epoch, + fetch_anchor_block_number = fetch_anchor_block_number, + committed_block_number = committed_number, + ledger_info_count = batch.ledger_infos.len(), + "Forward epoch sync batch persisted and processed" + ); + if self.forward_epoch_sync_target_committed(&manifest)? { + self.send_committed_epoch_change(retriever, &manifest.target_ledger_info).await?; + return Ok(true); + } + } + } + + fn ensure_forward_epoch_sync_target_fetched( + &self, + manifest: &ForwardEpochSyncManifest, + ) -> anyhow::Result<()> { + let target = self.get_block(manifest.target_block_id).ok_or_else(|| { + anyhow!( + "Forward epoch sync source exhausted before target block {} was fetched", + manifest.target_block_id + ) + })?; + ensure!( + target.block().block_number() == Some(manifest.target_block_number), + "Forward epoch sync target block number does not match manifest" + ); + let commit_proof_target = manifest.target_ledger_info.ledger_info().commit_info().id(); + ensure!( + self.block_exists(commit_proof_target), + "Forward epoch sync source exhausted before commit-proof target {} was fetched", + commit_proof_target + ); + Ok(()) + } + + fn forward_epoch_sync_target_committed( + &self, + manifest: &ForwardEpochSyncManifest, + ) -> anyhow::Result { + let committed = self.commit_root(); + let committed_number = committed + .block() + .block_number() + .ok_or_else(|| anyhow!("Commit root has no block number during forward epoch sync"))?; + if committed_number == manifest.target_block_number { + ensure!( + committed.id() == manifest.target_block_id, + "Forward epoch sync target conflicts with local commit root" + ); + } + Ok(committed_number >= manifest.target_block_number) + } + + fn forward_epoch_sync_commit_proof_target_is_ordered( + &self, + manifest: &ForwardEpochSyncManifest, + ) -> bool { + // For a non-blocking epoch change the signed commit decision can point at a suffix block, + // while `target_block_id` is the earlier epoch-change boundary. The buffer manager indexes + // commit decisions by `commit_info().id()`, so only take this recovery path after that + // exact block is already present on the local ordered path. Otherwise ordinary + // paging must keep fetching the suffix until the proof can be applied. + let commit_proof_target = manifest.target_ledger_info.ledger_info().commit_info().id(); + self.path_from_commit_root(self.ordered_root().id()) + .is_some_and(|path| path.iter().any(|block| block.id() == commit_proof_target)) + } + + async fn wait_for_forward_epoch_sync_target( + &self, + manifest: &ForwardEpochSyncManifest, + ) -> anyhow::Result { + match time::timeout(Duration::from_millis(RPC_TIMEOUT_MSEC), async { + loop { + if self.forward_epoch_sync_target_committed(manifest)? { + return Ok::<(), anyhow::Error>(()); + } + time::sleep(Duration::from_millis(10)).await; + } + }) + .await + { + Ok(result) => { + result?; + Ok(true) + } + Err(_) => Ok(false), + } + } + + fn persist_forward_epoch_sync_ledger_infos( + &self, + ledger_infos: &[LedgerInfoWithSignatures], + ) -> anyhow::Result<()> { + if ledger_infos.is_empty() { + return Ok(()); + } + let consensus_db = self.storage.consensus_db(); + let metadata_db = consensus_db.ledger_db.metadata_db(); + let mut ledger_info_batch = SchemaBatch::new(); + for ledger_info in ledger_infos { + metadata_db.put_ledger_info(ledger_info, &mut ledger_info_batch)?; + } + metadata_db.write_schemas(ledger_info_batch)?; + metadata_db.update_latest_ledger_info()?; + Ok(()) + } + + async fn persist_and_process_forward_epoch_sync_batch( + &self, + batch: &ForwardEpochSyncBatch, + ) -> anyhow::Result<()> { + for record in &batch.records { + if let Some(payload) = record.block.payload() { + self.payload_manager.prefetch_payload_data(payload, record.block.timestamp_usecs()); + } + } + let blocks = batch.records.iter().map(|record| record.block.clone()).collect::>(); + let quorum_certs = + batch.records.iter().map(|record| record.quorum_cert.clone()).collect::>(); + let block_numbers = batch + .records + .iter() + .filter_map(|record| { + record + .block_number + .map(|block_number| (batch.epoch, block_number, record.block.id())) + }) + .collect::>(); + self.storage.save_tree(blocks, quorum_certs.clone(), block_numbers)?; + self.storage.consensus_db().put_randomness( + &batch + .records + .iter() + .filter_map(|record| record.block_number.zip(record.randomness.clone())) + .collect::>(), + )?; + + self.persist_forward_epoch_sync_ledger_infos(&batch.ledger_infos)?; + + let sync_blocks = batch + .records + .iter() + .map(|record| (record.block.clone(), record.block_number, record.randomness.clone())) + .collect(); + self.append_blocks_for_sync_checked(sync_blocks, quorum_certs).await + } +} + +impl BlockRetriever { + async fn request_forward_epoch_sync( + &mut self, + request: ForwardEpochSyncRequest, + peers: Vec, + rpc_timeout: Duration, + max_attempts: usize, + ) -> anyhow::Result<(ForwardEpochSyncResponse, AccountAddress)> { + ensure!(!peers.is_empty(), "No peers available for forward epoch sync"); + let mut candidates = peers; + let attempts = max_attempts.max(1); + let mut last_error = None; + for attempt in 0..attempts { + if candidates.is_empty() && attempt > 0 { + break; + } + let peer = self.pick_peer(attempt == 0, &mut candidates); + match self + .network + .request_forward_epoch_sync( + request.clone(), + PeerNetworkId::new(self.network_id, peer), + rpc_timeout, + ) + .await + { + Ok(ForwardEpochSyncResponse::V1(ForwardEpochSyncResponseV1::Error( + ForwardEpochSyncError::Busy, + ))) => { + last_error = Some(anyhow!("Forward epoch sync peer {peer} is busy")); + time::sleep(Duration::from_millis(RETRY_INTERVAL_MSEC)).await; + } + Ok(response) => return Ok((response, peer)), + Err(error) => { + warn!(remote_peer = peer, error = ?error, "Forward epoch sync RPC failed"); + last_error = Some(error); + } + } + } + Err(last_error.unwrap_or_else(|| anyhow!("No forward epoch sync peer available"))) + } + + async fn request_forward_epoch_sync_from_peer( + &self, + request: ForwardEpochSyncRequest, + peer: AccountAddress, + rpc_timeout: Duration, + max_attempts: usize, + ) -> anyhow::Result { + let attempts = max_attempts.max(1); + let mut last_error = None; + for attempt in 0..attempts { + match self + .network + .request_forward_epoch_sync( + request.clone(), + PeerNetworkId::new(self.network_id, peer), + rpc_timeout, + ) + .await + { + Ok(ForwardEpochSyncResponse::V1(ForwardEpochSyncResponseV1::Error( + ForwardEpochSyncError::Busy, + ))) => { + last_error = Some(anyhow!("Forward epoch sync peer {peer} is busy")); + } + Ok(response) => return Ok(response), + Err(error) => { + warn!(remote_peer = peer, error = ?error, "Forward epoch sync RPC failed"); + last_error = Some(error); + } + } + if attempt + 1 < attempts { + time::sleep(Duration::from_millis(RETRY_INTERVAL_MSEC)).await; + } + } + Err(last_error.unwrap_or_else(|| anyhow!("Forward epoch sync peer {peer} unavailable"))) + } + + async fn try_prepare_forward_epoch_sync( + &mut self, + epoch: u64, + anchor_block_number: u64, + anchor_block_id: HashValue, + ) -> anyhow::Result> { + let request = ForwardEpochSyncRequest::V1(ForwardEpochSyncRequestV1::Prepare( + ForwardEpochSyncPrepareRequest { epoch, anchor_block_number, anchor_block_id }, + )); + // Capability probing is deliberately short and bounded. During a rolling upgrade an old + // peer cannot decode the appended enum variant, so the caller must quickly fall back to + // the legacy reverse retrieval path. + let (response, serving_peer) = match self + .request_forward_epoch_sync( + request, + self.available_peers.clone(), + Duration::from_millis(FORWARD_EPOCH_SYNC_PREPARE_TIMEOUT_MSEC), + 2, + ) + .await + { + Ok(response) => response, + Err(error) => { + info!(epoch = epoch, error = ?error, "Forward epoch sync unavailable; use legacy fallback"); + return Ok(None); + } + }; + let ForwardEpochSyncResponse::V1(response) = response; + match response { + ForwardEpochSyncResponseV1::Prepared(manifest) => { + ensure!(manifest.epoch == epoch, "Forward manifest epoch mismatch"); + ensure!( + manifest.target_ledger_info.ledger_info().epoch() == epoch, + "Forward manifest target LI epoch mismatch" + ); + ensure!( + manifest.target_ledger_info.ledger_info().ends_epoch(), + "Forward manifest target does not end epoch" + ); + manifest.target_ledger_info.verify_signatures(self.network.validators())?; + let epoch_info = + manifest.target_ledger_info.ledger_info().commit_info().epoch_block_info(); + let expected_target_id = + epoch_info.map(|info| info.block_id).unwrap_or_else(|| { + manifest.target_ledger_info.ledger_info().consensus_block_id() + }); + let expected_target_number = epoch_info + .map(|info| info.block_number) + .unwrap_or_else(|| manifest.target_ledger_info.ledger_info().block_number()); + ensure!( + manifest.target_block_id == expected_target_id && + manifest.target_block_number == expected_target_number, + "Forward manifest target mismatch" + ); + ensure!( + manifest.first_block_number <= manifest.target_block_number, + "Forward manifest block range is invalid" + ); + Ok(Some((*manifest, serving_peer))) + } + ForwardEpochSyncResponseV1::Error(error) => { + info!(epoch = epoch, error = ?error, "Forward epoch sync prepare rejected; use legacy fallback"); + Ok(None) + } + ForwardEpochSyncResponseV1::Batch(_) => { + bail!("Forward epoch sync prepare returned a batch") + } + } + } + + /// Returns `Ok(None)` only when the server accepted the cursor and reported that no page + /// remains. All other server and verification failures remain hard errors. + async fn fetch_forward_epoch_sync_batch( + &self, + request: ForwardEpochSyncFetchRequest, + serving_peer: AccountAddress, + ) -> anyhow::Result> { + let response = self + .request_forward_epoch_sync_from_peer( + ForwardEpochSyncRequest::V1(ForwardEpochSyncRequestV1::Fetch(request.clone())), + serving_peer, + Duration::from_millis(RPC_TIMEOUT_MSEC), + NUM_RETRIES, + ) + .await?; + let ForwardEpochSyncResponse::V1(response) = response; + let Some(batch) = decode_forward_epoch_sync_fetch_response(response)? else { + return Ok(None); + }; + self.verify_forward_epoch_sync_batch(&request, &batch)?; + Ok(Some(batch)) + } + + fn verify_forward_epoch_sync_batch( + &self, + request: &ForwardEpochSyncFetchRequest, + batch: &ForwardEpochSyncBatch, + ) -> anyhow::Result<()> { + ensure!(batch.epoch == request.epoch, "Forward batch epoch mismatch"); + ensure!(batch.manifest_id == request.manifest_id, "Forward batch manifest mismatch"); + ensure!( + batch.anchor_block_number == request.anchor_block_number && + batch.anchor_block_id == request.anchor_block_id, + "Forward batch anchor echo mismatch" + ); + ensure!(!batch.records.is_empty(), "Forward batch is empty"); + ensure!( + batch.records.len() as u64 <= request.batch_size_blocks, + "Forward batch exceeds requested size" + ); + + let mut expected_parent = request.anchor_block_id; + let mut anchor_block_number = request.anchor_block_number; + for record in &batch.records { + ensure!(record.block.epoch() == request.epoch, "Forward block epoch mismatch"); + ensure!( + record.block.id() == record.block.block_data().hash(), + "Forward block ID does not match its contents" + ); + ensure!(record.block.parent_id() == expected_parent, "Forward blocks are not chained"); + if let Some(block_number) = record.block_number { + ensure!( + block_number == anchor_block_number.saturating_add(1), + "Forward block number gap" + ); + anchor_block_number = block_number; + } else { + ensure!(record.randomness.is_none(), "Unnumbered forward block carries randomness"); + } + if let Some(embedded_number) = record.block.block_number() { + ensure!( + Some(embedded_number) == record.block_number, + "Forward block carries conflicting block number" + ); + } + record.block.validate_signature(self.network.validators())?; + record.block.verify_well_formed()?; + ensure!( + record.quorum_cert.certified_block().id() == record.block.id(), + "Forward QC certifies a different block" + ); + record.quorum_cert.verify(self.network.validators())?; + expected_parent = record.block.id(); + } + let tail = batch.records.last().expect("non-empty checked above"); + ensure!( + batch.next_anchor_block_id == tail.block.id() && + batch.next_anchor_block_number == anchor_block_number, + "Forward batch next anchor is not its tail" + ); + for ledger_info in &batch.ledger_infos { + ensure!( + ledger_info.ledger_info().epoch() == request.epoch, + "Forward ledger info belongs to a different epoch" + ); + ledger_info.verify_signatures(self.network.validators())?; + ensure!( + batch.records.iter().any(|record| { + record.quorum_cert.commit_info().id() == + ledger_info.ledger_info().consensus_block_id() + }), + "Forward ledger info has no certifying QC in its batch" + ); + } + Ok(()) + } +} + +#[cfg(test)] +mod forward_epoch_sync_tests { + use super::{ + certifying_position_in_batch, decode_forward_epoch_sync_fetch_response, + select_forward_batch_end, + }; + use aptos_consensus_types::forward_epoch_sync::{ + ForwardEpochSyncError, ForwardEpochSyncResponseV1, + }; + + #[test] + fn forward_batches_are_regular_pages() { + assert_eq!(select_forward_batch_end(0, 3, 10), Some(3)); + assert_eq!(select_forward_batch_end(3, 4, 10), Some(7)); + assert_eq!(select_forward_batch_end(7, 4, 10), Some(10)); + } + + #[test] + fn forward_batch_has_no_page_after_end() { + assert_eq!(select_forward_batch_end(10, 4, 10), None); + } + + #[test] + fn batch_boundary_not_found_is_end_of_data() { + let response = + ForwardEpochSyncResponseV1::Error(ForwardEpochSyncError::BatchBoundaryNotFound); + assert!(decode_forward_epoch_sync_fetch_response(response).unwrap().is_none()); + } + + #[test] + fn other_fetch_errors_are_not_end_of_data() { + let response = ForwardEpochSyncResponseV1::Error(ForwardEpochSyncError::AnchorMismatch); + assert!(decode_forward_epoch_sync_fetch_response(response).is_err()); + } + + #[test] + fn proof_is_attached_by_certifying_position_only() { + assert!(!certifying_position_in_batch(4, 5, 8)); + assert!(certifying_position_in_batch(5, 5, 8)); + assert!(certifying_position_in_batch(7, 5, 8)); + assert!(!certifying_position_in_batch(8, 5, 8)); + } +} diff --git a/aptos-core/consensus/src/epoch_manager.rs b/aptos-core/consensus/src/epoch_manager.rs index 7e55354b..cd07bf69 100644 --- a/aptos-core/consensus/src/epoch_manager.rs +++ b/aptos-core/consensus/src/epoch_manager.rs @@ -31,8 +31,8 @@ use crate::{ monitor, network::{ self, IncomingBatchRetrievalRequest, IncomingBlockRetrievalRequest, IncomingDAGRequest, - IncomingRandGenRequest, IncomingRpcRequest, IncomingSyncInfoRequest, NetworkReceivers, - NetworkSender, + IncomingForwardEpochSyncRequest, IncomingRandGenRequest, IncomingRpcRequest, + IncomingSyncInfoRequest, NetworkReceivers, NetworkSender, }, network_interface::{ConsensusMsg, ConsensusNetworkClient}, payload_client::{ @@ -59,6 +59,9 @@ use aptos_consensus_types::{ common::{Author, Round}, delayed_qc_msg::DelayedQcMsg, epoch_retrieval::EpochRetrievalRequest, + forward_epoch_sync::{ + ForwardEpochSyncError, ForwardEpochSyncResponse, ForwardEpochSyncResponseV1, + }, proof_of_store::ProofCache, sync_info::SyncInfo, }; @@ -134,6 +137,7 @@ const PROPOSER_ELECTION_CACHING_WINDOW_ADDITION: usize = 3; /// Number of rounds we expect storage to be ahead of the proposer round, /// used for fetching data from DB. const PROPOSER_ROUND_BEHIND_STORAGE_BUFFER: usize = 10; +const FORWARD_EPOCH_SYNC_MAX_CONCURRENT_REQUESTS: usize = 4; #[allow(clippy::large_enum_variant)] pub enum LivenessStorageData { @@ -171,6 +175,8 @@ pub struct EpochManager { epoch_state: Option>, block_retrieval_tx: Option>, + forward_epoch_sync_tx: + Option>, sync_info_request_tx: Option>, quorum_store_msg_tx: Option>, quorum_store_coordinator_tx: Option>, @@ -284,6 +290,7 @@ impl EpochManager

{ buffered_proposal_tx: None, epoch_state: None, block_retrieval_tx: None, + forward_epoch_sync_tx: None, sync_info_request_tx: None, quorum_store_msg_tx: None, quorum_store_coordinator_tx: None, @@ -626,6 +633,67 @@ impl EpochManager

{ tokio::spawn(task); } + fn spawn_forward_epoch_sync_task( + &mut self, + epoch: u64, + block_store: Arc, + max_blocks_allowed: u64, + ) { + let (request_tx, mut request_rx) = + aptos_channel::new::<_, IncomingForwardEpochSyncRequest>(QueueStyle::FIFO, 1, None); + let permits = + Arc::new(tokio::sync::Semaphore::new(FORWARD_EPOCH_SYNC_MAX_CONCURRENT_REQUESTS)); + let task = async move { + info!(epoch = epoch, "Forward epoch sync task starts"); + while let Some(request) = request_rx.next().await { + let permit = match permits.clone().try_acquire_owned() { + Ok(permit) => permit, + Err(_) => { + warn!( + epoch = epoch, + remote_peer = request.sender, + "Reject forward epoch sync request because the service is busy" + ); + Self::respond_forward_epoch_sync_error( + request, + ForwardEpochSyncError::Busy, + ); + continue; + } + }; + let block_store = block_store.clone(); + tokio::spawn(async move { + let _permit = permit; + if let Err(error) = + block_store.process_forward_epoch_sync(request, max_blocks_allowed).await + { + warn!(epoch = epoch, error = ?error, kind = error_kind(&error)); + } + }); + } + info!(epoch = epoch, "Forward epoch sync task stops"); + }; + self.forward_epoch_sync_tx = Some(request_tx); + tokio::spawn(task); + } + + fn respond_forward_epoch_sync_error( + request: IncomingForwardEpochSyncRequest, + error: ForwardEpochSyncError, + ) { + let response = ConsensusMsg::ForwardEpochSyncResponse(Box::new( + ForwardEpochSyncResponse::V1(ForwardEpochSyncResponseV1::Error(error)), + )); + match request.protocol.to_bytes(&response) { + Ok(bytes) => { + let _ = request.response_sender.send(Ok(bytes.into())); + } + Err(encode_error) => { + warn!(error = ?encode_error, "Failed to encode forward epoch sync error"); + } + } + } + fn spawn_sync_info_retrieval_task(&mut self, epoch: u64, block_store: Arc) { let (request_tx, mut request_rx) = aptos_channel::new::<_, IncomingSyncInfoRequest>(QueueStyle::KLAST, 10, None); @@ -671,6 +739,7 @@ impl EpochManager

{ // Shutdown the block retrieval task by dropping the sender self.block_retrieval_tx = None; + self.forward_epoch_sync_tx = None; self.batch_retrieval_tx = None; if let Some(mut quorum_store_coordinator_tx) = self.quorum_store_coordinator_tx.take() { @@ -968,6 +1037,11 @@ impl EpochManager

{ )); self.spawn_block_retrieval_task(epoch, block_store.clone(), max_blocks_allowed); + if crate::forward_epoch_sync_enabled() { + self.spawn_forward_epoch_sync_task(epoch, block_store.clone(), max_blocks_allowed); + } else { + info!(epoch = epoch, "Forward epoch sync service is not enabled"); + } self.spawn_sync_info_retrieval_task(epoch, block_store); } @@ -1774,6 +1848,7 @@ impl EpochManager

{ fn rpc_request_filter(&self, peer_id: &Author, request: &IncomingRpcRequest) -> bool { match request { IncomingRpcRequest::BlockRetrieval(_) | + IncomingRpcRequest::ForwardEpochSync(_) | IncomingRpcRequest::SyncInfoRequest(_) | IncomingRpcRequest::BatchRetrieval(_) => true, _ => self.is_current_epoch_validator, @@ -1805,6 +1880,7 @@ impl EpochManager

{ None => { ensure!( matches!(request, IncomingRpcRequest::BlockRetrieval(_)) || + matches!(request, IncomingRpcRequest::ForwardEpochSync(_)) || matches!(request, IncomingRpcRequest::BatchRetrieval(_)) || matches!(request, IncomingRpcRequest::SyncInfoRequest(_)) ); @@ -1821,6 +1897,32 @@ impl EpochManager

{ Ok(()) } } + IncomingRpcRequest::ForwardEpochSync(request) => { + if let Some(tx) = &self.forward_epoch_sync_tx { + let (status_tx, status_rx) = oneshot::channel(); + tx.push_with_feedback(peer_id, request, Some(status_tx))?; + tokio::spawn(async move { + if let Ok(aptos_channel::ElementStatus::Dropped(request)) = status_rx.await + { + Self::respond_forward_epoch_sync_error( + request, + ForwardEpochSyncError::Busy, + ); + } + }); + Ok(()) + } else { + warn!( + remote_peer = peer_id, + "Reject forward epoch sync request because the service is disabled or not started" + ); + Self::respond_forward_epoch_sync_error( + request, + ForwardEpochSyncError::Internal, + ); + Ok(()) + } + } IncomingRpcRequest::BatchRetrieval(request) => { if let Some(tx) = &self.batch_retrieval_tx { tx.push(peer_id, request) diff --git a/aptos-core/consensus/src/lib.rs b/aptos-core/consensus/src/lib.rs index c1babe56..3ffe897d 100644 --- a/aptos-core/consensus/src/lib.rs +++ b/aptos-core/consensus/src/lib.rs @@ -79,6 +79,17 @@ pub use quorum_store::quorum_store_db::QUORUM_STORE_DB_NAME; #[cfg(feature = "fuzzing")] pub use round_manager::round_manager_fuzzing; +pub(crate) const ENABLE_FORWARD_EPOCH_SYNC_ENV: &str = "ENABLE_FORWARD_EPOCH_SYNC"; + +/// Opt-in switch for the block-number anchored epoch sync path. Nodes use the legacy reverse sync +/// path unless operators explicitly set `ENABLE_FORWARD_EPOCH_SYNC=true`. +pub(crate) fn forward_epoch_sync_enabled() -> bool { + std::env::var(ENABLE_FORWARD_EPOCH_SYNC_ENV) + .ok() + .and_then(|value| value.parse::().ok()) + .unwrap_or(false) +} + struct IntGaugeGuard { gauge: IntGauge, } diff --git a/aptos-core/consensus/src/network.rs b/aptos-core/consensus/src/network.rs index b9b29154..b4a955a4 100644 --- a/aptos-core/consensus/src/network.rs +++ b/aptos-core/consensus/src/network.rs @@ -22,6 +22,7 @@ use anyhow::{anyhow, bail, ensure}; use aptos_consensus_types::{ block_retrieval::{BlockRetrievalRequest, BlockRetrievalResponse}, common::Author, + forward_epoch_sync::{ForwardEpochSyncRequest, ForwardEpochSyncResponse}, order_vote_msg::OrderVoteMsg, pipeline::{commit_decision::CommitDecision, commit_vote::CommitVote}, proof_of_store::{ProofOfStore, ProofOfStoreMsg, SignedBatchInfo, SignedBatchInfoMsg}, @@ -103,6 +104,16 @@ pub struct IncomingBlockRetrievalRequest { pub response_sender: oneshot::Sender>, } +/// A forward-sync RPC together with the authenticated network peer identity. The sender is kept +/// through the whole ingress path so admission control cannot be bypassed by request contents. +#[derive(Debug)] +pub struct IncomingForwardEpochSyncRequest { + pub req: ForwardEpochSyncRequest, + pub sender: Author, + pub protocol: ProtocolId, + pub response_sender: oneshot::Sender>, +} + #[derive(Debug)] pub struct IncomingBatchRetrievalRequest { pub req: BatchRequest, @@ -141,6 +152,7 @@ pub struct IncomingSyncInfoRequest { #[derive(Debug)] pub enum IncomingRpcRequest { BlockRetrieval(IncomingBlockRetrievalRequest), + ForwardEpochSync(IncomingForwardEpochSyncRequest), BatchRetrieval(IncomingBatchRetrievalRequest), DAGRequest(IncomingDAGRequest), CommitRequest(IncomingCommitRequest), @@ -156,6 +168,7 @@ impl IncomingRpcRequest { IncomingRpcRequest::RandGenRequest(req) => Some(req.req.epoch()), IncomingRpcRequest::CommitRequest(req) => req.req.epoch(), IncomingRpcRequest::BlockRetrieval(_) => None, + IncomingRpcRequest::ForwardEpochSync(_) => None, IncomingRpcRequest::SyncInfoRequest(_) => None, } } @@ -231,6 +244,10 @@ impl NetworkSender { } } + pub(crate) fn validators(&self) -> &ValidatorVerifier { + &self.validators + } + /// Tries to retrieve num of blocks backwards starting from id from the given peer: the function /// returns a future that is fulfilled with BlockRetrievalResponse. pub async fn request_block( @@ -287,6 +304,28 @@ impl NetworkSender { Ok(response) } + pub async fn request_forward_epoch_sync( + &self, + request: ForwardEpochSyncRequest, + from: PeerNetworkId, + rpc_timeout: Duration, + ) -> anyhow::Result { + ensure!(from.peer_id() != self.author, "Retrieve epoch from self"); + let msg = ConsensusMsg::ForwardEpochSyncRequest(Box::new(request)); + counters::CONSENSUS_SENT_MSGS.with_label_values(&[msg.name()]).inc(); + let response = monitor!( + "forward_epoch_sync", + self.consensus_network_client + .network_client + .send_to_peer_rpc(msg, rpc_timeout, from) + .await + )?; + match response { + ConsensusMsg::ForwardEpochSyncResponse(response) => Ok(*response), + _ => Err(anyhow!("Invalid forward epoch sync response")), + } + } + pub async fn send_rpc_to_self( &self, msg: ConsensusMsg, @@ -826,6 +865,21 @@ impl NetworkTask { response_sender: callback, }) } + ConsensusMsg::ForwardEpochSyncRequest(request) => { + debug!( + remote_peer = peer_id, + epoch = request.epoch(), + "Receive forward epoch sync request" + ); + IncomingRpcRequest::ForwardEpochSync( + IncomingForwardEpochSyncRequest { + req: *request, + sender: peer_id, + protocol, + response_sender: callback, + }, + ) + } ConsensusMsg::BatchRequestMsg(request) => { debug!( remote_peer = peer_id, diff --git a/aptos-core/consensus/src/network_interface.rs b/aptos-core/consensus/src/network_interface.rs index 85ee0fbb..a04ec865 100644 --- a/aptos-core/consensus/src/network_interface.rs +++ b/aptos-core/consensus/src/network_interface.rs @@ -13,6 +13,7 @@ use crate::{ use aptos_consensus_types::{ block_retrieval::{BlockRetrievalRequest, BlockRetrievalResponse}, epoch_retrieval::EpochRetrievalRequest, + forward_epoch_sync::{ForwardEpochSyncRequest, ForwardEpochSyncResponse}, order_vote_msg::OrderVoteMsg, pipeline::{commit_decision::CommitDecision, commit_vote::CommitVote}, proof_of_store::{ProofOfStoreMsg, SignedBatchInfoMsg}, @@ -84,6 +85,11 @@ pub enum ConsensusMsg { OrderVoteMsg(Box), /// Request to get the sync info from the destination peer. SyncInfoRequest, + /// Versioned RPC for block-number anchored, forward epoch synchronization. + /// + /// Keep new variants at the end of this BCS enum for rolling-upgrade compatibility. + ForwardEpochSyncRequest(Box), + ForwardEpochSyncResponse(Box), } /// Network type for consensus @@ -111,6 +117,8 @@ impl ConsensusMsg { ConsensusMsg::RandGenMessage(_) => "RandGenMessage", ConsensusMsg::BatchResponseV2(_) => "BatchResponseV2", ConsensusMsg::SyncInfoRequest => "SyncInfoRequest", + ConsensusMsg::ForwardEpochSyncRequest(_) => "ForwardEpochSyncRequest", + ConsensusMsg::ForwardEpochSyncResponse(_) => "ForwardEpochSyncResponse", } } } @@ -207,3 +215,15 @@ impl> ConsensusNetworkClient self.network_client.sort_peers_by_latency(NetworkId::Validator, peers); } } + +#[cfg(test)] +mod tests { + use super::ConsensusMsg; + + #[test] + fn forward_sync_variants_are_appended_to_consensus_msg() { + // SyncInfoRequest was the last variant before forward sync was introduced. Its BCS tag + // must stay stable so old messages remain decodable during a rolling upgrade. + assert_eq!(bcs::to_bytes(&ConsensusMsg::SyncInfoRequest).unwrap(), vec![19]); + } +} diff --git a/aptos-core/consensus/src/round_manager.rs b/aptos-core/consensus/src/round_manager.rs index 27e9318e..326decad 100644 --- a/aptos-core/consensus/src/round_manager.rs +++ b/aptos-core/consensus/src/round_manager.rs @@ -1721,7 +1721,15 @@ impl RoundManager { ), VerifiedEvent::EpochChange(epoch) => { if !self.wait_change_epoch_flag { - if let Err(e) = self.block_store.fast_forward_sync_by_epoch(self.create_block_retriever(peer_id), epoch).await { + let batch_size_blocks = self.local_config + .max_blocks_per_sending_request( + self.onchain_config.quorum_store_enabled(), + ); + if let Err(e) = self.block_store.fast_forward_sync_by_epoch( + self.create_block_retriever(peer_id), + epoch, + batch_size_blocks, + ).await { Err(e) } else { self.wait_change_epoch_flag = true;