Implement generating and using for State Sync

chain-signatures/node/src/protocol/consensus.rs

@@ -97,6 +97,9 @@ impl<G: Governance> ConsensusProtocol<G> for StartedState {
                             );
 
                             let threshold = contract_state.threshold;
+                            // Initialize identity for storage; this is an entry point into Running.
+                            ctx.triple_storage.set_me(me);
+                            ctx.presignature_storage.set_me(me);
                             let triple_task = TripleSpawnerTask::run(me, threshold, epoch, ctx);
                             let presign_task = PresignatureSpawnerTask::run(
                                 me,
@@ -399,6 +402,9 @@ impl<G: Governance> ConsensusProtocol<G> for WaitingForConsensusState {
                         return NodeState::WaitingForConsensus(self);
                     };
 
+                    // Initialize identity for storage; this is an entry point into Running.
+                    ctx.triple_storage.set_me(me);
+                    ctx.presignature_storage.set_me(me);
                     let triple_task = TripleSpawnerTask::run(me, self.threshold, self.epoch, ctx);
                     let presign_task = PresignatureSpawnerTask::run(
                         me,

chain-signatures/node/src/protocol/presignature.rs

@@ -375,10 +375,6 @@ impl PresignatureSpawner {
         self.ongoing.contains_key(&id)
     }
 
-    pub async fn contains_used(&self, id: PresignatureId) -> bool {
-        self.presignatures.contains_used(id).await
-    }
-
     /// Returns the number of unspent presignatures available in the manager.
     pub async fn len_generated(&self) -> usize {
         self.presignatures.len_generated().await
@@ -480,7 +476,7 @@ impl PresignatureSpawner {
         // to use the same triple as any other node.
         // TODO: have all this part be a separate task such that finding a pair of triples is done in parallel instead
         // of waiting for storage to respond here.
-        let Some(triples) = self.triples.take_mine(self.me).await else {
+        let Some(triples) = self.triples.take_mine().await else {
             return;
         };
 
@@ -566,9 +562,10 @@ impl PresignatureSpawner {
             "starting protocol to generate a new presignature",
         );
 
-        let Some(slot) = self.presignatures.reserve(id.id).await else {
+        let Some(slot) = self.presignatures.create_slot(id.id, owner).await else {
             return Err(InitializationError::BadParameters(format!(
-                "id collision: presignature_id={id:?}"
+                "presignature {} is already generating, in use, or stored",
+                id.id
             )));
         };
 
@@ -696,6 +693,7 @@ impl PresignatureSpawner {
     ) {
         let mut last_active_warn: Option<Instant> = None;
         let mut stockpile_interval = time::interval(Duration::from_millis(100));
+        stockpile_interval.set_missed_tick_behavior(time::MissedTickBehavior::Skip);
         let mut expiration_interval = tokio::time::interval(Duration::from_secs(1));
         let mut posits = self.msg.subscribe_presignature_posit().await;
 

chain-signatures/node/src/protocol/signature.rs

@@ -328,7 +328,7 @@ impl SignOrganizer {
             let remaining = state.budget.remaining();
             let fetch = tokio::time::timeout(remaining, async {
                 loop {
-                    if let Some(taken) = ctx.presignatures.take_mine(ctx.me).await {
+                    if let Some(taken) = ctx.presignatures.take_mine().await {
                         let Some(holders) = taken.artifact.holders() else {
                             tracing::error!(
                                 id = taken.artifact.id,

chain-signatures/node/src/protocol/sync/mod.rs

@@ -219,7 +219,7 @@ impl SyncTask {
                     match handle.await {
                         Ok(responses) => {
                             // Process sync responses: update artifact participants based on not_found data
-                            if let Err(err) = self.process_sync_responses(responses, me, threshold).await {
+                            if let Err(err) = self.process_sync_responses(responses, threshold).await {
                                 tracing::warn!(?err, "failed to process sync responses");
                             }
                             tracing::debug!(elapsed = ?start.elapsed(), "processed broadcast");
@@ -236,9 +236,8 @@ impl SyncTask {
         }
     }
 
-    // TODO: use reserved values instead. Note that we cannot fetch our own triples via reserved
     async fn new_update(&self, me: Participant) -> Option<SyncUpdate> {
-        let triples = match self.triples.fetch_owned(me).await {
+        let triples = match self.triples.fetch_owned_with_reserved().await {
             Ok(ids) => ids,
             Err(err) => {
                 tracing::warn!(
@@ -248,7 +247,7 @@ impl SyncTask {
                 return None;
             }
         };
-        let presignatures = match self.presignatures.fetch_owned(me).await {
+        let presignatures = match self.presignatures.fetch_owned_with_reserved().await {
             Ok(ids) => ids,
             Err(err) => {
                 tracing::warn!(
@@ -272,7 +271,6 @@ impl SyncTask {
     async fn process_sync_responses(
         &self,
         responses: Vec<(Participant, SyncPeerResponse)>,
-        me: Participant,
         threshold: usize,
     ) -> Result<(), String> {
         for (peer, result) in responses {
@@ -294,13 +292,13 @@ impl SyncTask {
                     // Batch remove peer from all triples and prune
                     let triple_res = self
                         .triples
-                        .remove_holder_and_prune(me, peer, threshold, &response.triples)
+                        .remove_holder_and_prune(peer, threshold, &response.triples)
                         .await;
 
                     // Batch remove peer from all presignatures and prune
                     let presig_res = self
                         .presignatures
-                        .remove_holder_and_prune(me, peer, threshold, &response.presignatures)
+                        .remove_holder_and_prune(peer, threshold, &response.presignatures)
                         .await;
 
                     match (triple_res, presig_res) {

chain-signatures/node/src/protocol/triple.rs

@@ -388,10 +388,6 @@ impl TripleSpawner {
         self.ongoing.contains_key(&id)
     }
 
-    pub async fn contains_used(&self, id: TripleId) -> bool {
-        self.triple_storage.contains_used(id).await
-    }
-
     /// Returns the number of unspent triples assigned to this node.
     pub async fn len_mine(&self) -> usize {
         self.triple_storage.len_by_owner(self.me).await
@@ -528,9 +524,9 @@ impl TripleSpawner {
         timeout: Duration,
     ) -> Result<(), InitializationError> {
         // Check if the `id` is already in the system. Error out and have the next cycle try again.
-        let Some(slot) = self.triple_storage.reserve(id).await else {
+        let Some(slot) = self.triple_storage.create_slot(id, proposer).await else {
             return Err(InitializationError::BadParameters(format!(
-                "id collision: pair_id={id}"
+                "triple {id} is already generating, in use, or stored"
             )));
         };
 
@@ -581,6 +577,7 @@ impl TripleSpawner {
         ongoing_gen_tx: watch::Sender<usize>,
     ) {
         let mut stockpile_interval = tokio::time::interval(Duration::from_millis(100));
+        stockpile_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
         let mut expiration_interval = tokio::time::interval(Duration::from_secs(1));
         let mut posits = self.msg.subscribe_triple_posit().await;