Have the scanner's report task ensure handovers only occur if Batchs are valid

This is incomplete at this time. The logic is fine, but needs to be moved to a
distinct location to handle singular blocks which produce multiple Batches.

processor/scanner/src/report/db.rs

@@ -25,6 +25,10 @@ create_db!(
   ScannerReport {
     // The next block to potentially report
     NextToPotentiallyReportBlock: () -> u64,
+
+    // The last session to sign a Batch and their first Batch signed
+    LastSessionToSignBatchAndFirstBatch: () -> (Session, u32),
+
     // The next Batch ID to use
     NextBatchId: () -> u32,
 
@@ -43,6 +47,19 @@ pub(crate) struct ReturnInformation<S: ScannerFeed> {
 
 pub(crate) struct ReportDb<S: ScannerFeed>(PhantomData<S>);
 impl<S: ScannerFeed> ReportDb<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_to_potentially_report_block(
     txn: &mut impl DbTxn,
     next_to_potentially_report_block: u64,

processor/scanner/src/report/mod.rs

@@ -5,13 +5,14 @@ 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::{Returnable, ScannerGlobalDb, InInstructionData, ScanToReportDb, Batches, BatchesToSign},
   scan::next_to_scan_for_outputs_block,
-  ScannerFeed, KeyFor,
+  substrate, ScannerFeed, KeyFor,
 };
 
 mod db;
@@ -92,6 +93,7 @@ impl<D: Db, S: ScannerFeed> ContinuallyRan for ReportTask<D, S> {
           external_key_for_session_to_sign_batch,
           returnable_in_instructions: in_instructions,
         } = ScanToReportDb::<S>::recv_in_instructions(&mut txn, block_number);
+
         let notable = ScannerGlobalDb::<S>::is_block_notable(&txn, block_number);
         if !notable {
           assert!(in_instructions.is_empty(), "block wasn't notable yet had InInstructions");
@@ -101,6 +103,62 @@ impl<D: Db, S: ScannerFeed> ContinuallyRan for ReportTask<D, S> {
           let network = S::NETWORK;
           let mut batch_id = ReportDb::<S>::acquire_batch_id(&mut txn);
 
+          /*
+            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) =
+              ReportDb::<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 this txn, restoring the Batch to be worked on in the future
+                drop(txn);
+                return Ok(block_number > next_to_potentially_report);
+              }
+            }
+
+            // If this is the handover Batch, update the last session to sign a Batch
+            if handover_batch {
+              ReportDb::<S>::set_last_session_to_sign_batch_and_first_batch(
+                &mut txn,
+                session_to_sign_batch,
+                batch_id,
+              );
+            }
+          }
+
+          // TODO: The above code doesn't work if we end up with two Batches (the handover and the
+          // following) within this one Block due to Batch size limits
+
           // start with empty batch
           let mut batches = vec![Batch { network, id: batch_id, instructions: vec![] }];
           // We also track the return information for the InInstructions within a Batch in case

processor/scanner/src/substrate/db.rs

@@ -40,6 +40,12 @@ pub(crate) enum Action<S: ScannerFeed> {
   QueueBurns(Vec<OutInstructionWithBalance>),
 }
 
+create_db!(
+  ScannerSubstrate {
+    LastAcknowledgedBatch: () -> u32,
+  }
+);
+
 db_channel!(
   ScannerSubstrate {
     Actions: () -> ActionEncodable,
@@ -48,6 +54,14 @@ db_channel!(
 
 pub(crate) struct SubstrateDb<S: ScannerFeed>(PhantomData<S>);
 impl<S: ScannerFeed> SubstrateDb<S> {
+  pub(crate) fn last_acknowledged_batch(getter: &impl Get) -> Option<u32> {
+    LastAcknowledgedBatch::get(getter)
+  }
+
+  pub(crate) fn set_last_acknowledged_batch(txn: &mut impl DbTxn, id: u32) {
+    LastAcknowledgedBatch::set(txn, &id)
+  }
+
   pub(crate) fn queue_acknowledge_batch(
     txn: &mut impl DbTxn,
     batch_id: u32,

processor/scanner/src/substrate/mod.rs

@@ -1,6 +1,6 @@
 use core::{marker::PhantomData, future::Future};
 
-use serai_db::{DbTxn, Db};
+use serai_db::{Get, DbTxn, Db};
 
 use serai_coins_primitives::{OutInstruction, OutInstructionWithBalance};
 use serai_validator_sets_primitives::Session;
@@ -14,6 +14,9 @@ use crate::{
 mod db;
 use db::*;
 
+pub(crate) fn last_acknowledged_batch<S: ScannerFeed>(getter: &impl Get) -> Option<u32> {
+  SubstrateDb::<S>::last_acknowledged_batch(getter)
+}
 pub(crate) fn queue_acknowledge_batch<S: ScannerFeed>(
   txn: &mut impl DbTxn,
   batch_id: u32,
@@ -99,6 +102,7 @@ impl<D: Db, S: ScannerFeed> ContinuallyRan for SubstrateTask<D, S> {
               "batch acknowledged on-chain was distinct"
             );
 
+            SubstrateDb::<S>::set_last_acknowledged_batch(&mut txn, batch_id);
             AcknowledgedBatches::send(
               &mut txn,
               &external_key_for_session_to_sign_batch.0,