Only enqueue DKG bundles if the decrypter has capacity.

Author: twittner

timeboost-sequencer/src/decrypt.rs

@@ -10,7 +10,7 @@ use multisig::{Committee, CommitteeId, PublicKey};
 use parking_lot::RwLock;
 use sailfish::types::{CommitteeVec, Evidence, Round, RoundNumber};
 use serde::{Deserialize, Serialize};
-use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet, VecDeque};
+use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
 use std::result::Result as StdResult;
 use std::sync::Arc;
 use timeboost_crypto::prelude::{LabeledDkgDecKey, ThresholdEncKeyCell, Vess, Vss};
@@ -232,18 +232,13 @@ impl Decrypter {
         Ok(())
     }
 
-    /// Send the received DKG bundles to worker
-    pub async fn enqueue_dkg(
-        &self,
-        pending_dkgs: &mut VecDeque<DkgBundle>,
-    ) -> StdResult<(), DecrypterDown> {
-        while let Some(b) = pending_dkgs.pop_front() {
-            self.worker_tx
-                .send(Command::Dkg(b))
-                .await
-                .map_err(|_| DecrypterDown(()))?;
-            debug!(node = %self.label, "enqueued one dkg bundle");
-        }
+    /// Send the received DKG bundle to worker
+    pub async fn enqueue_dkg(&self, dkg: DkgBundle) -> StdResult<(), DecrypterDown> {
+        self.worker_tx
+            .send(Command::Dkg(dkg))
+            .await
+            .map_err(|_| DecrypterDown(()))?;
+        debug!(node = %self.label, "enqueued one dkg bundle");
         Ok(())
     }
 
@@ -1621,11 +1616,12 @@ mod tests {
 
         // enqueuing them all to decrypters
         for decrypter in decrypters.iter_mut() {
-            let mut pending_dkgs = dkg_bundles.clone();
-            decrypter
-                .enqueue_dkg(&mut pending_dkgs)
-                .await
-                .expect("DKG bundles should be enqueued successfully");
+            for dkg in dkg_bundles.clone() {
+                decrypter
+                    .enqueue_dkg(dkg)
+                    .await
+                    .expect("DKG bundles should be enqueued successfully");
+            }
         }
     }
 

timeboost-sequencer/src/lib.rs

@@ -261,23 +261,22 @@ impl Task {
     // processing its actions continues unhindered.
     async fn go(mut self) -> Result<()> {
         let mut pending = None;
-        let mut pending_dkgs = VecDeque::new();
+        let mut dkg_bundles = VecDeque::new();
         let mut candidates = Candidates::new();
 
         if !self.sailfish.is_init() {
             let actions = self.sailfish.init();
             candidates = self.execute(actions).await?;
+        }
 
-            // DKG dealing generation
-            // TODO: move/copy to main loop when resharing
-            if let Some(bundle) = self.decrypter.gen_dkg_bundle() {
-                self.bundles.add_bundles(once(BundleVariant::Dkg(bundle)));
-            }
+        // TODO: move/copy to main loop when resharing
+        if let Some(bundle) = self.decrypter.gen_dkg_bundle() {
+            self.bundles.add_bundles(once(BundleVariant::Dkg(bundle)));
         }
 
         loop {
             if pending.is_none() {
-                while let Some(ilist) = self.next_inclusion(&mut candidates, &mut pending_dkgs) {
+                while let Some(ilist) = self.next_inclusion(&mut candidates, &mut dkg_bundles) {
                     if !self.decrypter.has_capacity() {
                         pending = Some(ilist);
                         break;
@@ -288,12 +287,15 @@ impl Task {
                 }
             }
 
-            // always sync DKG bundles
-            self.next_dkg(&mut candidates, &mut pending_dkgs);
-            if !pending_dkgs.is_empty() {
-                tracing::debug!(num_bundles = %pending_dkgs.len(), "enqueuing dkg bundles");
-                if let Err(err) = self.decrypter.enqueue_dkg(&mut pending_dkgs).await {
-                    error!(node = %self.label, %err, "dkg enqueue error");
+            if pending.is_none() {
+                while let Some(dkg) = dkg_bundles.pop_front() {
+                    if !self.decrypter.has_capacity() {
+                        dkg_bundles.push_front(dkg);
+                        break;
+                    }
+                    if let Err(err) = self.decrypter.enqueue_dkg(dkg).await {
+                        error!(node = %self.label, %err, "dkg enqueue error");
+                    }
                 }
             }
 
@@ -435,7 +437,7 @@ impl Task {
             // preprocess the candidate list to pull out the DKG bundles first
             for cl in lists.iter() {
                 if let Some(dkg) = cl.dkg_bundle() {
-                    pending_dkgs.push_back(dkg);
+                    pending_dkgs.push_back(dkg.clone());
                 }
             }
             // then process it to construct the next inclusion list
@@ -456,19 +458,6 @@ impl Task {
         }
         None
     }
-
-    /// Handle candidate lists and "pull" out the DKG bundles, it won't touch or drop or consume
-    /// regular/priority bundles, but only consume/take the DKG bundles inside `candidates` then
-    /// append them to `pending_dkgs`.
-    fn next_dkg(&mut self, candidates: &mut Candidates, pending_dkgs: &mut VecDeque<DkgBundle>) {
-        for (_, _, list) in candidates.iter_mut() {
-            for cl in list.iter_mut() {
-                if let Some(dkg) = cl.take_dkg_bundle() {
-                    pending_dkgs.push_back(dkg);
-                }
-            }
-        }
-    }
 }
 
 #[derive(Debug, thiserror::Error)]

timeboost-types/src/candidate_list.rs

@@ -1,5 +1,5 @@
+use std::ops::Deref;
 use std::sync::Arc;
-use std::{ops::Deref, sync::RwLock};
 
 use bytes::{BufMut, Bytes, BytesMut};
 use committable::{Commitment, Committable, RawCommitmentBuilder};
@@ -20,7 +20,7 @@ struct Inner {
     index: DelayedInboxIndex,
     priority: Vec<SignedPriorityBundle>,
     regular: Vec<Bundle>,
-    dkg: RwLock<Option<DkgBundle>>,
+    dkg: Option<DkgBundle>,
 }
 
 #[derive(Debug)]
@@ -29,7 +29,7 @@ pub struct Builder {
     index: DelayedInboxIndex,
     priority: Vec<SignedPriorityBundle>,
     regular: Vec<Bundle>,
-    dkg: RwLock<Option<DkgBundle>>,
+    dkg: Option<DkgBundle>,
 }
 
 impl Builder {
@@ -44,7 +44,7 @@ impl Builder {
     }
 
     pub fn with_dkg(mut self, d: Option<DkgBundle>) -> Self {
-        self.dkg = RwLock::new(d);
+        self.dkg = d;
         self
     }
 
@@ -69,7 +69,7 @@ impl CandidateList {
             index: i.into(),
             regular: Vec::new(),
             priority: Vec::new(),
-            dkg: RwLock::new(None),
+            dkg: None,
         }
     }
 
@@ -112,26 +112,8 @@ impl CandidateList {
         self.0.index
     }
 
-    pub fn dkg_bundle(&self) -> Option<DkgBundle> {
-        match self.0.dkg.read() {
-            Ok(dkg) => dkg.clone(),
-            Err(poisoned) => {
-                tracing::error!("RwLock on DkgBundle poisoned during read");
-                let dkg = poisoned.into_inner();
-                dkg.clone()
-            }
-        }
-    }
-
-    pub fn take_dkg_bundle(&self) -> Option<DkgBundle> {
-        match self.0.dkg.write() {
-            Ok(mut dkg) => dkg.take(),
-            Err(poisoned) => {
-                tracing::error!("RwLock on DkgBundle poisoned during write");
-                let mut dkg = poisoned.into_inner();
-                dkg.take()
-            }
-        }
+    pub fn dkg_bundle(&self) -> Option<&DkgBundle> {
+        self.0.dkg.as_ref()
     }
 }
 
@@ -140,7 +122,7 @@ impl Committable for CandidateList {
         let mut builder = RawCommitmentBuilder::new("CandidateList")
             .u64_field("time", self.0.time.into())
             .u64_field("index", self.0.index.into())
-            .optional("dkg", &self.dkg_bundle())
+            .optional("dkg", &self.0.dkg)
             .u64_field("priority", self.0.priority.len() as u64);
         builder = self
             .0