Update the flow for completed signing processes

Now, an on-chain transaction exists. This resolves some ambiguities and
provides greater coordination.

coordinator/src/main.rs

@@ -190,7 +190,6 @@ pub async fn scan_tributaries<D: Db, Pro: Processors, P: P2p>(
   }
 }
 
-#[allow(clippy::type_complexity)]
 pub async fn heartbeat_tributaries<D: Db, P: P2p>(
   p2p: P,
   tributaries: Arc<RwLock<Tributaries<D, P>>>,
@@ -217,7 +216,6 @@ pub async fn heartbeat_tributaries<D: Db, P: P2p>(
   }
 }
 
-#[allow(clippy::type_complexity)]
 pub async fn handle_p2p<D: Db, P: P2p>(
   our_key: <Ristretto as Ciphersuite>::G,
   p2p: P,
@@ -364,7 +362,6 @@ pub async fn publish_transaction<D: Db, P: P2p>(
   }
 }
 
-#[allow(clippy::type_complexity)]
 pub async fn handle_processors<D: Db, Pro: Processors, P: P2p>(
   mut db: D,
   key: Zeroizing<<Ristretto as Ciphersuite>::F>,

coordinator/src/tributary/mod.rs

@@ -232,6 +232,7 @@ pub enum Transaction {
 
   SignPreprocess(SignData),
   SignShare(SignData),
+  SignCompleted([u8; 32], Vec<u8>, Signed),
 }
 
 impl ReadWrite for Transaction {
@@ -304,6 +305,19 @@ impl ReadWrite for Transaction {
       6 => SignData::read(reader).map(Transaction::SignPreprocess),
       7 => SignData::read(reader).map(Transaction::SignShare),
 
+      8 => {
+        let mut plan = [0; 32];
+        reader.read_exact(&mut plan)?;
+
+        let mut tx_len = [0];
+        reader.read_exact(&mut tx_len)?;
+        let mut tx = vec![0; usize::from(tx_len[0])];
+        reader.read_exact(&mut tx)?;
+
+        let signed = Signed::read(reader)?;
+        Ok(Transaction::SignCompleted(plan, tx, signed))
+      }
+
       _ => Err(io::Error::new(io::ErrorKind::Other, "invalid transaction type")),
     }
   }
@@ -376,6 +390,13 @@ impl ReadWrite for Transaction {
         writer.write_all(&[7])?;
         data.write(writer)
       }
+      Transaction::SignCompleted(plan, tx, signed) => {
+        writer.write_all(&[8])?;
+        writer.write_all(plan)?;
+        writer.write_all(&[u8::try_from(tx.len()).expect("tx hash length exceed 255 bytes")])?;
+        writer.write_all(tx)?;
+        signed.write(writer)
+      }
     }
   }
 }
@@ -394,6 +415,7 @@ impl TransactionTrait for Transaction {
 
       Transaction::SignPreprocess(data) => TransactionKind::Signed(&data.signed),
       Transaction::SignShare(data) => TransactionKind::Signed(&data.signed),
+      Transaction::SignCompleted(_, _, signed) => TransactionKind::Signed(signed),
     }
   }
 
@@ -451,6 +473,7 @@ impl Transaction {
 
         Transaction::SignPreprocess(ref mut data) => &mut data.signed,
         Transaction::SignShare(ref mut data) => &mut data.signed,
+        Transaction::SignCompleted(_, _, ref mut signed) => signed,
       }
     }
 

coordinator/src/tributary/scanner.rs

@@ -297,6 +297,7 @@ async fn handle_block<D: Db, Pro: Processors>(
               .await;
           }
         }
+        Transaction::SignCompleted(_, _, _) => todo!(),
       }
 
       TributaryDb::<D>::handle_event(&mut txn, hash, event_id);