Topdown consistency checks: Validate number of top down messages

fendermint/vm/topdown/src/error.rs

@@ -15,4 +15,8 @@ pub enum Error {
     ParentChainReorgDetected,
     #[error("Cannot query parent at height {1}: {0}")]
     CannotQueryParent(String, BlockHeight),
+    #[error("Cannot query parent with hash {1}: {0}")]
+    CannotQueryParentHash(String, String),
+    #[error("Number of topdown messages incorrect at height {0}, expected: {1}, found: {2}")]
+    TopDownMsgsLengthIncorrect(BlockHeight, u64, u64),
 }

fendermint/vm/topdown/src/finality/fetch.rs

@@ -334,6 +334,29 @@ mod tests {
                 block_hash: r.0,
             })
         }
+
+        async fn get_top_down_nonce(
+            &self,
+            block_hash: &[u8],
+        ) -> anyhow::Result<TopDownQueryPayload<u64>> {
+            let mut nonce = 0;
+            for v in self.blocks.values() {
+                let Some(p) = v else {
+                    continue;
+                };
+
+                nonce += p.2.len();
+
+                if p.0 == *block_hash {
+                    return Ok(TopDownQueryPayload {
+                        value: nonce as u64,
+                        block_hash: block_hash.to_vec(),
+                    });
+                }
+            }
+
+            Err(anyhow!("not found"))
+        }
     }
 
     fn new_provider(

fendermint/vm/topdown/src/finality/mod.rs

@@ -88,6 +88,16 @@ mod tests {
                 block_hash: vec![],
             })
         }
+
+        async fn get_top_down_nonce(
+            &self,
+            block_hash: &[u8],
+        ) -> anyhow::Result<TopDownQueryPayload<u64>> {
+            Ok(TopDownQueryPayload {
+                value: 0,
+                block_hash: block_hash.to_vec(),
+            })
+        }
     }
 
     fn mocked_agent_proxy() -> Arc<MockedParentQuery> {

fendermint/vm/topdown/src/lib.rs

@@ -29,6 +29,7 @@ pub use crate::toggle::Toggle;
 pub type BlockHeight = u64;
 pub type Bytes = Vec<u8>;
 pub type BlockHash = Bytes;
+pub type Nonce = u64;
 
 /// The null round error message
 pub(crate) const NULL_ROUND_ERR_MSG: &str = "requested epoch was a null round";

fendermint/vm/topdown/src/proxy.rs

@@ -36,6 +36,12 @@ pub trait ParentQueryProxy {
         &self,
         height: BlockHeight,
     ) -> anyhow::Result<TopDownQueryPayload<Vec<StakingChangeRequest>>>;
+
+    /// Get the top down message nonce at the target block
+    async fn get_top_down_nonce(
+        &self,
+        block_hash: &[u8],
+    ) -> anyhow::Result<TopDownQueryPayload<u64>>;
 }
 
 /// The proxy to the subnet's parent
@@ -115,4 +121,13 @@ impl ParentQueryProxy for IPCProviderProxy {
                 v
             })
     }
+
+    async fn get_top_down_nonce(
+        &self,
+        block_hash: &[u8],
+    ) -> anyhow::Result<TopDownQueryPayload<u64>> {
+        self.ipc_provider
+            .get_top_down_nonce(&self.child_subnet, block_hash)
+            .await
+    }
 }

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

@@ -7,7 +7,7 @@ use crate::proxy::ParentQueryProxy;
 use crate::sync::{query_starting_finality, ParentFinalityStateQuery};
 use crate::voting::{self, VoteTally};
 use crate::{
-    is_null_round_str, BlockHash, BlockHeight, CachedFinalityProvider, Config, Error, Toggle,
+    is_null_round_str, BlockHash, BlockHeight, CachedFinalityProvider, Config, Error, Nonce, Toggle,
 };
 use anyhow::anyhow;
 use async_stm::{atomically, atomically_or_err, StmError};
@@ -89,14 +89,19 @@ where
             return Ok(());
         }
 
+        let mut parent_block_nonce = None;
         loop {
             if self.exceed_cache_size_limit().await {
                 tracing::debug!("exceeded cache size limit");
                 break;
             }
 
-            first_non_null_parent_hash = match self
-                .poll_next(latest_height_fetched + 1, first_non_null_parent_hash)
+            (first_non_null_parent_hash, parent_block_nonce) = match self
+                .poll_next(
+                    latest_height_fetched + 1,
+                    first_non_null_parent_hash,
+                    parent_block_nonce,
+                )
                 .await
             {
                 Ok(h) => h,
@@ -179,7 +184,8 @@ where
         &mut self,
         height: BlockHeight,
         parent_block_hash: BlockHash,
-    ) -> Result<BlockHash, Error> {
+        parent_block_nonce: Option<Nonce>,
+    ) -> Result<(BlockHash, Option<Nonce>), Error> {
         tracing::debug!(
             height,
             parent_block_hash = hex::encode(&parent_block_hash),
@@ -215,7 +221,7 @@ where
 
                     // Null block received, no block hash for the current height being polled.
                     // Return the previous parent hash as the non-null block hash.
-                    return Ok(parent_block_hash);
+                    return Ok((parent_block_hash, None));
                 }
                 return Err(Error::CannotQueryParent(
                     format!("get_block_hash: {e}"),
@@ -235,6 +241,14 @@ where
         }
 
         let data = self.fetch_data(height, block_hash_res.block_hash).await?;
+        let nonce = self
+            .check_data(
+                height,
+                &data,
+                &block_hash_res.parent_block_hash,
+                parent_block_nonce,
+            )
+            .await?;
 
         tracing::debug!(
             height,
@@ -264,9 +278,68 @@ where
             num_validator_changes: data.1.len(),
         });
 
-        Ok(data.0)
+        Ok((data.0, Some(nonce)))
     }
 
+    #[instrument(skip(self))]
+    async fn check_data(
+        &self,
+        height: BlockHeight,
+        current_view: &ParentViewPayload,
+        parent_block_hash: &BlockHash,
+        parent_block_nonce: Option<Nonce>,
+    ) -> Result<Nonce, Error> {
+        let prev_nonce = if let Some(v) = parent_block_nonce {
+            v
+        } else {
+            self.top_down_nonce(parent_block_hash).await?
+        };
+        let curr_nonce = self.top_down_nonce(&current_view.0).await?;
+
+        tracing::debug!(
+            prev_block = hex::encode(parent_block_hash),
+            prev_nonce,
+            curr_block = hex::encode(&current_view.0),
+            curr_nonce,
+            "consecutive height nonces"
+        );
+
+        let num_msgs = curr_nonce - prev_nonce;
+        let msgs = &current_view.2;
+
+        tracing::debug!(
+            expected = num_msgs,
+            actual = msgs.len(),
+            "expected topdown messages"
+        );
+
+        if num_msgs != msgs.len() as u64 {
+            return Err(Error::TopDownMsgsLengthIncorrect(
+                height,
+                num_msgs,
+                current_view.2.len() as u64,
+            ));
+        }
+
+        Ok(curr_nonce)
+    }
+
+    #[inline]
+    async fn top_down_nonce(&self, block_hash: &[u8]) -> Result<u64, Error> {
+        Ok(self
+            .parent_proxy
+            .get_top_down_nonce(block_hash)
+            .await
+            .map_err(|e| {
+                Error::CannotQueryParentHash(
+                    format!("top down nonce: {e}"),
+                    hex::encode(block_hash),
+                )
+            })?
+            .value)
+    }
+
+    #[instrument(skip(self))]
     async fn fetch_data(
         &self,
         height: BlockHeight,
@@ -474,6 +547,16 @@ mod tests {
                 block_hash: self.blocks.get_value(height).cloned().unwrap().unwrap(),
             })
         }
+
+        async fn get_top_down_nonce(
+            &self,
+            block_hash: &[u8],
+        ) -> anyhow::Result<TopDownQueryPayload<u64>> {
+            Ok(TopDownQueryPayload {
+                value: 0,
+                block_hash: block_hash.to_vec(),
+            })
+        }
     }
 
     async fn new_syncer(

ipc/cli/src/commands/crossmsg/mod.rs

@@ -13,6 +13,7 @@ use fund::FundArgs;
 use propagate::PropagateArgs;
 use release::ReleaseArgs;
 
+use crate::commands::crossmsg::topdown_cross::{GetTopDownMsgNonce, GetTopDownMsgNonceArgs};
 use clap::{Args, Subcommand};
 
 pub mod fund;
@@ -38,6 +39,7 @@ impl CrossMsgsCommandsArgs {
             Commands::PreRelease(args) => PreRelease::handle(global, args).await,
             Commands::Propagate(args) => Propagate::handle(global, args).await,
             Commands::ListTopdownMsgs(args) => ListTopdownMsgs::handle(global, args).await,
+            Commands::ListTopdownNonce(args) => GetTopDownMsgNonce::handle(global, args).await,
             Commands::ParentFinality(args) => LatestParentFinality::handle(global, args).await,
         }
     }
@@ -52,5 +54,6 @@ pub(crate) enum Commands {
     PreRelease(PreReleaseArgs),
     Propagate(PropagateArgs),
     ListTopdownMsgs(ListTopdownMsgsArgs),
+    ListTopdownNonce(GetTopDownMsgNonceArgs),
     ParentFinality(LatestParentFinalityArgs),
 }

ipc/cli/src/commands/crossmsg/topdown_cross.rs

@@ -83,3 +83,33 @@ pub(crate) struct LatestParentFinalityArgs {
     #[arg(long, help = "The subnet id to check parent finality")]
     pub subnet: String,
 }
+
+#[derive(Debug, Args)]
+#[command(about = "Get the top down message nonce at block")]
+pub(crate) struct GetTopDownMsgNonceArgs {
+    #[arg(long, help = "The subnet id to check")]
+    pub subnet: String,
+    #[arg(long, help = "The block hash")]
+    pub block_hash: String,
+}
+
+/// The command to get top down message nonce
+pub(crate) struct GetTopDownMsgNonce;
+
+#[async_trait]
+impl CommandLineHandler for GetTopDownMsgNonce {
+    type Arguments = GetTopDownMsgNonceArgs;
+
+    async fn handle(global: &GlobalArguments, arguments: &Self::Arguments) -> anyhow::Result<()> {
+        log::debug!("get top down message nonce: {:?}", arguments);
+
+        let provider = get_ipc_provider(global)?;
+        let subnet = SubnetID::from_str(&arguments.subnet)?;
+        let block_hash = hex::decode(&arguments.block_hash)?;
+
+        let nonce = provider.get_top_down_nonce(&subnet, &block_hash).await?;
+        println!("{}", nonce.value);
+
+        Ok(())
+    }
+}

ipc/provider/src/lib.rs

@@ -697,12 +697,26 @@ impl IpcProvider {
         public_keys: &[Vec<u8>],
         federated_power: &[u128],
     ) -> anyhow::Result<ChainEpoch> {
-        let parent = subnet.parent().ok_or_else(|| anyhow!("no parent found"))?;
-        let conn = self.get_connection(&parent)?;
+        let conn = self.parent_connection(subnet)?;
         conn.manager()
             .set_federated_power(from, subnet, validators, public_keys, federated_power)
             .await
     }
+
+    pub async fn get_top_down_nonce(
+        &self,
+        subnet: &SubnetID,
+        block_hash: &[u8],
+    ) -> anyhow::Result<TopDownQueryPayload<u64>> {
+        let conn = self.parent_connection(subnet)?;
+        conn.manager().get_top_down_nonce(subnet, block_hash).await
+    }
+
+    fn parent_connection(&self, subnet: &SubnetID) -> anyhow::Result<Connection> {
+        let parent = subnet.parent().ok_or_else(|| anyhow!("no parent found"))?;
+        self.connection(&parent)
+            .ok_or_else(|| anyhow!("parent subnet does not exist"))
+    }
 }
 
 /// Lotus JSON keytype format

ipc/provider/src/manager/evm/manager.rs

@@ -34,7 +34,7 @@ use anyhow::{anyhow, Context, Result};
 use async_trait::async_trait;
 use ethers::abi::Tokenizable;
 use ethers::prelude::k256::ecdsa::SigningKey;
-use ethers::prelude::{Signer, SignerMiddleware};
+use ethers::prelude::{Signer, SignerMiddleware, H256};
 use ethers::providers::{Authorization, Http, Middleware, Provider};
 use ethers::signers::{LocalWallet, Wallet};
 use ethers::types::{BlockId, Eip1559TransactionRequest, ValueOrArray, I256, U256};
@@ -50,6 +50,7 @@ use ipc_api::subnet_id::SubnetID;
 use ipc_wallet::{EthKeyAddress, EvmKeyStore, PersistentKeyStore};
 use num_traits::ToPrimitive;
 use std::result;
+use tracing::log;
 
 pub type DefaultSignerMiddleware = SignerMiddleware<Provider<Http>, Wallet<SigningKey>>;
 
@@ -845,6 +846,36 @@ impl SubnetManager for EthSubnetManager {
         let receipt = pending_tx.retries(TRANSACTION_RECEIPT_RETRIES).await?;
         block_number_from_receipt(receipt)
     }
+
+    async fn get_top_down_nonce(
+        &self,
+        subnet: &SubnetID,
+        block_hash: &[u8],
+    ) -> Result<TopDownQueryPayload<u64>> {
+        let gateway_contract = gateway_getter_facet::GatewayGetterFacet::new(
+            self.ipc_contract_info.gateway_addr,
+            Arc::new(self.ipc_contract_info.provider.clone()),
+        );
+
+        let evm_subnet_id = gateway_getter_facet::SubnetID::try_from(subnet)?;
+        let (exists, nonce) = gateway_contract
+            .get_top_down_nonce(evm_subnet_id)
+            .block(BlockId::Hash(H256::from_slice(block_hash)))
+            .call()
+            .await?;
+
+        let nonce = if !exists {
+            log::warn!("subnet does not exists at block hash, return 0 nonce");
+            0
+        } else {
+            nonce
+        };
+
+        Ok(TopDownQueryPayload {
+            value: nonce,
+            block_hash: block_hash.to_vec(),
+        })
+    }
 }
 
 #[async_trait]

ipc/provider/src/manager/subnet.rs

@@ -166,6 +166,12 @@ pub trait SubnetManager: Send + Sync + TopDownFinalityQuery + BottomUpCheckpoint
         public_keys: &[Vec<u8>],
         federated_power: &[u128],
     ) -> Result<ChainEpoch>;
+
+    async fn get_top_down_nonce(
+        &self,
+        subnet: &SubnetID,
+        block_hash: &[u8],
+    ) -> Result<TopDownQueryPayload<u64>>;
 }
 
 #[derive(Debug)]