Route the SubstrateBlock message, which is the last Tributary transaction type

coordinator/src/substrate/mod.rs

@@ -197,6 +197,7 @@ async fn handle_batch_and_burns<Pro: Processor>(
             serai_time: block.time().unwrap(),
             coin_latest_finalized_block,
           },
+          block: block.number(),
           key: serai
             .get_keys(ValidatorSet { network, session: Session(0) }) // TODO2
             .await?

coordinator/src/tests/tributary.rs

@@ -76,7 +76,7 @@ fn serialize_transaction() {
     OsRng.fill_bytes(&mut ext_block);
     test_read_write(Transaction::ExternalBlock(ext_block));
   }
-  test_read_write(Transaction::SeraiBlock(OsRng.next_u64()));
+  test_read_write(Transaction::SubstrateBlock(OsRng.next_u64()));
 
   test_read_write(Transaction::BatchPreprocess(random_sign_data(&mut OsRng)));
   test_read_write(Transaction::BatchShare(random_sign_data(&mut OsRng)));

coordinator/src/tributary/db.rs

@@ -1,3 +1,5 @@
+use std::io::Read;
+
 use ciphersuite::{group::GroupEncoding, Ciphersuite, Ristretto};
 
 pub use serai_db::*;
@@ -33,6 +35,22 @@ impl<D: Db> TributaryDb<D> {
     getter.get(Self::batch_id_key(&genesis, ext_block)).map(|bytes| bytes.try_into().unwrap())
   }
 
+  fn plan_ids_key(genesis: &[u8], block: u64) -> Vec<u8> {
+    Self::tributary_key(b"plan_ids", [genesis, block.to_le_bytes().as_ref()].concat())
+  }
+  pub fn plan_ids<G: Get>(getter: &G, genesis: [u8; 32], block: u64) -> Option<Vec<[u8; 32]>> {
+    getter.get(Self::plan_ids_key(&genesis, block)).map(|bytes| {
+      let mut res = vec![];
+      let mut bytes_ref: &[u8] = bytes.as_ref();
+      while !bytes_ref.is_empty() {
+        let mut id = [0; 32];
+        bytes_ref.read_exact(&mut id).unwrap();
+        res.push(id);
+      }
+      res
+    })
+  }
+
   fn recognized_id_key(label: &'static str, genesis: [u8; 32], id: [u8; 32]) -> Vec<u8> {
     Self::tributary_key(b"recognized", [label.as_bytes(), genesis.as_ref(), id.as_ref()].concat())
   }

coordinator/src/tributary/mod.rs

@@ -162,7 +162,7 @@ pub enum Transaction {
   ExternalBlock([u8; 32]),
   // When a Serai block is finalized, with the contained batches, we can allow the associated plan
   // IDs
-  SeraiBlock(u64),
+  SubstrateBlock(u64),
 
   BatchPreprocess(SignData),
   BatchShare(SignData),
@@ -232,7 +232,7 @@ impl ReadWrite for Transaction {
       3 => {
         let mut block = [0; 8];
         reader.read_exact(&mut block)?;
-        Ok(Transaction::SeraiBlock(u64::from_le_bytes(block)))
+        Ok(Transaction::SubstrateBlock(u64::from_le_bytes(block)))
       }
 
       4 => SignData::read(reader).map(Transaction::BatchPreprocess),
@@ -291,7 +291,7 @@ impl ReadWrite for Transaction {
         writer.write_all(block)
       }
 
-      Transaction::SeraiBlock(block) => {
+      Transaction::SubstrateBlock(block) => {
         writer.write_all(&[3])?;
         writer.write_all(&block.to_le_bytes())
       }
@@ -324,7 +324,7 @@ impl TransactionTrait for Transaction {
       Transaction::DkgShares(_, _, signed) => TransactionKind::Signed(signed),
 
       Transaction::ExternalBlock(_) => TransactionKind::Provided("external"),
-      Transaction::SeraiBlock(_) => TransactionKind::Provided("serai"),
+      Transaction::SubstrateBlock(_) => TransactionKind::Provided("serai"),
 
       Transaction::BatchPreprocess(data) => TransactionKind::Signed(&data.signed),
       Transaction::BatchShare(data) => TransactionKind::Signed(&data.signed),

coordinator/src/tributary/scanner.rs

@@ -173,11 +173,10 @@ async fn handle_block<D: Db, Pro: Processor, P: P2p>(
           // If we didn't provide this transaction, we should halt until we do
           // If we provided a distinct transaction, we should error
           // If we did provide this transaction, we should've set the batch ID for the block
-          let batch_id =
-            TributaryDb::<D>::batch_id(&txn, tributary.genesis(), block).expect(concat!(
-              "synced a tributary block finalizing a block in a provided transaction despite ",
-              "us not providing that transaction"
-            ));
+          let batch_id = TributaryDb::<D>::batch_id(&txn, tributary.genesis(), block).expect(
+            "synced a tributary block finalizing a external block in a provided transaction \
+            despite us not providing that transaction",
+          );
 
           TributaryDb::<D>::recognize_id(
             &mut txn,
@@ -186,7 +185,17 @@ async fn handle_block<D: Db, Pro: Processor, P: P2p>(
             batch_id,
           );
         }
-        Transaction::SeraiBlock(..) => todo!(),
+
+        Transaction::SubstrateBlock(block) => {
+          let plan_ids = TributaryDb::<D>::plan_ids(&txn, tributary.genesis(), block).expect(
+            "synced a tributary block finalizing a substrate block in a provided transaction \
+            despite us not providing that transaction",
+          );
+
+          for id in plan_ids {
+            TributaryDb::<D>::recognize_id(&mut txn, Zone::Sign.label(), tributary.genesis(), id);
+          }
+        }
 
         Transaction::BatchPreprocess(data) => {
           if let Some(preprocesses) = handle(
@@ -291,6 +300,8 @@ pub async fn handle_new_blocks<D: Db, Pro: Processor, P: P2p>(
     return;
   }
 
+  // TODO: Only handle n blocks at a time.
+  // This may load the entire chain into RAM as-is.
   let mut blocks = vec![tributary.block(&latest).unwrap()];
   while blocks.last().unwrap().parent() != *last_block {
     blocks.push(tributary.block(&blocks.last().unwrap().parent()).unwrap());