Split reporting Batches to the signer from the Batch test

processor/scanner/src/batch/db.rs

@@ -26,9 +26,6 @@ create_db!(
     // The next block to create batches for
     NextBlockToBatch: () -> u64,
 
-    // The last session to sign a Batch and their first Batch signed
-    LastSessionToSignBatchAndFirstBatch: () -> (Session, u32),
-
     // The next Batch ID to use
     NextBatchId: () -> u32,
 
@@ -47,19 +44,6 @@ pub(crate) struct ReturnInformation<S: ScannerFeed> {
 
 pub(crate) struct BatchDb<S: ScannerFeed>(PhantomData<S>);
 impl<S: ScannerFeed> BatchDb<S> {
-  pub(crate) fn set_last_session_to_sign_batch_and_first_batch(
-    txn: &mut impl DbTxn,
-    session: Session,
-    id: u32,
-  ) {
-    LastSessionToSignBatchAndFirstBatch::set(txn, &(session, id));
-  }
-  pub(crate) fn last_session_to_sign_batch_and_first_batch(
-    getter: &impl Get,
-  ) -> Option<(Session, u32)> {
-    LastSessionToSignBatchAndFirstBatch::get(getter)
-  }
-
   pub(crate) fn set_next_block_to_batch(txn: &mut impl DbTxn, next_block_to_batch: u64) {
     NextBlockToBatch::set(txn, &next_block_to_batch);
   }

processor/scanner/src/batch/mod.rs

@@ -5,17 +5,13 @@ use blake2::{digest::typenum::U32, Digest, Blake2b};
 use scale::Encode;
 use serai_db::{DbTxn, Db};
 
-use serai_validator_sets_primitives::Session;
 use serai_in_instructions_primitives::{MAX_BATCH_SIZE, Batch};
 
 use primitives::{EncodableG, task::ContinuallyRan};
 use crate::{
-  db::{
+  db::{Returnable, ScannerGlobalDb, InInstructionData, ScanToBatchDb, BatchData, BatchToReportDb},
-    Returnable, ScannerGlobalDb, InInstructionData, ScanToBatchDb, BatchData, BatchToReportDb,
-    BatchesToSign,
-  },
   scan::next_to_scan_for_outputs_block,
-  substrate, ScannerFeed, KeyFor,
+  ScannerFeed, KeyFor,
 };
 
 mod db;
@@ -175,76 +171,6 @@ impl<D: Db, S: ScannerFeed> ContinuallyRan for BatchTask<D, S> {
         txn.commit();
       }
 
-      // TODO: This should be its own task. The above doesn't error, doesn't return early, so this
-      // is fine, but this is precarious and would be better as its own task
-      loop {
-        let mut txn = self.db.txn();
-        let Some(BatchData {
-          session_to_sign_batch,
-          external_key_for_session_to_sign_batch,
-          batch,
-        }) = BatchToReportDb::<S>::try_recv_batch(&mut txn)
-        else {
-          break;
-        };
-
-        /*
-          If this is the handover Batch, the first Batch signed by a session which retires the
-          prior validator set, then this should only be signed after the prior validator set's
-          actions are fully validated.
-
-          The new session will only be responsible for signing this Batch if the prior key has
-          retired, successfully completed all its on-external-network actions.
-
-          We check here the prior session has successfully completed all its on-Serai-network
-          actions by ensuring we've validated all Batches expected from it. Only then do we sign
-          the Batch confirming the handover.
-
-          We also wait for the Batch confirming the handover to be accepted on-chain, ensuring we
-          don't verify the prior session's Batches, sign the handover Batch and the following
-          Batch, have the prior session publish a malicious Batch where our handover Batch should
-          be, before our following Batch becomes our handover Batch.
-        */
-        if session_to_sign_batch != Session(0) {
-          // We may have Session(1)'s first Batch be Batch 0 if Session(0) never publishes a
-          // Batch. This is fine as we'll hit the distinct Session check and then set the correct
-          // values into this DB entry. All other sessions must complete the handover process,
-          // which requires having published at least one Batch
-          let (last_session, first_batch) =
-            BatchDb::<S>::last_session_to_sign_batch_and_first_batch(&txn)
-              .unwrap_or((Session(0), 0));
-          // Because this boolean was expanded, we lose short-circuiting. That's fine
-          let handover_batch = last_session != session_to_sign_batch;
-          let batch_after_handover_batch =
-            (last_session == session_to_sign_batch) && ((first_batch + 1) == batch.id);
-          if handover_batch || batch_after_handover_batch {
-            let verified_prior_batch = substrate::last_acknowledged_batch::<S>(&txn)
-              // Since `batch.id = 0` in the Session(0)-never-published-a-Batch case, we don't
-              // check `last_acknowledged_batch >= (batch.id - 1)` but instead this
-              .map(|last_acknowledged_batch| (last_acknowledged_batch + 1) >= batch.id)
-              // We've never verified any Batches
-              .unwrap_or(false);
-            if !verified_prior_batch {
-              // Drop the txn to restore the Batch to report to the DB
-              drop(txn);
-              break;
-            }
-          }
-
-          // If this is the handover Batch, update the last session to sign a Batch
-          if handover_batch {
-            BatchDb::<S>::set_last_session_to_sign_batch_and_first_batch(
-              &mut txn,
-              session_to_sign_batch,
-              batch.id,
-            );
-          }
-        }
-
-        BatchesToSign::send(&mut txn, &external_key_for_session_to_sign_batch.0, &batch);
-        txn.commit();
-      }
-
       // Run dependents if were able to batch any blocks
       Ok(next_block_to_batch <= highest_batchable)
     }

processor/scanner/src/lib.rs

@@ -30,6 +30,8 @@ mod index;
 mod scan;
 /// Task which creates Batches for Substrate.
 mod batch;
+/// Task which reports Batches for signing.
+mod report;
 /// Task which handles events from Substrate once we can.
 mod substrate;
 /// Check blocks for transactions expected to eventually occur.
@@ -380,6 +382,7 @@ impl<S: ScannerFeed> Scanner<S> {
     let index_task = index::IndexTask::new(db.clone(), feed.clone(), start_block).await;
     let scan_task = scan::ScanTask::new(db.clone(), feed.clone(), start_block);
     let batch_task = batch::BatchTask::<_, S>::new(db.clone(), start_block);
+    let report_task = report::ReportTask::<_, S>::new(db.clone());
     let substrate_task = substrate::SubstrateTask::<_, S>::new(db.clone());
     let eventuality_task =
       eventuality::EventualityTask::<_, _, _>::new(db, feed, scheduler, start_block);
@@ -387,6 +390,7 @@ impl<S: ScannerFeed> Scanner<S> {
     let (index_task_def, _index_handle) = Task::new();
     let (scan_task_def, scan_handle) = Task::new();
     let (batch_task_def, batch_handle) = Task::new();
+    let (report_task_def, report_handle) = Task::new();
     let (substrate_task_def, substrate_handle) = Task::new();
     let (eventuality_task_def, eventuality_handle) = Task::new();
 
@@ -394,9 +398,11 @@ impl<S: ScannerFeed> Scanner<S> {
     tokio::spawn(index_task.continually_run(index_task_def, vec![scan_handle.clone()]));
     // Upon scanning a block, creates the batches for it
     tokio::spawn(scan_task.continually_run(scan_task_def, vec![batch_handle]));
-    // Upon creating batches for a block, we do nothing (as the burden is on Substrate which won't
+    // Upon creating batches for a block, we run the report task
-    // be immediately ready)
+    tokio::spawn(batch_task.continually_run(batch_task_def, vec![report_handle]));
-    tokio::spawn(batch_task.continually_run(batch_task_def, vec![]));
+    // Upon reporting the batches for signing, we do nothing (as the burden is on a tributary which
+    // won't immediately yield a result)
+    tokio::spawn(report_task.continually_run(report_task_def, vec![]));
     // Upon handling an event from Substrate, we run the Eventuality task (as it's what's affected)
     tokio::spawn(substrate_task.continually_run(substrate_task_def, vec![eventuality_handle]));
     // Upon handling the Eventualities in a block, we run the scan task as we've advanced the

processor/scanner/src/report/db.rs

@@ -0,0 +1,26 @@
+use serai_db::{Get, DbTxn, create_db};
+
+use serai_validator_sets_primitives::Session;
+
+create_db!(
+  ScannerBatch {
+    // The last session to sign a Batch and their first Batch signed
+    LastSessionToSignBatchAndFirstBatch: () -> (Session, u32),
+  }
+);
+
+pub(crate) struct BatchDb;
+impl BatchDb {
+  pub(crate) fn set_last_session_to_sign_batch_and_first_batch(
+    txn: &mut impl DbTxn,
+    session: Session,
+    id: u32,
+  ) {
+    LastSessionToSignBatchAndFirstBatch::set(txn, &(session, id));
+  }
+  pub(crate) fn last_session_to_sign_batch_and_first_batch(
+    getter: &impl Get,
+  ) -> Option<(Session, u32)> {
+    LastSessionToSignBatchAndFirstBatch::get(getter)
+  }
+}

processor/scanner/src/report/mod.rs

@@ -0,0 +1,105 @@
+use core::{marker::PhantomData, future::Future};
+
+use serai_db::{DbTxn, Db};
+
+use serai_validator_sets_primitives::Session;
+
+use primitives::task::ContinuallyRan;
+use crate::{
+  db::{BatchData, BatchToReportDb, BatchesToSign},
+  substrate, ScannerFeed,
+};
+
+mod db;
+use db::BatchDb;
+
+// This task begins reporting Batches for signing once the pre-requisities are met.
+#[allow(non_snake_case)]
+pub(crate) struct ReportTask<D: Db, S: ScannerFeed> {
+  db: D,
+  _S: PhantomData<S>,
+}
+
+impl<D: Db, S: ScannerFeed> ReportTask<D, S> {
+  pub(crate) fn new(db: D) -> Self {
+    Self { db, _S: PhantomData }
+  }
+}
+
+impl<D: Db, S: ScannerFeed> ContinuallyRan for ReportTask<D, S> {
+  fn run_iteration(&mut self) -> impl Send + Future<Output = Result<bool, String>> {
+    async move {
+      let mut made_progress = false;
+      loop {
+        let mut txn = self.db.txn();
+        let Some(BatchData {
+          session_to_sign_batch,
+          external_key_for_session_to_sign_batch,
+          batch,
+        }) = BatchToReportDb::<S>::try_recv_batch(&mut txn)
+        else {
+          break;
+        };
+
+        /*
+          If this is the handover Batch, the first Batch signed by a session which retires the
+          prior validator set, then this should only be signed after the prior validator set's
+          actions are fully validated.
+
+          The new session will only be responsible for signing this Batch if the prior key has
+          retired, successfully completed all its on-external-network actions.
+
+          We check here the prior session has successfully completed all its on-Serai-network
+          actions by ensuring we've validated all Batches expected from it. Only then do we sign
+          the Batch confirming the handover.
+
+          We also wait for the Batch confirming the handover to be accepted on-chain, ensuring we
+          don't verify the prior session's Batches, sign the handover Batch and the following
+          Batch, have the prior session publish a malicious Batch where our handover Batch should
+          be, before our following Batch becomes our handover Batch.
+        */
+        if session_to_sign_batch != Session(0) {
+          // We may have Session(1)'s first Batch be Batch 0 if Session(0) never publishes a
+          // Batch. This is fine as we'll hit the distinct Session check and then set the correct
+          // values into this DB entry. All other sessions must complete the handover process,
+          // which requires having published at least one Batch
+          let (last_session, first_batch) =
+            BatchDb::last_session_to_sign_batch_and_first_batch(&txn).unwrap_or((Session(0), 0));
+          // Because this boolean was expanded, we lose short-circuiting. That's fine
+          let handover_batch = last_session != session_to_sign_batch;
+          let batch_after_handover_batch =
+            (last_session == session_to_sign_batch) && ((first_batch + 1) == batch.id);
+          if handover_batch || batch_after_handover_batch {
+            let verified_prior_batch = substrate::last_acknowledged_batch::<S>(&txn)
+              // Since `batch.id = 0` in the Session(0)-never-published-a-Batch case, we don't
+              // check `last_acknowledged_batch >= (batch.id - 1)` but instead this
+              .map(|last_acknowledged_batch| (last_acknowledged_batch + 1) >= batch.id)
+              // We've never verified any Batches
+              .unwrap_or(false);
+            if !verified_prior_batch {
+              // Drop the txn to restore the Batch to report to the DB
+              drop(txn);
+              break;
+            }
+          }
+
+          // If this is the handover Batch, update the last session to sign a Batch
+          if handover_batch {
+            BatchDb::set_last_session_to_sign_batch_and_first_batch(
+              &mut txn,
+              session_to_sign_batch,
+              batch.id,
+            );
+          }
+        }
+
+        BatchesToSign::send(&mut txn, &external_key_for_session_to_sign_batch.0, &batch);
+        txn.commit();
+
+        made_progress = true;
+      }
+
+      Ok(made_progress)
+    }
+  }
+}