Randomly assign triples to a node after creation (#412)

* Randomly assign triples to a node after creation

* Use a stable hash function

While it's not strictly necessary right now (every node should be
running the same version) it will save people trouble should they ever
want to loosen that restriction/if nodes are running subtley different
binaries.

* Remove unnecessary type

* First version of the testing code (it's dog slow)

* Working test

* Expanded test slightly

* Clippy fix

* Added mailbox debuggers, but disabled them

node/Cargo.toml

@@ -22,6 +22,7 @@ clap = { version = "4.2", features = ["derive", "env"] }
 google-secretmanager1 = "5"
 hex = "0.4.3"
 hkdf = "0.12.4"
+highway = "1.1.0"
 k256 = { version = "0.13.1", features = ["sha256", "ecdsa", "serde"] }
 local-ip-address = "0.5.4"
 rand = "0.8"
@@ -45,3 +46,6 @@ near-sdk = "5.0.0-alpha.1"
 
 mpc-contract = { path = "../contract" }
 mpc-keys = { path = "../keys" }
+
+[dev-dependencies]
+itertools = "0.12.0"

node/src/protocol/triple.rs

@@ -4,6 +4,8 @@ use crate::types::TripleProtocol;
 use crate::util::AffinePointExt;
 use cait_sith::protocol::{Action, InitializationError, Participant, ProtocolError};
 use cait_sith::triples::{TriplePub, TripleShare};
+use highway::{HighwayHash, HighwayHasher};
+use k256::elliptic_curve::group::GroupEncoding;
 use k256::Secp256k1;
 use std::collections::hash_map::Entry;
 use std::collections::{HashMap, VecDeque};
@@ -21,28 +23,20 @@ pub struct Triple {
     pub public: TriplePub<Secp256k1>,
 }
 
-/// An ongoing triple generator.
-pub struct TripleGenerator {
-    /// Ongoing cait-sith triple generation protocol.
-    pub protocol: TripleProtocol,
-    /// Whether this triple generation was initiated by the current node.
-    pub mine: bool,
-}
-
 /// Abstracts how triples are generated by providing a way to request a new triple that will be
 /// complete some time in the future and a way to take an already generated triple.
 pub struct TripleManager {
     /// Completed unspent triples
-    triples: HashMap<TripleId, Triple>,
+    pub triples: HashMap<TripleId, Triple>,
     /// Ongoing triple generation protocols
-    generators: HashMap<TripleId, TripleGenerator>,
+    pub generators: HashMap<TripleId, TripleProtocol>,
     /// List of triple ids generation of which was initiated by the current node.
-    mine: VecDeque<TripleId>,
+    pub mine: VecDeque<TripleId>,
 
-    participants: Vec<Participant>,
-    me: Participant,
-    threshold: usize,
-    epoch: u64,
+    pub participants: Vec<Participant>,
+    pub me: Participant,
+    pub threshold: usize,
+    pub epoch: u64,
 }
 
 impl TripleManager {
@@ -90,13 +84,7 @@ impl TripleManager {
                 self.threshold,
             )?,
         ));
-        self.generators.insert(
-            id,
-            TripleGenerator {
-                protocol,
-                mine: true,
-            },
-        );
+        self.generators.insert(id, protocol);
         Ok(())
     }
 
@@ -157,13 +145,10 @@ impl TripleManager {
                             self.threshold,
                         )?,
                     ));
-                    let generator = e.insert(TripleGenerator {
-                        protocol,
-                        mine: false,
-                    });
-                    Ok(Some(&mut generator.protocol))
+                    let generator = e.insert(protocol);
+                    Ok(Some(generator))
                 }
-                Entry::Occupied(e) => Ok(Some(&mut e.into_mut().protocol)),
+                Entry::Occupied(e) => Ok(Some(e.into_mut())),
             }
         }
     }
@@ -177,7 +162,7 @@ impl TripleManager {
         let mut result = Ok(());
         self.generators.retain(|id, generator| {
             loop {
-                let mut protocol = match generator.protocol.write() {
+                let mut protocol = match generator.write() {
                     Ok(protocol) => protocol,
                     Err(err) => {
                         tracing::error!(
@@ -232,18 +217,38 @@ impl TripleManager {
                             big_c = ?output.1.big_c.to_base58(),
                             "completed triple generation"
                         );
-                        self.triples.insert(
-                            *id,
-                            Triple {
-                                id: *id,
-                                share: output.0,
-                                public: output.1,
-                            },
-                        );
 
-                        if generator.mine {
+                        let triple = Triple {
+                            id: *id,
+                            share: output.0,
+                            public: output.1,
+                        };
+
+                        // After creation the triple is assigned to a random node, which is NOT necessarily the one that initiated it's creation
+                        let triple_is_mine = {
+                            // This is an entirely unpredictable value to all participants because it's a combination of big_c_i
+                            // It is the same value across all participants
+                            let big_c = triple.public.big_c;
+
+                            // We turn this into a u64 in a way not biased to the structure of the byte serialisation so we hash it
+                            // We use Highway Hash because the DefaultHasher doesn't guarantee a consistent output across versions
+                            let entropy =
+                                HighwayHasher::default().hash64(&big_c.to_bytes()) as usize;
+
+                            let num_participants = self.participants.len();
+                            // This has a *tiny* bias towards lower indexed participants, they're up to (1 + num_participants / u64::MAX)^2 times more likely to be selected
+                            // This is acceptably small that it will likely never result in a biased selection happening
+                            let triple_owner = self.participants[entropy % num_participants];
+
+                            triple_owner == self.me
+                        };
+
+                        if triple_is_mine {
                             self.mine.push_back(*id);
                         }
+
+                        self.triples.insert(*id, triple);
+
                         // Do not retain the protocol
                         break false;
                     }
@@ -253,3 +258,167 @@ impl TripleManager {
         result.map(|_| messages)
     }
 }
+
+#[cfg(test)]
+mod test {
+    use std::{collections::HashMap, fs::OpenOptions, ops::Range};
+
+    use crate::protocol::message::TripleMessage;
+    use cait_sith::protocol::{InitializationError, Participant, ProtocolError};
+    use itertools::multiunzip;
+    use std::io::prelude::*;
+
+    use super::TripleManager;
+
+    struct TestManagers {
+        managers: Vec<TripleManager>,
+    }
+
+    impl TestManagers {
+        fn new(number: u32) -> Self {
+            let range = 0..number;
+            // Self::wipe_mailboxes(range.clone());
+            let participants: Vec<Participant> = range.map(Participant::from).collect();
+            let managers = participants
+                .iter()
+                .map(|me| TripleManager::new(participants.clone(), *me, number as usize, 0))
+                .collect();
+            TestManagers { managers }
+        }
+
+        fn generate(&mut self, index: usize) -> Result<(), InitializationError> {
+            self.managers[index].generate()
+        }
+
+        fn poke(&mut self, index: usize) -> Result<bool, ProtocolError> {
+            let mut quiet = true;
+            let messages = self.managers[index].poke()?;
+            for (
+                participant,
+                ref tm @ TripleMessage {
+                    id, from, ref data, ..
+                },
+            ) in messages
+            {
+                // Self::debug_mailbox(participant.into(), &tm);
+                quiet = false;
+                let participant_i: u32 = participant.into();
+                let manager = &mut self.managers[participant_i as usize];
+                if let Some(protocol) = manager.get_or_generate(id).unwrap() {
+                    let mut protocol = protocol.write().unwrap();
+                    protocol.message(from, data.to_vec());
+                } else {
+                    println!("Tried to write to completed mailbox {:?}", tm);
+                }
+            }
+            Ok(quiet)
+        }
+
+        #[allow(unused)]
+        fn wipe_mailboxes(mailboxes: Range<u32>) {
+            for m in mailboxes {
+                let mut file = OpenOptions::new()
+                    .write(true)
+                    .append(false)
+                    .create(true)
+                    .open(format!("{}.csv", m))
+                    .unwrap();
+                write!(file, "").unwrap();
+            }
+        }
+
+        // This allows you to see what each node is recieving and when
+        #[allow(unused)]
+        fn debug_mailbox(participant: u32, TripleMessage { id, from, data, .. }: &TripleMessage) {
+            let mut file = OpenOptions::new()
+                .write(true)
+                .append(true)
+                .open(format!("{}.csv", participant))
+                .unwrap();
+
+            writeln!(file, "'{id}, {from:?}, {}", hex::encode(data)).unwrap();
+        }
+
+        fn poke_until_quiet(&mut self) -> Result<(), ProtocolError> {
+            loop {
+                let mut quiet = true;
+                for i in 0..self.managers.len() {
+                    let poke = self.poke(i)?;
+                    quiet = quiet && poke;
+                }
+                if quiet {
+                    return Ok(());
+                }
+            }
+        }
+    }
+
+    // TODO: This test currently takes 22 seconds on my machine, which is much slower than it should be
+    // Improve this before we make more similar tests
+    #[test]
+    fn happy_triple_generation() {
+        let mut tm = TestManagers::new(5);
+
+        const M: usize = 2;
+        const N: usize = M + 3;
+        // Generate 5 triples
+        for _ in 0..M {
+            tm.generate(0).unwrap();
+        }
+        tm.poke_until_quiet().unwrap();
+        tm.generate(1).unwrap();
+        tm.generate(2).unwrap();
+        tm.generate(4).unwrap();
+
+        tm.poke_until_quiet().unwrap();
+
+        let inputs = tm
+            .managers
+            .into_iter()
+            .map(|m| (m.my_len(), m.len(), m.generators, m.triples));
+
+        let (my_lens, lens, generators, mut triples): (Vec<_>, Vec<_>, Vec<_>, Vec<_>) =
+            multiunzip(inputs);
+
+        assert_eq!(
+            my_lens.iter().sum::<usize>(),
+            N,
+            "There should be {N} owned completed triples in total",
+        );
+
+        for l in lens {
+            assert_eq!(l, N, "All nodes should have {N} completed triples")
+        }
+
+        // This passes, but we don't have deterministic entropy or enough triples
+        // to ensure that it will no coincidentally fail
+        // TODO: deterministic entropy for testing
+        // assert_ne!(
+        //     my_lens,
+        //     vec![M, 1, 1, 0, 1],
+        //     "The nodes that started the triple don't own it"
+        // );
+
+        for g in generators.iter() {
+            assert!(g.is_empty(), "There are no triples still being generated")
+        }
+
+        assert_ne!(
+            triples.len(),
+            1,
+            "The number of triples is not 1 before deduping"
+        );
+
+        triples.dedup_by_key(|kv| {
+            kv.iter_mut()
+                .map(|(id, triple)| (*id, (triple.id, triple.public.clone())))
+                .collect::<HashMap<_, _>>()
+        });
+
+        assert_eq!(
+            triples.len(),
+            1,
+            "All triple IDs and public parts are identical"
+        )
+    }
+}