[checkpoints] Use consensus adapter to send checkpoint signatures (My…

…stenLabs#6470)

This PR uses persistent `ConsensusAdapter` to send checkpoint signatures, instead of directly using stateless `ConsensusClient`.

This is needed for liveness, to ensure that if node restarts the consensus signatures produced by this validators will be eventually submitted to consensus.

crates/sui-core/src/authority_server.rs

@@ -280,8 +280,13 @@ impl ValidatorService {
 
         let certified_checkpoint_output = SendCheckpointToStateSync::new(state_sync_handle);
 
+        let ca_metrics = ConsensusAdapterMetrics::new(&prometheus_registry);
+        // The consensus adapter allows the authority to send user certificates through consensus.
+        let consensus_adapter =
+            ConsensusAdapter::new(Box::new(consensus_client), state.clone(), ca_metrics);
+
         let checkpoint_output = Box::new(SubmitCheckpointToConsensus {
-            sender: consensus_client.clone(),
+            sender: consensus_adapter.clone(),
             signer: state.secret.clone(),
             authority: config.protocol_public_key(),
         });
@@ -322,12 +327,6 @@ impl ValidatorService {
             &registry,
         ));
 
-        let ca_metrics = ConsensusAdapterMetrics::new(&prometheus_registry);
-
-        // The consensus adapter allows the authority to send user certificates through consensus.
-        let consensus_adapter =
-            ConsensusAdapter::new(Box::new(consensus_client), state.clone(), ca_metrics);
-
         Ok(Self {
             state,
             consensus_adapter,

crates/sui-core/src/consensus_adapter.rs

@@ -439,6 +439,13 @@ impl<'a> Drop for InflightDropGuard<'a> {
 // TODO: replace by a ConsensusTxValidator in order to make Narwhal-side validation effective
 pub use narwhal_worker::TrivialTransactionValidator as SuiTxValidator;
 
+#[async_trait::async_trait]
+impl SubmitToConsensus for Arc<ConsensusAdapter> {
+    async fn submit_to_consensus(&self, transaction: &ConsensusTransaction) -> SuiResult {
+        self.submit(transaction.clone()).map(|_| ())
+    }
+}
+
 #[cfg(test)]
 mod adapter_tests {
     use super::ConsensusAdapter;