Fix signer liveness by completing on quorum signatures

src/signature_aggregator/signer.rs

@@ -193,15 +193,14 @@ impl Display for LeaderState {
                 round,
                 signatures,
                 packages,
-                waiting_for,
                 ..
             } => {
                 let committed: BTreeSet<_> = packages
                     .values()
                     .flat_map(|p| p.signing_commitments().keys())
                     .collect();
                 let signed: BTreeSet<_> = signatures.iter().map(|s| s.identifier).collect();
-                f.write_fmt(format_args!("{{role=Leader,state=CollectingSignatures,round={},committed={:?},signed={:?},waiting_for={:?}}}", round, committed, signed, waiting_for))
+                f.write_fmt(format_args!("{{role=Leader,state=CollectingSignatures,round={},committed={:?},signed={:?}}}", round, committed, signed))
             }
         }
     }
@@ -365,19 +364,26 @@ impl Signer {
     }
 
     async fn finish_round_early(&mut self) -> Result<()> {
-        let round = match &self.state {
-            SignerState::Leader(LeaderState::CollectingCommitments { round, .. }) => round,
-            SignerState::Leader(LeaderState::CollectingSignatures { round, .. }) => round,
-            _ => return Ok(()),
-        };
-        info!(round, "Finishing round before collecting enough signatures");
-        let now = SystemTime::now();
-        let payload = SignedEntries {
-            timestamp: now,
-            synthetics: vec![],
-            generics: vec![],
-        };
-        self.publish(self.id.clone(), payload).await
+        match &self.state {
+            SignerState::Leader(LeaderState::CollectingCommitments { round, .. }) => {
+                info!(
+                    round,
+                    "Finishing round before collecting enough commitments; publishing empty payload"
+                );
+                let now = SystemTime::now();
+                let payload = SignedEntries {
+                    timestamp: now,
+                    synthetics: vec![],
+                    generics: vec![],
+                };
+                self.publish(self.id.clone(), payload).await
+            }
+            SignerState::Leader(LeaderState::CollectingSignatures { round, .. }) => {
+                info!(round, "Finishing round before collecting enough signatures");
+                self.complete_round().await
+            }
+            _ => Ok(()),
+        }
     }
 
     async fn begin_round(&mut self, round: String) -> Result<()> {
@@ -816,10 +822,9 @@ impl Signer {
         let SignerState::Leader(LeaderState::CollectingSignatures {
             round: curr_round,
             round_span,
-            price_data,
             signatures,
-            packages,
             waiting_for,
+            ..
         }) = &mut self.state
         else {
             // We're not a leader, or we're not waiting for signatures
@@ -848,6 +853,23 @@ impl Signer {
         round_span
             .in_scope(|| info!("Collected all signatures, signing and completing this round"));
 
+        self.complete_round().await
+    }
+
+    async fn complete_round(&mut self) -> Result<()> {
+        let SignerState::Leader(LeaderState::CollectingSignatures {
+            round,
+            price_data,
+            signatures,
+            packages,
+            ..
+        }) = &mut self.state
+        else {
+            // We're not a leader, or we're not waiting for signatures
+            return Ok(());
+        };
+        let round = round.clone();
+
         // We've finally collected all the signatures we need! Now just aggregate them
         let mut signature_maps = BTreeMap::new();
         for signature in signatures {
@@ -1055,7 +1077,10 @@ mod tests {
     use num_rational::BigRational;
     use num_traits::FromPrimitive as _;
     use rand::thread_rng;
-    use tokio::sync::{mpsc, watch};
+    use tokio::sync::{
+        mpsc::{self, error::TryRecvError},
+        watch,
+    };
     use tracing::{Instrument, Span};
 
     use crate::{
@@ -1065,7 +1090,7 @@ mod tests {
             Validity,
         },
         raft::RaftLeader,
-        signature_aggregator::signer::SignerMessage,
+        signature_aggregator::signer::{SignerMessage, SignerMessageData},
     };
 
     use super::{Signer, SignerEvent};
@@ -1074,6 +1099,7 @@ mod tests {
         id: NodeId,
         signer: Signer,
         receiver: NetworkReceiver<SignerMessage>,
+        withhold_signature: bool,
     }
 
     impl SignerOrchestrator {
@@ -1082,6 +1108,11 @@ mod tests {
         }
         async fn process_incoming_messages(&mut self) {
             while let Some(message) = self.receiver.try_recv().await {
+                if self.withhold_signature
+                    && let SignerMessageData::RequestSignature { .. } = &message.data.data
+                {
+                    continue;
+                }
                 self.signer
                     .process(SignerEvent::Message(
                         message.from,
@@ -1129,6 +1160,7 @@ mod tests {
                     id,
                     signer,
                     receiver,
+                    withhold_signature: false,
                 }
             })
             .collect();
@@ -1175,6 +1207,20 @@ mod tests {
         }
     }
 
+    async fn start_second_round(
+        signers: &mut [SignerOrchestrator],
+        network: &mut TestNetwork<SignerMessage>,
+    ) {
+        signers[0]
+            .process(SignerEvent::RoundStarted("round2".into()))
+            .await;
+        while network.drain().await {
+            for signer in signers.iter_mut() {
+                signer.process_incoming_messages().await;
+            }
+        }
+    }
+
     fn assert_round_complete(
         payload_source: &mut mpsc::Receiver<(NodeId, SignedEntries)>,
     ) -> Result<SignedEntries> {
@@ -1184,6 +1230,14 @@ mod tests {
         }
     }
 
+    fn assert_round_incomplete(payload_source: &mut mpsc::Receiver<(NodeId, SignedEntries)>) {
+        match payload_source.try_recv() {
+            Ok(_) => panic!("round unexpectedly completed"),
+            Err(TryRecvError::Disconnected) => panic!("network unexpected shut down"),
+            Err(TryRecvError::Empty) => {}
+        }
+    }
+
     fn synthetic_data(
         synthetic: &str,
         price: f64,
@@ -1527,6 +1581,50 @@ mod tests {
         Ok(())
     }
 
+    #[tokio::test]
+    async fn should_take_what_we_can_get_when_round_ends() -> Result<()> {
+        let leader_prices = PriceData {
+            synthetics: vec![
+                synthetic_data("FIRST", 3.50, &["A"], &[2], 1),
+                synthetic_data("SECOND", 3.50, &["A"], &[2], 1),
+            ],
+            generics: vec![],
+        };
+        let follower1_prices = PriceData {
+            synthetics: vec![
+                synthetic_data("FIRST", 3.50, &["A"], &[2], 1),
+                synthetic_data("SECOND", 3.50, &["A"], &[3], 2),
+            ],
+            generics: vec![],
+        };
+        let follower2_prices = PriceData {
+            synthetics: vec![
+                synthetic_data("FIRST", 3.50, &["A"], &[3], 2),
+                synthetic_data("SECOND", 3.50, &["A"], &[2], 1),
+            ],
+            generics: vec![],
+        };
+
+        // Follower #1 agrees with the leader on one price, follower #2 on another.
+        // Follower #2 is nefarious and will not actually send a signature when asked.
+        // it does what it can with the signatures it has.
+        let (mut signers, mut network, mut payload_source) =
+            construct_signers(2, vec![leader_prices, follower1_prices, follower2_prices]).await?;
+        signers[2].withhold_signature = true;
+
+        // The leader waits as long as it can for follower #2...
+        assign_roles(&mut signers).await;
+        run_round(&mut signers, &mut network).await;
+        assert_round_incomplete(&mut payload_source);
+
+        // But when the round ends, it publishes the one payload it can.
+        start_second_round(&mut signers, &mut network).await;
+        let signed_entries = assert_round_complete(&mut payload_source)?;
+        assert_eq!(signed_entries.synthetics.len(), 1);
+
+        Ok(())
+    }
+
     #[tokio::test]
     async fn should_end_round_early_when_new_round_begins() -> Result<()> {
         let leader_prices = PriceData {