diff --git a/src/signature_aggregator/signer.rs b/src/signature_aggregator/signer.rs index 2fa07fa..8ce1301 100644 --- a/src/signature_aggregator/signer.rs +++ b/src/signature_aggregator/signer.rs @@ -193,7 +193,6 @@ impl Display for LeaderState { round, signatures, packages, - waiting_for, .. } => { let committed: BTreeSet<_> = packages @@ -201,7 +200,7 @@ impl Display for LeaderState { .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, + 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, + ) { + 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 { @@ -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 {