Add support for multiple orderings in Provided

Necessary as our Tributary chains needed to agree when a Serai block has
occurred, and when a Monero block has occurred. Since those could happen at the
same time, some validators may put SeraiBlock before ExternalBlock and vice
versa, causing a chain halt. Now they can have distinct ordering queues.

coordinator/src/tributary/mod.rs

@@ -322,9 +322,8 @@ impl TransactionTrait for Transaction {
       Transaction::DkgCommitments(_, _, signed) => TransactionKind::Signed(signed),
       Transaction::DkgShares(_, _, signed) => TransactionKind::Signed(signed),
 
-      // TODO: Tributary requires these be perfectly ordered, yet they have two separate clocks
-      Transaction::ExternalBlock(_) => TransactionKind::Provided,
-      Transaction::SeraiBlock(_) => TransactionKind::Provided,
+      Transaction::ExternalBlock(_) => TransactionKind::Provided("external"),
+      Transaction::SeraiBlock(_) => TransactionKind::Provided("serai"),
 
       Transaction::BatchPreprocess(data) => TransactionKind::Signed(&data.signed),
       Transaction::BatchShare(data) => TransactionKind::Signed(&data.signed),

coordinator/tributary/src/block.rs

@@ -1,4 +1,7 @@
-use std::{io, collections::HashMap};
+use std::{
+  io,
+  collections::{VecDeque, HashMap},
+};
 
 use thiserror::Error;
 
@@ -101,7 +104,10 @@ impl<T: Transaction> Block<T> {
   pub(crate) fn new(parent: [u8; 32], provided: Vec<T>, mempool: Vec<T>) -> Self {
     let mut txs = provided;
     for tx in mempool {
-      assert!(tx.kind() != TransactionKind::Provided, "provided transaction entered mempool");
+      assert!(
+        !matches!(tx.kind(), TransactionKind::Provided(_)),
+        "provided transaction entered mempool"
+      );
       txs.push(tx);
     }
 
@@ -144,7 +150,7 @@ impl<T: Transaction> Block<T> {
     &self,
     genesis: [u8; 32],
     last_block: [u8; 32],
-    locally_provided: &[[u8; 32]],
+    mut locally_provided: HashMap<&'static str, VecDeque<T>>,
     mut next_nonces: HashMap<<Ristretto as Ciphersuite>::G, u32>,
   ) -> Result<(), BlockError> {
     if self.serialize().len() > BLOCK_SIZE_LIMIT {
@@ -157,18 +163,19 @@ impl<T: Transaction> Block<T> {
 
     let mut found_non_provided = false;
     let mut txs = Vec::with_capacity(self.transactions.len());
-    for (i, tx) in self.transactions.iter().enumerate() {
+    for tx in self.transactions.iter() {
       txs.push(tx.hash());
 
-      if tx.kind() == TransactionKind::Provided {
+      if let TransactionKind::Provided(order) = tx.kind() {
         if found_non_provided {
           Err(BlockError::ProvidedAfterNonProvided)?;
         }
 
-        let Some(local) = locally_provided.get(i) else {
+        let Some(local) =
+          locally_provided.get_mut(order).and_then(|deque| deque.pop_front()) else {
             Err(BlockError::NonLocalProvided(txs.pop().unwrap()))?
           };
-        if txs.last().unwrap() != local {
+        if tx != &local {
           Err(BlockError::DistinctProvided)?;
         }
 

coordinator/tributary/src/blockchain.rs

@@ -125,7 +125,7 @@ impl<D: Db, T: Transaction> Blockchain<D, T> {
   pub(crate) fn build_block(&mut self) -> Block<T> {
     let block = Block::new(
       self.tip,
-      self.provided.transactions.iter().cloned().collect(),
+      self.provided.transactions.values().flatten().cloned().collect(),
       self.mempool.block(&self.next_nonces),
     );
     // build_block should not return invalid blocks
@@ -137,7 +137,7 @@ impl<D: Db, T: Transaction> Blockchain<D, T> {
     block.verify(
       self.genesis,
       self.tip,
-      &self.provided.transactions.iter().map(Transaction::hash).collect::<Vec<_>>(),
+      self.provided.transactions.clone(),
       self.next_nonces.clone(),
     )
   }
@@ -164,8 +164,8 @@ impl<D: Db, T: Transaction> Blockchain<D, T> {
 
     for tx in &block.transactions {
       match tx.kind() {
-        TransactionKind::Provided => {
-          self.provided.complete(&mut txn, tx.hash());
+        TransactionKind::Provided(order) => {
+          self.provided.complete(&mut txn, order, tx.hash());
         }
         TransactionKind::Unsigned => {}
         TransactionKind::Signed(Signed { signer, nonce, .. }) => {

coordinator/tributary/src/provided.rs

@@ -24,7 +24,7 @@ pub struct ProvidedTransactions<D: Db, T: Transaction> {
   db: D,
   genesis: [u8; 32],
 
-  pub(crate) transactions: VecDeque<T>,
+  pub(crate) transactions: HashMap<&'static str, VecDeque<T>>,
 }
 
 impl<D: Db, T: Transaction> ProvidedTransactions<D, T> {
@@ -36,21 +36,25 @@ impl<D: Db, T: Transaction> ProvidedTransactions<D, T> {
   }
 
   pub(crate) fn new(db: D, genesis: [u8; 32]) -> Self {
-    let mut res = ProvidedTransactions { db, genesis, transactions: VecDeque::new() };
+    let mut res = ProvidedTransactions { db, genesis, transactions: HashMap::new() };
 
     let currently_provided = res.db.get(res.current_provided_key()).unwrap_or(vec![]);
     let mut i = 0;
     while i < currently_provided.len() {
-      res.transactions.push_back(
-        T::read::<&[u8]>(
-          &mut res
-            .db
-            .get(res.transaction_key(&currently_provided[i .. (i + 32)]))
-            .unwrap()
-            .as_ref(),
+      let tx = T::read::<&[u8]>(
+        &mut res.db.get(res.transaction_key(&currently_provided[i .. (i + 32)])).unwrap().as_ref(),
       )
-        .unwrap(),
-      );
+      .unwrap();
+
+      let TransactionKind::Provided(order) = tx.kind() else {
+        panic!("provided transaction saved to disk wasn't provided");
+      };
+
+      if res.transactions.get(order).is_none() {
+        res.transactions.insert(order, VecDeque::new());
+      }
+      res.transactions.get_mut(order).unwrap().push_back(tx);
+
       i += 32;
     }
 
@@ -59,9 +63,9 @@ impl<D: Db, T: Transaction> ProvidedTransactions<D, T> {
 
   /// Provide a transaction for inclusion in a block.
   pub(crate) fn provide(&mut self, tx: T) -> Result<(), ProvidedError> {
-    if tx.kind() != TransactionKind::Provided {
-      Err(ProvidedError::NotProvided)?;
-    }
+    let TransactionKind::Provided(order) = tx.kind() else {
+      Err(ProvidedError::NotProvided)?
+    };
 
     match verify_transaction(&tx, self.genesis, &mut HashMap::new()) {
       Ok(()) => {}
@@ -83,17 +87,39 @@ impl<D: Db, T: Transaction> ProvidedTransactions<D, T> {
     txn.put(current_provided_key, currently_provided);
     txn.commit();
 
-    self.transactions.push_back(tx);
+    if self.transactions.get(order).is_none() {
+      self.transactions.insert(order, VecDeque::new());
+    }
+    self.transactions.get_mut(order).unwrap().push_back(tx);
     Ok(())
   }
 
   /// Complete a provided transaction, no longer proposing it nor voting for its validity.
-  pub(crate) fn complete(&mut self, txn: &mut D::Transaction<'_>, tx: [u8; 32]) {
-    assert_eq!(self.transactions.pop_front().unwrap().hash(), tx);
+  pub(crate) fn complete(
+    &mut self,
+    txn: &mut D::Transaction<'_>,
+    order: &'static str,
+    tx: [u8; 32],
+  ) {
+    assert_eq!(self.transactions.get_mut(order).unwrap().pop_front().unwrap().hash(), tx);
 
     let current_provided_key = self.current_provided_key();
     let mut currently_provided = txn.get(&current_provided_key).unwrap();
-    assert_eq!(&currently_provided.drain(.. 32).collect::<Vec<_>>(), &tx);
+
+    // Find this TX's hash
+    let mut i = 0;
+    loop {
+      if currently_provided[i .. (i + 32)] == tx {
+        assert_eq!(&currently_provided.drain(i .. (i + 32)).collect::<Vec<_>>(), &tx);
+        break;
+      }
+
+      i += 32;
+      if i >= currently_provided.len() {
+        panic!("couldn't find completed TX in currently provided");
+      }
+    }
+
     txn.put(current_provided_key, currently_provided);
   }
 }

coordinator/tributary/src/tests/block.rs

@@ -69,7 +69,7 @@ fn empty_block() {
   const GENESIS: [u8; 32] = [0xff; 32];
   const LAST: [u8; 32] = [0x01; 32];
   Block::<NonceTransaction>::new(LAST, vec![], vec![])
-    .verify(GENESIS, LAST, &[], HashMap::new())
+    .verify(GENESIS, LAST, HashMap::new(), HashMap::new())
     .unwrap();
 }
 
@@ -89,7 +89,7 @@ fn duplicate_nonces() {
     let res = Block::new(LAST, vec![], mempool).verify(
       GENESIS,
       LAST,
-      &[],
+      HashMap::new(),
       HashMap::from([(<Ristretto as Ciphersuite>::G::identity(), 0)]),
     );
     if i == 1 {

coordinator/tributary/src/tests/blockchain.rs

@@ -1,4 +1,4 @@
-use std::collections::VecDeque;
+use std::collections::{VecDeque, HashMap};
 
 use zeroize::Zeroizing;
 use rand::{RngCore, rngs::OsRng};
@@ -187,10 +187,10 @@ fn provided_transaction() {
   assert_eq!(txs.provide(tx.clone()), Err(ProvidedError::AlreadyProvided));
   assert_eq!(
     ProvidedTransactions::<_, ProvidedTransaction>::new(db.clone(), genesis).transactions,
-    VecDeque::from([tx.clone()]),
+    HashMap::from([("provided", VecDeque::from([tx.clone()]))]),
   );
   let mut txn = db.txn();
-  txs.complete(&mut txn, tx.hash());
+  txs.complete(&mut txn, "provided", tx.hash());
   txn.commit();
   assert!(ProvidedTransactions::<_, ProvidedTransaction>::new(db.clone(), genesis)
     .transactions

coordinator/tributary/src/tests/transaction/mod.rs

@@ -47,7 +47,7 @@ impl ReadWrite for ProvidedTransaction {
 
 impl Transaction for ProvidedTransaction {
   fn kind(&self) -> TransactionKind<'_> {
-    TransactionKind::Provided
+    TransactionKind::Provided("provided")
   }
 
   fn hash(&self) -> [u8; 32] {

coordinator/tributary/src/transaction.rs

@@ -65,12 +65,16 @@ impl ReadWrite for Signed {
 pub enum TransactionKind<'a> {
   /// This tranaction should be provided by every validator, in an exact order.
   ///
+  /// The contained static string names the orderer to use. This allows two distinct provided
+  /// transaction kinds, without a synchronized order, to be ordered within their own kind without
+  /// requiring ordering with each other.
+  ///
   /// The only malleability is in when this transaction appears on chain. The block producer will
   /// include it when they have it. Block verification will fail for validators without it.
   ///
   /// If a supermajority of validators still produce a commit for a block with a provided
   /// transaction which isn't locally held, the chain will sleep until it is locally provided.
-  Provided,
+  Provided(&'static str),
 
   /// An unsigned transaction, only able to be included by the block producer.
   Unsigned,
@@ -114,7 +118,7 @@ pub(crate) fn verify_transaction<T: Transaction>(
   tx.verify()?;
 
   match tx.kind() {
-    TransactionKind::Provided => {}
+    TransactionKind::Provided(_) => {}
     TransactionKind::Unsigned => {}
     TransactionKind::Signed(Signed { signer, nonce, signature }) => {
       if let Some(next_nonce) = next_nonces.get(signer) {