Have Tributary's add_transaction return a proper error

Modifies main.rs to properly handle the returned error.

coordinator/src/main.rs

@@ -31,7 +31,9 @@ use tokio::{
   time::sleep,
 };
 
-use ::tributary::{ProvidedError, TransactionKind, TransactionTrait, Block, Tributary};
+use ::tributary::{
+  ProvidedError, TransactionKind, TransactionError, TransactionTrait, Block, Tributary,
+};
 
 mod tributary;
 use crate::tributary::{
@@ -150,10 +152,16 @@ async fn publish_signed_transaction<D: Db, P: P2p>(
       .await
       .expect("we don't have a nonce, meaning we aren't a participant on this tributary"),
   ) {
-    // TODO: Assert if we didn't create a valid transaction
     // We need to return a proper error here to enable that, due to a race condition around
     // multiple publications
-    tributary.add_transaction(tx).await;
+    match tributary.add_transaction(tx.clone()).await {
+      Ok(_) => {}
+      // Some asynchonicity if InvalidNonce, assumed safe to deterministic nonces
+      Err(TransactionError::InvalidNonce) => {
+        log::warn!("publishing TX {tx:?} returned InvalidNonce. was it already added?")
+      }
+      Err(e) => panic!("created an invalid transaction: {e:?}"),
+    }
   }
 }
 
@@ -630,10 +638,10 @@ async fn handle_processor_message<D: Db, P: P2p>(
         }
         TransactionKind::Unsigned => {
           log::trace!("publishing unsigned transaction {}", hex::encode(tx.hash()));
-          // Ignores the result since we can't differentiate already in-mempool from
-          // already on-chain from invalid
-          // TODO: Don't ignore the result
-          tributary.add_transaction(tx).await;
+          match tributary.add_transaction(tx.clone()).await {
+            Ok(_) => {}
+            Err(e) => panic!("created an invalid unsigned transaction: {e:?}"),
+          }
         }
         TransactionKind::Signed(_) => {
           log::trace!("getting next nonce for Tributary TX in response to processor message");

coordinator/src/tests/tributary/dkg.rs

@@ -56,7 +56,7 @@ async fn dkg_test() {
 
   // Publish all commitments but one
   for (i, tx) in txs.iter().enumerate().skip(1) {
-    assert!(tributaries[i].1.add_transaction(tx.clone()).await);
+    assert_eq!(tributaries[i].1.add_transaction(tx.clone()).await, Ok(true));
   }
 
   // Wait until these are included
@@ -104,7 +104,7 @@ async fn dkg_test() {
 
   // Publish the last commitment
   let block_before_tx = tributaries[0].1.tip().await;
-  assert!(tributaries[0].1.add_transaction(txs[0].clone()).await);
+  assert_eq!(tributaries[0].1.add_transaction(txs[0].clone()).await, Ok(true));
   wait_for_tx_inclusion(&tributaries[0].1, block_before_tx, txs[0].hash()).await;
   sleep(Duration::from_secs(Tributary::<MemDb, Transaction, LocalP2p>::block_time().into())).await;
 
@@ -181,7 +181,7 @@ async fn dkg_test() {
 
   let block_before_tx = tributaries[0].1.tip().await;
   for (i, tx) in txs.iter().enumerate().skip(1) {
-    assert!(tributaries[i].1.add_transaction(tx.clone()).await);
+    assert_eq!(tributaries[i].1.add_transaction(tx.clone()).await, Ok(true));
   }
   for tx in txs.iter().skip(1) {
     wait_for_tx_inclusion(&tributaries[0].1, block_before_tx, tx.hash()).await;
@@ -205,7 +205,7 @@ async fn dkg_test() {
 
   // Publish the final set of shares
   let block_before_tx = tributaries[0].1.tip().await;
-  assert!(tributaries[0].1.add_transaction(txs[0].clone()).await);
+  assert_eq!(tributaries[0].1.add_transaction(txs[0].clone()).await, Ok(true));
   wait_for_tx_inclusion(&tributaries[0].1, block_before_tx, txs[0].hash()).await;
   sleep(Duration::from_secs(Tributary::<MemDb, Transaction, LocalP2p>::block_time().into())).await;
 
@@ -296,7 +296,7 @@ async fn dkg_test() {
   }
   let block_before_tx = tributaries[0].1.tip().await;
   for (i, tx) in txs.iter().enumerate() {
-    assert!(tributaries[i].1.add_transaction(tx.clone()).await);
+    assert_eq!(tributaries[i].1.add_transaction(tx.clone()).await, Ok(true));
   }
   for tx in txs.iter() {
     wait_for_tx_inclusion(&tributaries[0].1, block_before_tx, tx.hash()).await;

coordinator/src/tests/tributary/tx.rs

@@ -43,7 +43,7 @@ async fn tx_test() {
     Transaction::DkgCommitments(attempt, commitments.clone(), Transaction::empty_signed());
   tx.sign(&mut OsRng, spec.genesis(), &key, 0);
 
-  assert!(tributaries[sender].1.add_transaction(tx.clone()).await);
+  assert_eq!(tributaries[sender].1.add_transaction(tx.clone()).await, Ok(true));
   let included_in = wait_for_tx_inclusion(&tributaries[sender].1, block_before_tx, tx.hash()).await;
   // Also sleep for the block time to ensure the block is synced around before we run checks on it
   sleep(Duration::from_secs(Tributary::<MemDb, Transaction, LocalP2p>::block_time().into())).await;