fix: state-sync

chain/src/application.rs

@@ -18,7 +18,7 @@ use futures::{lock::Mutex as AsyncMutex, StreamExt};
 use nunchi_common::{Overlay, QmdbBatch, QmdbDatabaseSet, QmdbMerkleized, Runtime, RuntimeContext};
 use nunchi_dkg::{Context, Scheme};
 use nunchi_mempool::{MempoolHandle, PoolTransaction};
-use rand::Rng;
+use rand::{CryptoRng, Rng};
 use std::{
     collections::HashMap,
     marker::PhantomData,
@@ -29,8 +29,8 @@ use std::{
 use tracing::{debug, error};
 
 use crate::{
-    Block, ConsensusExtension, DkgMailbox, EventConsumer, NoConsensusExtension, NoopEventConsumer,
-    StateCommitment, TransactionEventContext,
+    Block, ConsensusExtension, DkgMailbox, DkgState, EventConsumer, NoConsensusExtension,
+    NoopEventConsumer, StateCommitment, TransactionEventContext,
 };
 
 /// The height of the last finalized block applied to a node's ledger.
@@ -67,6 +67,7 @@ where
     submitter: MempoolHandle<R::Transaction>,
     max_block_transactions: usize,
     dkg: Option<DkgMailbox<R::Transaction, Ext>>,
+    dkg_state: Option<DkgState>,
     consensus: Ext,
     events: Events,
     applied_height: SharedAppliedHeight,
@@ -83,6 +84,7 @@ struct ApplicationMetrics {
     proposal_validate_duration: Histogram,
     proposal_merkleize_duration: Histogram,
     apply_transactions_duration: Histogram,
+    dkg_state_duration: Histogram,
     apply_merkleize_duration: Histogram,
 }
 
@@ -109,6 +111,11 @@ impl ApplicationMetrics {
                 "duration spent applying block transactions",
                 Buckets::LOCAL,
             ),
+            dkg_state_duration: context.histogram(
+                "dkg_public_state_duration_seconds",
+                "duration spent validating and transitioning authenticated public DKG state",
+                Buckets::LOCAL,
+            ),
             apply_merkleize_duration: context.histogram(
                 "apply_merkleize_duration_seconds",
                 "duration spent merkleizing applied block state",
@@ -162,6 +169,7 @@ where
             submitter,
             max_block_transactions,
             dkg,
+            dkg_state: None,
             consensus,
             events,
             applied_height,
@@ -273,17 +281,21 @@ where
         Some((included, merkleized))
     }
 
-    async fn build_proposal_state<E: BufferPooler + Storage + Clock + Metrics>(
+    async fn build_proposal_state<
+        E: BufferPooler + Storage + Clock + Metrics + CryptoRng,
+    >(
         &mut self,
-        timing: Option<(&E, &ApplicationMetrics)>,
+        timing: Option<&ApplicationMetrics>,
         batches: <QmdbDatabaseSet<E> as DatabaseSet<E>>::Unmerkleized,
         context: RuntimeContext,
         candidates: Vec<R::Transaction>,
+        reshare_log: Option<&nunchi_dkg::DealerLog>,
+        rng: &mut E,
     ) -> Option<(Vec<R::Transaction>, Ext::Payload, QmdbMerkleized<E>)> {
         let mut batch = QmdbBatch::new(batches);
         let mut included = Vec::new();
 
-        let validate_start = timing.map(|(runtime_context, _)| runtime_context.current());
+        let validate_start = timing.map(|_| rng.current());
         for transaction in candidates {
             if included.len() == self.max_block_transactions {
                 break;
@@ -303,10 +315,10 @@ where
                 }
             }
         }
-        if let Some(((runtime_context, metrics), validate_start)) = timing.zip(validate_start) {
+        if let Some((metrics, validate_start)) = timing.zip(validate_start) {
             metrics
                 .proposal_validate_duration
-                .observe_between(validate_start, runtime_context.current());
+                .observe_between(validate_start, rng.current());
         }
 
         let extension = self.consensus.propose().await;
@@ -317,37 +329,58 @@ where
         {
             return None;
         }
+        let dkg_start = timing.map(|_| rng.current());
+        if let Some(dkg_state) = &self.dkg_state {
+            if let Err(error) = dkg_state
+                .apply_block(
+                    &mut batch,
+                    Height::new(context.height),
+                    reshare_log,
+                    rng,
+                )
+                .await
+            {
+                debug!(?error, "invalid authenticated DKG public-state update");
+                return None;
+            }
+        }
+        if let Some((metrics, dkg_start)) = timing.zip(dkg_start) {
+            metrics
+                .dkg_state_duration
+                .observe_between(dkg_start, rng.current());
+        }
 
-        let merkleize_start = timing.map(|(runtime_context, _)| runtime_context.current());
+        let merkleize_start = timing.map(|_| rng.current());
         let merkleized = match batch.merkleize().await {
             Ok(merkleized) => merkleized,
             Err(error) => {
                 error!(?error, "merkleization failed while building block");
                 return None;
             }
         };
-        if let Some(((runtime_context, metrics), merkleize_start)) = timing.zip(merkleize_start) {
+        if let Some((metrics, merkleize_start)) = timing.zip(merkleize_start) {
             metrics
                 .proposal_merkleize_duration
-                .observe_between(merkleize_start, runtime_context.current());
+                .observe_between(merkleize_start, rng.current());
         }
         Some((included, extension, merkleized))
     }
 
     #[allow(clippy::too_many_arguments)]
     async fn execute_block<E, EventHandler>(
         &mut self,
-        runtime_context: &E,
+        runtime_context: &mut E,
         metrics: &ApplicationMetrics,
         batches: <QmdbDatabaseSet<E> as DatabaseSet<E>>::Unmerkleized,
         context: RuntimeContext,
         transactions: &[R::Transaction],
         extension: &Ext::Payload,
+        reshare_log: Option<&nunchi_dkg::DealerLog>,
         events: &EventHandler,
         commit_extension: bool,
     ) -> Option<QmdbMerkleized<E>>
     where
-        E: BufferPooler + Storage + Clock + Metrics,
+        E: BufferPooler + Storage + Clock + Metrics + CryptoRng,
         EventHandler: EventConsumer,
     {
         events.begin_block(context).await;
@@ -389,6 +422,27 @@ where
                 .commit_payload(&mut batch, context, extension)
                 .await;
         }
+        let dkg_start = runtime_context.current();
+        if let Some(dkg_state) = &self.dkg_state {
+            if let Err(error) = dkg_state
+                .apply_block(
+                    &mut batch,
+                    Height::new(context.height),
+                    reshare_log,
+                    runtime_context,
+                )
+                .await
+            {
+                debug!(?error, "invalid authenticated DKG public-state update");
+                if let Some(digest) = context.block_digest {
+                    events.discard_block(digest).await;
+                }
+                return None;
+            }
+        }
+        metrics
+            .dkg_state_duration
+            .observe_between(dkg_start, runtime_context.current());
         metrics
             .apply_transactions_duration
             .observe_between(apply_start, runtime_context.current());
@@ -476,6 +530,31 @@ where
             dkg,
         )
     }
+
+    /// Construct an application whose DKG progress is authenticated by QMDB.
+    #[allow(clippy::too_many_arguments)]
+    pub fn with_authenticated_dkg(
+        submitter: MempoolHandle<R::Transaction>,
+        max_block_transactions: usize,
+        consensus: Ext,
+        dkg: DkgMailbox<R::Transaction, Ext>,
+        dkg_state: DkgState,
+        applied_height: SharedAppliedHeight,
+        genesis_state: StateCommitment,
+        genesis_payload: sha256::Digest,
+    ) -> Self {
+        let mut application = Self::with_consensus(
+            submitter,
+            max_block_transactions,
+            consensus,
+            Some(dkg),
+            applied_height,
+            genesis_state,
+            genesis_payload,
+        );
+        application.dkg_state = Some(dkg_state);
+        application
+    }
 }
 
 impl<R> Application<R>
@@ -590,7 +669,7 @@ where
 
 impl<E, R, Ext, Events> StatefulApplication<E> for Application<R, Ext, Events>
 where
-    E: Rng + Spawner + Metrics + Clock + Storage + BufferPooler,
+    E: Rng + CryptoRng + Spawner + Metrics + Clock + Storage + BufferPooler,
     R: Runtime + Clone + Send + Sync + 'static,
     R::Transaction: PoolTransaction + Sync,
     Ext: ConsensusExtension + Sync,
@@ -612,7 +691,7 @@ where
 
     async fn propose(
         &mut self,
-        (runtime_context, context): (E, Self::Context),
+        (mut runtime_context, context): (E, Self::Context),
         ancestry: impl futures::Stream<Item = std::sync::Arc<Self::Block>> + Send,
         batches: <Self::Databases as DatabaseSet<E>>::Unmerkleized,
         input: &mut Self::InputProvider,
@@ -631,19 +710,23 @@ where
             parent.height.next(),
             timestamp,
         );
+        // Obtain the optional dealer log before executing the speculative
+        // state batch so that it is committed by the resulting state root.
+        let reshare_log = match &mut self.dkg {
+            Some(dkg) => dkg.act().await,
+            None => None,
+        };
         let (transactions, extension, merkleized) = self
             .build_proposal_state(
-                Some((&runtime_context, &metrics)),
+                Some(&metrics),
                 batches,
                 execution_context,
                 candidates,
+                reshare_log.as_ref(),
+                &mut runtime_context,
             )
             .await?;
         let state_range = Self::state_range(&merkleized);
-        let reshare_log = match &mut self.dkg {
-            Some(dkg) => dkg.act().await,
-            None => None,
-        };
         let block = Block::new(
             context,
             parent.digest(),
@@ -662,7 +745,7 @@ where
 
     async fn verify(
         &mut self,
-        (runtime_context, _): (E, Self::Context),
+        (mut runtime_context, _): (E, Self::Context),
         ancestry: impl futures::Stream<Item = std::sync::Arc<Self::Block>> + Send,
         batches: <Self::Databases as DatabaseSet<E>>::Unmerkleized,
     ) -> Option<<Self::Databases as DatabaseSet<E>>::Merkleized> {
@@ -700,12 +783,13 @@ where
         let execution_context = Self::block_runtime_context(&block);
         let merkleized = self
             .execute_block(
-                &runtime_context,
+                &mut runtime_context,
                 &metrics,
                 batches,
                 execution_context,
                 &block.transactions,
                 &block.extension,
+                block.reshare_log.as_ref(),
                 &NoopEventConsumer,
                 false,
             )
@@ -719,7 +803,7 @@ where
 
     async fn apply(
         &mut self,
-        (runtime_context, _): (E, Self::Context),
+        (mut runtime_context, _): (E, Self::Context),
         block: &Self::Block,
         batches: <Self::Databases as DatabaseSet<E>>::Unmerkleized,
     ) -> <Self::Databases as DatabaseSet<E>>::Merkleized {
@@ -728,12 +812,13 @@ where
         let events = self.events.clone();
         let merkleized = self
             .execute_block(
-                &runtime_context,
+                &mut runtime_context,
                 &metrics,
                 batches,
                 execution_context,
                 &block.transactions,
                 &block.extension,
+                block.reshare_log.as_ref(),
                 &events,
                 true,
             )

chain/src/consensus.rs

@@ -1,4 +1,5 @@
 mod dkg;
+pub mod dkg_state;
 mod extension;
 
 pub use dkg::{dkg_reporters, DkgActor, DkgMailbox, DkgReporters};