Pass the latest active key to the Block's scan function

Effectively necessary for networks on which we utilize account abstraction in
order to know what key to associate the received coins with.

processor/scanner/src/db.rs

@@ -28,6 +28,7 @@ struct SeraiKeyDbEntry<K: Borshy> {
   key: K,
 }
 
+#[derive(Clone)]
 pub(crate) struct SeraiKey<K> {
   pub(crate) key: K,
   pub(crate) stage: LifetimeStage,

processor/scanner/src/eventuality/mod.rs

@@ -273,6 +273,18 @@ impl<D: Db, S: ScannerFeed, Sch: Scheduler<S>> ContinuallyRan for EventualityTas
         log::debug!("checking eventuality completions in block: {} ({b})", hex::encode(block.id()));
 
         let (keys, keys_with_stages) = self.keys_and_keys_with_stages(b);
+        let latest_active_key = {
+          let mut keys_with_stages = keys_with_stages.clone();
+          loop {
+            // Use the most recent key
+            let (key, stage) = keys_with_stages.pop().unwrap();
+            // Unless this key is active, but not yet reporting
+            if stage == LifetimeStage::ActiveYetNotReporting {
+              continue;
+            }
+            break key;
+          }
+        };
 
         let mut txn = self.db.txn();
 
@@ -307,7 +319,7 @@ impl<D: Db, S: ScannerFeed, Sch: Scheduler<S>> ContinuallyRan for EventualityTas
           }
 
           // Fetch all non-External outputs
-          let mut non_external_outputs = block.scan_for_outputs(key.key);
+          let mut non_external_outputs = block.scan_for_outputs(latest_active_key, key.key);
           non_external_outputs.retain(|output| output.kind() != OutputType::External);
           // Drop any outputs less than the dust limit
           non_external_outputs.retain(|output| {

processor/scanner/src/lib.rs

@@ -46,11 +46,11 @@ pub(crate) fn sort_outputs<K: GroupEncoding, A: Address, O: ReceivedOutput<K, A>
 
 /// Extension traits around Block.
 pub(crate) trait BlockExt: Block {
-  fn scan_for_outputs(&self, key: Self::Key) -> Vec<Self::Output>;
+  fn scan_for_outputs(&self, latest_active_key: Self::Key, key: Self::Key) -> Vec<Self::Output>;
 }
 impl<B: Block> BlockExt for B {
-  fn scan_for_outputs(&self, key: Self::Key) -> Vec<Self::Output> {
-    let mut outputs = self.scan_for_outputs_unordered(key);
+  fn scan_for_outputs(&self, latest_active_key: Self::Key, key: Self::Key) -> Vec<Self::Output> {
+    let mut outputs = self.scan_for_outputs_unordered(latest_active_key, key);
     outputs.sort_by(sort_outputs);
     outputs
   }

processor/scanner/src/scan/mod.rs

@@ -122,6 +122,19 @@ impl<D: Db, S: ScannerFeed> ContinuallyRan for ScanTask<D, S> {
         let keys = ScannerGlobalDb::<S>::active_keys_as_of_next_to_scan_for_outputs_block(&txn)
           .expect("scanning for a blockchain without any keys set");
 
+        let latest_active_key = {
+          let mut keys = keys.clone();
+          loop {
+            // Use the most recent key
+            let key = keys.pop().unwrap();
+            // Unless this key is active, but not yet reporting
+            if key.stage == LifetimeStage::ActiveYetNotReporting {
+              continue;
+            }
+            break key.key;
+          }
+        };
+
         // The scan data for this block
         let mut scan_data = SenderScanData {
           block_number: b,
@@ -157,7 +170,7 @@ impl<D: Db, S: ScannerFeed> ContinuallyRan for ScanTask<D, S> {
 
         // Scan for each key
         for key in &keys {
-          for output in block.scan_for_outputs(key.key) {
+          for output in block.scan_for_outputs(latest_active_key, key.key) {
             assert_eq!(output.key(), key.key);
 
             /*