FM-259: Send signatures for all pending checkpoints

fendermint/vm/interpreter/src/fvm/checkpoint.rs

@@ -12,15 +12,15 @@ use fvm_shared::{address::Address, chainid::ChainID};
 use fendermint_crypto::SecretKey;
 use fendermint_vm_actor_interface::ipc::BottomUpCheckpoint;
 use fendermint_vm_genesis::{Power, Validator, ValidatorKey};
-use fendermint_vm_ipc_actors::gateway_router_facet::SubnetID;
+use fendermint_vm_ipc_actors::gateway_getter_facet as getter;
+use fendermint_vm_ipc_actors::gateway_router_facet as router;
 
 use super::{
     broadcast::Broadcaster,
     state::{ipc::GatewayCaller, FvmExecState},
+    ValidatorContext,
 };
 
-pub type Checkpoint = BottomUpCheckpoint;
-
 /// Validator voting power snapshot.
 #[derive(Debug, Clone)]
 pub struct PowerTable(pub Vec<Validator>);
@@ -38,7 +38,7 @@ pub async fn maybe_create_checkpoint<C, DB>(
     client: &C,
     gateway: &GatewayCaller<DB>,
     state: &mut FvmExecState<DB>,
-) -> anyhow::Result<Option<(Checkpoint, PowerTable, PowerUpdates)>>
+) -> anyhow::Result<Option<(router::BottomUpCheckpoint, PowerTable, PowerUpdates)>>
 where
     C: Client + Sync + Send + 'static,
     DB: Blockstore + Sync + Send + 'static,
@@ -87,11 +87,66 @@ where
     }
 }
 
+/// Sign the current and any incomplete checkpoints.
+pub async fn broadcast_incomplete_signatures<C, DB>(
+    client: &C,
+    validator_ctx: &ValidatorContext<C>,
+    gateway: &GatewayCaller<DB>,
+    chain_id: ChainID,
+    incomplete_checkpoints: Vec<getter::BottomUpCheckpoint>,
+) -> anyhow::Result<()>
+where
+    C: Client + Clone + Send + Sync + 'static,
+    DB: Blockstore + Send + Sync + 'static,
+{
+    for cp in incomplete_checkpoints {
+        let height = Height::try_from(cp.block_height)?;
+        let power_table = get_power_table(client, height)
+            .await
+            .context("failed to get power table")?;
+
+        if let Some(validator) = power_table
+            .0
+            .iter()
+            .find(|v| v.public_key.0 == validator_ctx.public_key)
+            .cloned()
+        {
+            // TODO: Code generation in the ipc-solidity-actors repo should cater for this.
+            let checkpoint = router::BottomUpCheckpoint {
+                subnet_id: router::SubnetID {
+                    root: cp.subnet_id.root,
+                    route: cp.subnet_id.route,
+                },
+                block_height: cp.block_height,
+                block_hash: cp.block_hash,
+                next_configuration_number: cp.next_configuration_number,
+                cross_messages_hash: cp.cross_messages_hash,
+            };
+
+            // We mustn't do these in parallel because of how nonces are fetched.
+            broadcast_signature(
+                &validator_ctx.broadcaster,
+                gateway,
+                checkpoint,
+                &power_table,
+                &validator,
+                &validator_ctx.secret_key,
+                chain_id,
+            )
+            .await
+            .context("failed to broadcast checkpoint signature")?;
+
+            tracing::debug!(?height, "submitted checkpoint signature");
+        }
+    }
+    Ok(())
+}
+
 /// As a validator, sign the checkpoint and broadcast a transaction to add our signature to the ledger.
 pub async fn broadcast_signature<C, DB>(
     broadcaster: &Broadcaster<C>,
     gateway: &GatewayCaller<DB>,
-    checkpoint: Checkpoint,
+    checkpoint: router::BottomUpCheckpoint,
     power_table: &PowerTable,
     validator: &Validator,
     secret_key: &SecretKey,
@@ -117,7 +172,7 @@ fn should_create_checkpoint<DB>(
     gateway: &GatewayCaller<DB>,
     state: &mut FvmExecState<DB>,
     height: Height,
-) -> anyhow::Result<Option<SubnetID>>
+) -> anyhow::Result<Option<router::SubnetID>>
 where
     DB: Blockstore,
 {
@@ -126,7 +181,7 @@ where
         let is_root = id.route.is_empty();
 
         if !is_root && height.value() % gateway.bottom_up_check_period(state)? == 0 {
-            let id = SubnetID {
+            let id = router::SubnetID {
                 root: id.root,
                 route: id.route,
             };

fendermint/vm/interpreter/src/fvm/exec.rs

@@ -129,45 +129,49 @@ where
     }
 
     async fn end(&self, mut state: Self::State) -> anyhow::Result<(Self::State, Self::EndOutput)> {
-        let updates = if let Some((checkpoint, power_table, updates)) =
+        let updates = if let Some((checkpoint, _, updates)) =
             checkpoint::maybe_create_checkpoint(&self.client, &self.gateway, &mut state)
                 .await
                 .context("failed to create checkpoint")?
         {
             // Asynchronously broadcast signature, if validating.
             if let Some(ref ctx) = self.validator_ctx {
-                if let Some(validator) = power_table
-                    .0
-                    .iter()
-                    .find(|v| v.public_key.0 == ctx.public_key)
-                    .cloned()
-                {
-                    // Do not resend past signatures.
-                    if !self.syncing().await? {
-                        let secret_key = ctx.secret_key.clone();
-                        let broadcaster = ctx.broadcaster.clone();
-                        let gateway = self.gateway.clone();
-                        let chain_id = state.chain_id();
-
-                        tokio::spawn(async move {
-                            let height = checkpoint.block_height;
-
-                            let res = checkpoint::broadcast_signature(
-                                &broadcaster,
-                                &gateway,
-                                checkpoint,
-                                &power_table,
-                                &validator,
-                                &secret_key,
-                                chain_id,
-                            )
-                            .await;
-
-                            if let Err(e) = res {
-                                tracing::error!(error =? e, height, "error broadcasting checkpoint signature");
-                            }
-                        });
-                    }
+                // Do not resend past signatures.
+                if !self.syncing().await? {
+                    // Fetch any incomplete checkpoints synchronously because the state can't be shared across threads.
+                    let incomplete_checkpoints = self
+                        .gateway
+                        .incomplete_checkpoints(&mut state)
+                        .context("failed to fetch incomplete checkpoints")?;
+
+                    debug_assert!(
+                        incomplete_checkpoints
+                            .iter()
+                            .any(|cp| cp.block_height == checkpoint.block_height
+                                && cp.block_hash == checkpoint.block_hash),
+                        "the current checkpoint is incomplete"
+                    );
+
+                    let client = self.client.clone();
+                    let gateway = self.gateway.clone();
+                    let chain_id = state.chain_id();
+                    let height = checkpoint.block_height;
+                    let validator_ctx = ctx.clone();
+
+                    tokio::spawn(async move {
+                        let res = checkpoint::broadcast_incomplete_signatures(
+                            &client,
+                            &validator_ctx,
+                            &gateway,
+                            chain_id,
+                            incomplete_checkpoints,
+                        )
+                        .await;
+
+                        if let Err(e) = res {
+                            tracing::error!(error =? e, height, "error broadcasting checkpoint signature");
+                        }
+                    });
                 }
             }
 

fendermint/vm/interpreter/src/fvm/state/ipc.rs

@@ -14,9 +14,11 @@ use fendermint_vm_actor_interface::{
     ipc::{ValidatorMerkleTree, GATEWAY_ACTOR_ID},
 };
 use fendermint_vm_genesis::Validator;
-use fendermint_vm_ipc_actors::gateway_getter_facet::{GatewayGetterFacet, SubnetID};
-use fendermint_vm_ipc_actors::gateway_router_facet::{BottomUpCheckpoint, GatewayRouterFacet};
+use fendermint_vm_ipc_actors::gateway_getter_facet as getter;
+use fendermint_vm_ipc_actors::gateway_router_facet as router;
 use fendermint_vm_message::signed::sign_secp256k1;
+use getter::GatewayGetterFacet;
+use router::GatewayRouterFacet;
 
 use super::{
     fevm::{ContractCaller, MockProvider},
@@ -66,7 +68,7 @@ impl<DB: Blockstore> GatewayCaller<DB> {
     }
 
     /// Return the current subnet ID.
-    pub fn subnet_id(&self, state: &mut FvmExecState<DB>) -> anyhow::Result<SubnetID> {
+    pub fn subnet_id(&self, state: &mut FvmExecState<DB>) -> anyhow::Result<getter::SubnetID> {
         self.getter.call(state, |c| c.get_network_name())
     }
 
@@ -79,7 +81,7 @@ impl<DB: Blockstore> GatewayCaller<DB> {
     pub fn create_bottom_up_checkpoint(
         &self,
         state: &mut FvmExecState<DB>,
-        checkpoint: BottomUpCheckpoint,
+        checkpoint: router::BottomUpCheckpoint,
         power_table: &[Validator],
     ) -> anyhow::Result<()> {
         // Construct a Merkle tree from the power table, which we can use to validate validator set membership
@@ -96,12 +98,20 @@ impl<DB: Blockstore> GatewayCaller<DB> {
         })
     }
 
+    /// Retrieve checkpoints which have not reached a quorum.
+    pub fn incomplete_checkpoints(
+        &self,
+        state: &mut FvmExecState<DB>,
+    ) -> anyhow::Result<Vec<getter::BottomUpCheckpoint>> {
+        self.getter.call(state, |c| c.get_incomplete_checkpoints())
+    }
+
     /// Construct the input parameters for adding a signature to the checkpoint.
     ///
     /// This will need to be broadcasted as a transaction.
     pub fn add_checkpoint_signature_calldata(
         &self,
-        checkpoint: BottomUpCheckpoint,
+        checkpoint: router::BottomUpCheckpoint,
         power_table: &[Validator],
         validator: &Validator,
         secret_key: &SecretKey,