feat(node): refactor parent view syncer

fendermint/vm/topdown/src/lib.rs

@@ -13,6 +13,7 @@ pub mod voting;
 
 pub mod observation;
 pub mod observe;
+pub mod syncer;
 pub mod vote;
 
 use async_stm::Stm;
@@ -28,6 +29,7 @@ use std::time::Duration;
 pub use crate::cache::{SequentialAppendError, SequentialKeyCache, ValueIter};
 pub use crate::error::Error;
 pub use crate::finality::CachedFinalityProvider;
+use crate::observation::Observation;
 pub use crate::toggle::Toggle;
 
 pub type BlockHeight = u64;
@@ -108,6 +110,13 @@ impl Config {
     }
 }
 
+/// On-chain data structure representing a topdown checkpoint agreed to by a
+/// majority of subnet validators. DAG-CBOR encoded, embedded in CertifiedCheckpoint.
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub enum Checkpoint {
+    V1(Observation),
+}
+
 /// The finality view for IPC parent at certain height.
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
 pub struct IPCParentFinality {
@@ -193,3 +202,23 @@ pub(crate) fn is_null_round_error(err: &anyhow::Error) -> bool {
 pub(crate) fn is_null_round_str(s: &str) -> bool {
     s.contains(NULL_ROUND_ERR_MSG)
 }
+
+impl Checkpoint {
+    pub fn target_height(&self) -> BlockHeight {
+        match self {
+            Checkpoint::V1(b) => b.parent_height,
+        }
+    }
+
+    pub fn target_hash(&self) -> &Bytes {
+        match self {
+            Checkpoint::V1(b) => &b.parent_hash,
+        }
+    }
+
+    pub fn cumulative_effects_comm(&self) -> &Bytes {
+        match self {
+            Checkpoint::V1(b) => &b.cumulative_effects_comm,
+        }
+    }
+}

fendermint/vm/topdown/src/observation.rs

@@ -1,25 +1,44 @@
 // Copyright 2022-2024 Protocol Labs
 // SPDX-License-Identifier: Apache-2.0, MIT
 
-use crate::{BlockHeight, Bytes};
+use crate::syncer::error::Error;
+use crate::syncer::store::ParentViewStore;
+use crate::{BlockHash, BlockHeight, Bytes, Checkpoint};
 use anyhow::anyhow;
 use arbitrary::Arbitrary;
+use cid::Cid;
 use fendermint_crypto::secp::RecoverableECDSASignature;
 use fendermint_crypto::SecretKey;
 use fendermint_vm_genesis::ValidatorKey;
+use fvm_ipld_encoding::DAG_CBOR;
+use multihash::Code;
+use multihash::MultihashDigest;
 use serde::{Deserialize, Serialize};
+use std::cmp::min;
 use std::fmt::{Display, Formatter};
 
+use crate::syncer::payload::ParentBlockView;
+
+/// Default topdown observation height range
+const DEFAULT_MAX_OBSERVATION_RANGE: BlockHeight = 100;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ObservationConfig {
+    /// The max number of blocks one should make the topdown observation from the previous
+    /// committed checkpoint
+    pub max_observation_range: Option<BlockHeight>,
+}
+
 /// The content that validators gossip among each other.
 #[derive(Serialize, Deserialize, Hash, Debug, Clone, Eq, PartialEq, Arbitrary)]
 pub struct Observation {
-    pub(crate) parent_height: u64,
+    pub(crate) parent_subnet_height: u64,
     /// The hash of the chain unit at that height. Usually a block hash, but could
     /// be another entity (e.g. tipset CID), depending on the parent chain
     /// and our interface to it. For example, if the parent is a Filecoin network,
     /// this would be a tipset CID coerced into a block hash if queried through
     /// the Eth API, or the tipset CID as-is if accessed through the Filecoin API.
-    pub(crate) parent_hash: Bytes,
+    pub(crate) parent_subnet_hash: Bytes,
     /// A rolling/cumulative commitment to topdown effects since the beginning of
     /// time, including the ones in this block.
     pub(crate) cumulative_effects_comm: Bytes,
@@ -41,6 +60,53 @@ pub struct CertifiedObservation {
     signature: RecoverableECDSASignature,
 }
 
+/// Check in the store to see if there is a new observation available.
+/// Caller should make sure:
+/// - the store has votes since the last committed checkpoint
+/// - the votes have at least 1 non-null block
+pub fn deduce_new_observation<S: ParentViewStore>(
+    store: &S,
+    checkpoint: &Checkpoint,
+    config: &ObservationConfig,
+) -> Result<Observation, Error> {
+    let Some(latest_height) = store.max_parent_view_height()? else {
+        tracing::info!("no observation yet as height not available");
+        return Err(Error::BlockStoreEmpty);
+    };
+
+    if latest_height < checkpoint.target_height() {
+        tracing::info!("committed vote height more than latest parent view");
+        return Err(Error::CommittedParentHeightNotPurged);
+    }
+
+    let max_observation_height = checkpoint.target_height() + config.max_observation_range();
+    let candidate_height = min(max_observation_height, latest_height);
+    tracing::debug!(
+        max_observation_height,
+        candidate_height,
+        "propose observation height"
+    );
+
+    // aggregate commitment for the observation
+    let mut agg = LinearizedParentBlockView::from(checkpoint);
+    for h in checkpoint.target_height() + 1..=candidate_height {
+        let Some(p) = store.get(h)? else {
+            tracing::debug!(height = h, "not parent block view");
+            return Err(Error::MissingBlockView(h, candidate_height));
+        };
+
+        agg.append(p)?;
+    }
+
+    let observation = agg.into_observation()?;
+    tracing::info!(
+        height = observation.parent_subnet_height,
+        "new observation derived"
+    );
+
+    Ok(observation)
+}
+
 impl TryFrom<&[u8]> for CertifiedObservation {
     type Error = anyhow::Error;
 
@@ -54,6 +120,10 @@ impl CertifiedObservation {
         &self.observation
     }
 
+    pub fn observation_signature(&self) -> &RecoverableECDSASignature {
+        &self.observation_signature
+    }
+
     pub fn ensure_valid(&self) -> anyhow::Result<ValidatorKey> {
         let to_sign = fvm_ipld_encoding::to_vec(&self.observation)?;
         let (pk1, _) = self.observation_signature.recover(&to_sign)?;
@@ -97,8 +167,8 @@ impl CertifiedObservation {
 impl Observation {
     pub fn new(parent_height: BlockHeight, parent_hash: Bytes, commitment: Bytes) -> Self {
         Self {
-            parent_height,
-            parent_hash,
+            parent_subnet_height: parent_height,
+            parent_subnet_hash: parent_hash,
             cumulative_effects_comm: commitment,
         }
     }
@@ -109,15 +179,89 @@ impl Display for Observation {
         write!(
             f,
             "Observation(parent_height={}, parent_hash={}, commitment={})",
-            self.parent_height,
-            hex::encode(&self.parent_hash),
+            self.parent_subnet_height,
+            hex::encode(&self.parent_subnet_hash),
             hex::encode(&self.cumulative_effects_comm),
         )
     }
 }
 
 impl Observation {
     pub fn parent_height(&self) -> BlockHeight {
-        self.parent_height
+        self.parent_subnet_height
     }
 }
+
+impl ObservationConfig {
+    pub fn max_observation_range(&self) -> BlockHeight {
+        self.max_observation_range
+            .unwrap_or(DEFAULT_MAX_OBSERVATION_RANGE)
+    }
+}
+
+pub(crate) struct LinearizedParentBlockView {
+    parent_height: u64,
+    parent_hash: Option<BlockHash>,
+    cumulative_effects_comm: Bytes,
+}
+
+impl From<&Checkpoint> for LinearizedParentBlockView {
+    fn from(value: &Checkpoint) -> Self {
+        LinearizedParentBlockView {
+            parent_height: value.target_height(),
+            parent_hash: Some(value.target_hash().clone()),
+            cumulative_effects_comm: value.cumulative_effects_comm().clone(),
+        }
+    }
+}
+
+impl From<&Observation> for LinearizedParentBlockView {
+    fn from(value: &Observation) -> Self {
+        LinearizedParentBlockView {
+            parent_height: value.parent_subnet_height,
+            parent_hash: Some(value.parent_subnet_hash.clone()),
+            cumulative_effects_comm: value.cumulative_effects_comm.clone(),
+        }
+    }
+}
+
+impl LinearizedParentBlockView {
+    fn new_commitment(&mut self, to_append: Bytes) {
+        let bytes = [
+            self.cumulative_effects_comm.as_slice(),
+            to_append.as_slice(),
+        ]
+            .concat();
+        let cid = Cid::new_v1(DAG_CBOR, Code::Blake2b256.digest(&bytes));
+        self.cumulative_effects_comm = cid.to_bytes();
+    }
+
+    pub fn append(&mut self, view: ParentBlockView) -> Result<(), Error> {
+        if self.parent_height + 1 != view.parent_height {
+            return Err(Error::NotSequential);
+        }
+
+        self.parent_height += 1;
+
+        self.new_commitment(view.effects_commitment()?);
+
+        if let Some(p) = view.payload {
+            self.parent_hash = Some(p.parent_hash);
+        }
+
+        Ok(())
+    }
+
+    pub fn into_observation(self) -> Result<Observation, Error> {
+        let Some(hash) = self.parent_hash else {
+            return Err(Error::CannotCommitObservationAtNullBlock(
+                self.parent_height,
+            ));
+        };
+        Ok(Observation::new(
+            self.parent_height,
+            hash,
+            self.cumulative_effects_comm,
+        ))
+    }
+}

fendermint/vm/topdown/src/proxy.rs

@@ -13,6 +13,7 @@ use ipc_api::subnet_id::SubnetID;
 use ipc_observability::emit;
 use ipc_provider::manager::{GetBlockHashResult, TopDownQueryPayload};
 use ipc_provider::IpcProvider;
+use std::sync::Arc;
 use std::time::Instant;
 use tracing::instrument;
 
@@ -42,6 +43,35 @@ pub trait ParentQueryProxy {
     ) -> anyhow::Result<TopDownQueryPayload<Vec<StakingChangeRequest>>>;
 }
 
+#[async_trait]
+impl<P: Send + Sync + 'static + ParentQueryProxy> ParentQueryProxy for Arc<P> {
+    async fn get_chain_head_height(&self) -> Result<BlockHeight> {
+        self.as_ref().get_chain_head_height().await
+    }
+
+    async fn get_genesis_epoch(&self) -> Result<BlockHeight> {
+        self.as_ref().get_genesis_epoch().await
+    }
+
+    async fn get_block_hash(&self, height: BlockHeight) -> Result<GetBlockHashResult> {
+        self.as_ref().get_block_hash(height).await
+    }
+
+    async fn get_top_down_msgs(
+        &self,
+        height: BlockHeight,
+    ) -> Result<TopDownQueryPayload<Vec<IpcEnvelope>>> {
+        self.as_ref().get_top_down_msgs(height).await
+    }
+
+    async fn get_validator_changes(
+        &self,
+        height: BlockHeight,
+    ) -> Result<TopDownQueryPayload<Vec<StakingChangeRequest>>> {
+        self.as_ref().get_validator_changes(height).await
+    }
+}
+
 /// The proxy to the subnet's parent
 pub struct IPCProviderProxy {
     ipc_provider: IpcProvider,

fendermint/vm/topdown/src/syncer/error.rs

@@ -0,0 +1,28 @@
+// Copyright 2022-2024 Protocol Labs
+// SPDX-License-Identifier: Apache-2.0, MIT
+
+use crate::BlockHeight;
+use thiserror::Error;
+
+/// The errors for top down checkpointing
+#[derive(Error, Debug, Eq, PartialEq, Clone)]
+pub enum Error {
+    #[error("Incoming items are not order sequentially")]
+    NotSequential,
+    #[error("The parent view update with block height is not sequential")]
+    NonSequentialParentViewInsert,
+    #[error("Parent chain reorg detected")]
+    ParentChainReorgDetected,
+    #[error("Cannot query parent at height {1}: {0}")]
+    CannotQueryParent(String, BlockHeight),
+    #[error("Parent block view store is empty")]
+    BlockStoreEmpty,
+    #[error("Committed block height not purged yet")]
+    CommittedParentHeightNotPurged,
+    #[error("Cannot serialize parent block view payload to bytes")]
+    CannotSerializeParentBlockView,
+    #[error("Cannot create commitment at null parent block {0}")]
+    CannotCommitObservationAtNullBlock(BlockHeight),
+    #[error("Missing block view at height {0} for target observation height {0}")]
+    MissingBlockView(BlockHeight, BlockHeight),
+}