Decide flow between scan/eventuality/report

Scan now only handles External outputs, with an associated essay going over
why. Scan directly creates the InInstruction (prior planned to be done in
Report), and Eventuality is declared to end up yielding the outputs.

That will require making the Eventuality flow two-stage. One stage to evaluate
existing Eventualities and yield outputs, and one stage to incorporate new
Eventualities before advancing the scan window.

processor/scanner/src/db.rs

@@ -5,25 +5,44 @@ use serai_db::{Get, DbTxn, create_db};
 
 use primitives::{Id, ReceivedOutput, Block, BorshG};
 
-use crate::{ScannerFeed, BlockIdFor, KeyFor, OutputFor};
+use crate::{lifetime::LifetimeStage, ScannerFeed, BlockIdFor, KeyFor, OutputFor};
 
 // The DB macro doesn't support `BorshSerialize + BorshDeserialize` as a bound, hence this.
 trait Borshy: BorshSerialize + BorshDeserialize {}
 impl<T: BorshSerialize + BorshDeserialize> Borshy for T {}
 
 #[derive(BorshSerialize, BorshDeserialize)]
-pub(crate) struct SeraiKey<K: Borshy> {
-  pub(crate) activation_block_number: u64,
-  pub(crate) retirement_block_number: Option<u64>,
+struct SeraiKeyDbEntry<K: Borshy> {
+  activation_block_number: u64,
+  key: K,
+}
+
+pub(crate) struct SeraiKey<K> {
+  pub(crate) stage: LifetimeStage,
   pub(crate) key: K,
 }
 
+pub(crate) struct OutputWithInInstruction<K: GroupEncoding, A, O: ReceivedOutput<K, A>> {
+  output: O,
+  refund_address: A,
+  in_instruction: InInstructionWithBalance,
+}
+
+impl<K: GroupEncoding, A, O: ReceivedOutput<K, A>> OutputWithInInstruction<K, A, O> {
+  fn write(&self, writer: &mut impl io::Write) -> io::Result<()> {
+    self.output.write(writer)?;
+    // TODO self.refund_address.write(writer)?;
+    self.in_instruction.encode_to(writer);
+    Ok(())
+  }
+}
+
 create_db!(
   Scanner {
     BlockId: <I: Id>(number: u64) -> I,
     BlockNumber: <I: Id>(id: I) -> u64,
 
-    ActiveKeys: <K: Borshy>() -> Vec<SeraiKey<K>>,
+    ActiveKeys: <K: Borshy>() -> Vec<SeraiKeyDbEntry<K>>,
 
     // The latest finalized block to appear of a blockchain
     LatestFinalizedBlock: () -> u64,
@@ -80,48 +99,60 @@ impl<S: ScannerFeed> ScannerDb<S> {
     NotableBlock::set(txn, activation_block_number, &());
 
     // Push the key
-    let mut keys: Vec<SeraiKey<BorshG<KeyFor<S>>>> = ActiveKeys::get(txn).unwrap_or(vec![]);
+    let mut keys: Vec<SeraiKeyDbEntry<BorshG<KeyFor<S>>>> = ActiveKeys::get(txn).unwrap_or(vec![]);
     for key_i in &keys {
       if key == key_i.key.0 {
         panic!("queueing a key prior queued");
       }
     }
-    keys.push(SeraiKey {
-      activation_block_number,
-      retirement_block_number: None,
-      key: BorshG(key),
-    });
+    keys.push(SeraiKeyDbEntry { activation_block_number, key: BorshG(key) });
     ActiveKeys::set(txn, &keys);
   }
-  // retirement_block_number is inclusive, so the key will no longer be scanned for as of the
-  // specified block
-  pub(crate) fn retire_key(txn: &mut impl DbTxn, retirement_block_number: u64, key: KeyFor<S>) {
-    let mut keys: Vec<SeraiKey<BorshG<KeyFor<S>>>> =
+  // TODO: This will be called from the Eventuality task yet this field is read by the scan task
+  // We need to write the argument for its safety
+  pub(crate) fn retire_key(txn: &mut impl DbTxn, key: KeyFor<S>) {
+    let mut keys: Vec<SeraiKeyDbEntry<BorshG<KeyFor<S>>>> =
       ActiveKeys::get(txn).expect("retiring key yet no active keys");
 
     assert!(keys.len() > 1, "retiring our only key");
-    for i in 0 .. keys.len() {
-      if key == keys[i].key.0 {
-        keys[i].retirement_block_number = Some(retirement_block_number);
-        ActiveKeys::set(txn, &keys);
-        return;
-      }
-
-      // This is not the key in question, but since it's older, it already should've been queued
-      // for retirement
-      assert!(
-        keys[i].retirement_block_number.is_some(),
-        "older key wasn't retired before newer key"
-      );
-    }
-    panic!("retiring key yet not present in keys")
+    assert_eq!(keys[0].key.0, key, "not retiring the oldest key");
+    keys.remove(0);
+    ActiveKeys::set(txn, &keys);
   }
-  pub(crate) fn keys(getter: &impl Get) -> Option<Vec<SeraiKey<BorshG<KeyFor<S>>>>> {
-    ActiveKeys::get(getter)
+  pub(crate) fn active_keys_as_of_next_to_scan_for_outputs_block(
+    getter: &impl Get,
+  ) -> Option<Vec<SeraiKey<KeyFor<S>>>> {
+    // We don't take this as an argument as we don't keep all historical keys in memory
+    // If we've scanned block 1,000,000, we can't answer the active keys as of block 0
+    let block_number = Self::next_to_scan_for_outputs_block(getter)?;
+
+    let raw_keys: Vec<SeraiKeyDbEntry<BorshG<KeyFor<S>>>> = ActiveKeys::get(getter)?;
+    let mut keys = Vec::with_capacity(2);
+    for i in 0 .. raw_keys.len() {
+      if block_number < raw_keys[i].activation_block_number {
+        continue;
+      }
+      keys.push(SeraiKey {
+        key: raw_keys[i].key.0,
+        stage: LifetimeStage::calculate::<S>(
+          block_number,
+          raw_keys[i].activation_block_number,
+          raw_keys.get(i + 1).map(|key| key.activation_block_number),
+        ),
+      });
+    }
+    assert!(keys.len() <= 2);
+    Some(keys)
   }
 
   pub(crate) fn set_start_block(txn: &mut impl DbTxn, start_block: u64, id: BlockIdFor<S>) {
+    assert!(
+      LatestFinalizedBlock::get(txn).is_none(),
+      "setting start block but prior set start block"
+    );
+
     Self::set_block(txn, start_block, id);
+
     LatestFinalizedBlock::set(txn, &start_block);
     NextToScanForOutputsBlock::set(txn, &start_block);
     // We can receive outputs in this block, but any descending transactions will be in the next
@@ -138,9 +169,10 @@ impl<S: ScannerFeed> ScannerDb<S> {
   }
 
   pub(crate) fn latest_scannable_block(getter: &impl Get) -> Option<u64> {
-    // This is whatever block we've checked the Eventualities of, plus the window length
+    // We can only scan up to whatever block we've checked the Eventualities of, plus the window
+    // length. Since this returns an inclusive bound, we need to subtract 1
     // See `eventuality.rs` for more info
-    NextToCheckForEventualitiesBlock::get(getter).map(|b| b + S::WINDOW_LENGTH)
+    NextToCheckForEventualitiesBlock::get(getter).map(|b| b + S::WINDOW_LENGTH - 1)
   }
 
   pub(crate) fn set_next_to_scan_for_outputs_block(
@@ -187,7 +219,7 @@ impl<S: ScannerFeed> ScannerDb<S> {
     HighestAcknowledgedBlock::get(getter)
   }
 
-  pub(crate) fn set_outputs(txn: &mut impl DbTxn, block_number: u64, outputs: Vec<OutputFor<S>>) {
+  pub(crate) fn set_in_instructions(txn: &mut impl DbTxn, block_number: u64, outputs: Vec<OutputWithInInstruction<KeyFor<S>, AddressFor<S>, OutputFor<S>>>) {
     if outputs.is_empty() {
       return;
     }

processor/scanner/src/eventuality.rs

@@ -109,6 +109,8 @@ impl<D: Db, S: ScannerFeed> ContinuallyRan for EventualityTask<D, S> {
 
       iterated = true;
 
+      // TODO: Not only check/clear eventualities, if this eventuality forwarded an output, queue
+      // it to be reported in however many blocks
       todo!("TODO");
 
       let mut txn = self.db.txn();

processor/scanner/src/lib.rs

@@ -5,10 +5,18 @@ use tokio::sync::mpsc;
 use serai_primitives::{Coin, Amount};
 use primitives::{ReceivedOutput, BlockHeader, Block};
 
+// Logic for deciding where in its lifetime a multisig is.
+mod lifetime;
+
+// Database schema definition and associated functions.
 mod db;
+// Task to index the blockchain, ensuring we don't reorganize finalized blocks.
 mod index;
+// Scans blocks for received coins.
 mod scan;
+/// Check blocks for transactions expected to eventually occur.
 mod eventuality;
+/// Task which reports `Batch`s to Substrate.
 mod report;
 
 /// A feed usable to scan a blockchain.
@@ -16,12 +24,22 @@ mod report;
 /// This defines the primitive types used, along with various getters necessary for indexing.
 #[async_trait::async_trait]
 pub trait ScannerFeed: Send + Sync {
+  /// The amount of confirmations a block must have to be considered finalized.
+  ///
+  /// This value must be at least `1`.
+  const CONFIRMATIONS: u64;
+
   /// The amount of blocks to process in parallel.
   ///
   /// This value must be at least `1`. This value should be the worst-case latency to handle a
   /// block divided by the expected block time.
   const WINDOW_LENGTH: u64;
 
+  /// The amount of blocks which will occur in 10 minutes (approximate).
+  ///
+  /// This value must be at least `1`.
+  const TEN_MINUTES: u64;
+
   /// The representation of a block for this blockchain.
   ///
   /// A block is defined as a consensus event associated with a set of transactions. It is not
@@ -152,6 +170,32 @@ pub(crate) trait ContinuallyRan: Sized {
   }
 }
 
+/// A representation of a scanner.
+pub struct Scanner;
+impl Scanner {
+  /// Create a new scanner.
+  ///
+  /// This will begin its execution, spawning several asynchronous tasks.
+  // TODO: Take start_time and binary search here?
+  pub fn new(start_block: u64) -> Self {
+    todo!("TODO")
+  }
+
+  /// Acknowledge a block.
+  ///
+  /// This means this block was ordered on Serai in relation to `Burn` events, and all validators
+  /// have achieved synchrony on it.
+  pub fn acknowledge_block(
+    &mut self,
+    block_number: u64,
+    key_to_activate: Option<()>,
+    forwarded_outputs: Vec<()>,
+    eventualities_created: Vec<()>,
+  ) {
+    todo!("TODO")
+  }
+}
+
 /*
 #[derive(Clone, Debug)]
 pub enum ScannerEvent<N: Network> {
@@ -172,8 +216,6 @@ pub enum ScannerEvent<N: Network> {
   ),
 }
 
-pub type ScannerEventChannel<N> = mpsc::UnboundedReceiver<ScannerEvent<N>>;
-
 #[derive(Clone, Debug)]
 struct ScannerDb<N: Network, D: Db>(PhantomData<N>, PhantomData<D>);
 impl<N: Network, D: Db> ScannerDb<N, D> {
@@ -184,38 +226,6 @@ impl<N: Network, D: Db> ScannerDb<N, D> {
     getter.get(Self::seen_key(id)).is_some()
   }
 
-  fn outputs_key(block: &<N::Block as Block<N>>::Id) -> Vec<u8> {
-    Self::scanner_key(b"outputs", block.as_ref())
-  }
-  fn save_outputs(
-    txn: &mut D::Transaction<'_>,
-    block: &<N::Block as Block<N>>::Id,
-    outputs: &[N::Output],
-  ) {
-    let mut bytes = Vec::with_capacity(outputs.len() * 64);
-    for output in outputs {
-      output.write(&mut bytes).unwrap();
-    }
-    txn.put(Self::outputs_key(block), bytes);
-  }
-  fn outputs(
-    txn: &D::Transaction<'_>,
-    block: &<N::Block as Block<N>>::Id,
-  ) -> Option<Vec<N::Output>> {
-    let bytes_vec = txn.get(Self::outputs_key(block))?;
-    let mut bytes: &[u8] = bytes_vec.as_ref();
-
-    let mut res = vec![];
-    while !bytes.is_empty() {
-      res.push(N::Output::read(&mut bytes).unwrap());
-    }
-    Some(res)
-  }
-
-  fn scanned_block_key() -> Vec<u8> {
-    Self::scanner_key(b"scanned_block", [])
-  }
-
   fn save_scanned_block(txn: &mut D::Transaction<'_>, block: usize) -> Vec<N::Output> {
     let id = Self::block(txn, block); // It may be None for the first key rotated to
     let outputs =
@@ -255,36 +265,6 @@ impl<N: Network, D: Db> ScannerDb<N, D> {
 }
 
 impl<N: Network, D: Db> ScannerHandle<N, D> {
-  /// Register a key to scan for.
-  pub async fn register_key(
-    &mut self,
-    txn: &mut D::Transaction<'_>,
-    activation_number: usize,
-    key: <N::Curve as Ciphersuite>::G,
-  ) {
-    info!("Registering key {} in scanner at {activation_number}", hex::encode(key.to_bytes()));
-
-    let mut scanner_lock = self.scanner.write().await;
-    let scanner = scanner_lock.as_mut().unwrap();
-    assert!(
-      activation_number > scanner.ram_scanned.unwrap_or(0),
-      "activation block of new keys was already scanned",
-    );
-
-    if scanner.keys.is_empty() {
-      assert!(scanner.ram_scanned.is_none());
-      scanner.ram_scanned = Some(activation_number);
-      assert!(ScannerDb::<N, D>::save_scanned_block(txn, activation_number).is_empty());
-    }
-
-    ScannerDb::<N, D>::register_key(txn, activation_number, key);
-    scanner.keys.push((activation_number, key));
-    #[cfg(not(test))] // TODO: A test violates this. Improve the test with a better flow
-    assert!(scanner.keys.len() <= 2);
-
-    scanner.eventualities.insert(key.to_bytes().as_ref().to_vec(), EventualitiesTracker::new());
-  }
-
   /// Acknowledge having handled a block.
   ///
   /// Creates a lock over the Scanner, preventing its independent scanning operations until
@@ -375,53 +355,6 @@ impl<N: Network, D: Db> Scanner<N, D> {
     mut multisig_completed: mpsc::UnboundedReceiver<bool>,
   ) {
     loop {
-      let (ram_scanned, latest_block_to_scan) = {
-        // Sleep 5 seconds to prevent hammering the node/scanner lock
-        sleep(Duration::from_secs(5)).await;
-
-        let ram_scanned = {
-          let scanner_lock = scanner_hold.read().await;
-          let scanner = scanner_lock.as_ref().unwrap();
-
-          // If we're not scanning for keys yet, wait until we are
-          if scanner.keys.is_empty() {
-            continue;
-          }
-
-          let ram_scanned = scanner.ram_scanned.unwrap();
-          // If a Batch has taken too long to be published, start waiting until it is before
-          // continuing scanning
-          // Solves a race condition around multisig rotation, documented in the relevant doc
-          // and demonstrated with mini
-          if let Some(needing_ack) = scanner.need_ack.front() {
-            let next = ram_scanned + 1;
-            let limit = needing_ack + N::CONFIRMATIONS;
-            assert!(next <= limit);
-            if next == limit {
-              continue;
-            }
-          };
-
-          ram_scanned
-        };
-
-        (
-          ram_scanned,
-          loop {
-            break match network.get_latest_block_number().await {
-              // Only scan confirmed blocks, which we consider effectively finalized
-              // CONFIRMATIONS - 1 as whatever's in the latest block already has 1 confirm
-              Ok(latest) => latest.saturating_sub(N::CONFIRMATIONS.saturating_sub(1)),
-              Err(_) => {
-                warn!("couldn't get latest block number");
-                sleep(Duration::from_secs(60)).await;
-                continue;
-              }
-            };
-          },
-        )
-      };
-
       for block_being_scanned in (ram_scanned + 1) ..= latest_block_to_scan {
         // Redo the checks for if we're too far ahead
         {

processor/scanner/src/lifetime.rs

@@ -0,0 +1,96 @@
+use crate::ScannerFeed;
+
+/// An enum representing the stage of a multisig within its lifetime.
+///
+/// This corresponds to `spec/processor/Multisig Rotation.md`, which details steps 1-8 of the
+/// rotation process. Steps 7-8 regard a multisig which isn't retiring yet retired, and
+/// accordingly, no longer exists, so they are not modelled here (as this only models active
+/// multisigs. Inactive multisigs aren't represented in the first place).
+pub(crate) enum LifetimeStage {
+  /// A new multisig, once active, shouldn't actually start receiving coins until several blocks
+  /// later. If any UI is premature in sending to this multisig, we delay to report the outputs to
+  /// prevent some DoS concerns.
+  ///
+  /// This represents steps 1-3 for a new multisig.
+  ActiveYetNotReporting,
+  /// Active with all outputs being reported on-chain.
+  ///
+  /// This represents step 4 onwards for a new multisig.
+  Active,
+  /// Retiring with all outputs being reported on-chain.
+  ///
+  /// This represents step 4 for a retiring multisig.
+  UsingNewForChange,
+  /// Retiring with outputs being forwarded, reported on-chain once forwarded.
+  ///
+  /// This represents step 5 for a retiring multisig.
+  Forwarding,
+  /// Retiring with only existing obligations being handled.
+  ///
+  /// This represents step 6 for a retiring multisig.
+  ///
+  /// Steps 7 and 8 are represented by the retiring multisig no longer existing, and these states
+  /// are only for multisigs which actively exist.
+  Finishing,
+}
+
+impl LifetimeStage {
+  /// Get the stage of its lifetime this multisig is in based on when the next multisig's key
+  /// activates.
+  ///
+  /// Panics if the multisig being calculated for isn't actually active and a variety of other
+  /// insane cases.
+  pub(crate) fn calculate<S: ScannerFeed>(
+    block_number: u64,
+    activation_block_number: u64,
+    next_keys_activation_block_number: Option<u64>,
+  ) -> Self {
+    assert!(
+      activation_block_number >= block_number,
+      "calculating lifetime stage for an inactive multisig"
+    );
+    // This is exclusive, not inclusive, since we want a CONFIRMATIONS + 10 minutes window and the
+    // activation block itself is the first block within this window
+    let active_yet_not_reporting_end_block =
+      activation_block_number + S::CONFIRMATIONS + S::TEN_MINUTES;
+    if block_number < active_yet_not_reporting_end_block {
+      return LifetimeStage::ActiveYetNotReporting;
+    }
+
+    let Some(next_keys_activation_block_number) = next_keys_activation_block_number else {
+      // If there is no next multisig, this is the active multisig
+      return LifetimeStage::Active;
+    };
+
+    assert!(
+      next_keys_activation_block_number > active_yet_not_reporting_end_block,
+      "next set of keys activated before this multisig activated"
+    );
+
+    // If the new multisig is still having its activation block finalized on-chain, this multisig
+    // is still active (step 3)
+    let new_active_yet_not_reporting_end_block =
+      next_keys_activation_block_number + S::CONFIRMATIONS + S::TEN_MINUTES;
+    if block_number < new_active_yet_not_reporting_end_block {
+      return LifetimeStage::Active;
+    }
+
+    // Step 4 details a further CONFIRMATIONS
+    let new_active_and_used_for_change_end_block =
+      new_active_yet_not_reporting_end_block + S::CONFIRMATIONS;
+    if block_number < new_active_and_used_for_change_end_block {
+      return LifetimeStage::UsingNewForChange;
+    }
+
+    // Step 5 details a further 6 hours
+    // 6 hours = 6 * 60 minutes = 6 * 6 * 10 minutes
+    let new_active_and_forwarded_to_end_block =
+      new_active_and_used_for_change_end_block + (6 * 6 * S::TEN_MINUTES);
+    if block_number < new_active_and_forwarded_to_end_block {
+      return LifetimeStage::Forwarding;
+    }
+
+    // Step 6
+    LifetimeStage::Finishing
+  }
+}

processor/scanner/src/report.rs

@@ -40,6 +40,20 @@ impl<D: Db, S: ScannerFeed> ContinuallyRan for ReportTask<D, S> {
 
     for b in next_to_potentially_report ..= highest_reportable {
       if ScannerDb::<S>::is_block_notable(&self.db, b) {
+        let outputs = todo!("TODO");
+        let in_instructions_to_report = vec![];
+        for output in outputs {
+          match output.kind() {
+            // These do get reported since the scanner eliminates any which shouldn't be reported
+            OutputType::External => todo!("TODO"),
+            // These do not get reported in Batches
+            OutputType::Branch | OutputType::Change => {}
+            // These now get reported if they're legitimately forwarded
+            OutputType::Forwarded => {
+              todo!("TODO")
+            }
+          }
+        }
         todo!("TODO: Make Batches, which requires handling Forwarded within this crate");
       }
 

processor/scanner/src/scan.rs

@@ -5,6 +5,51 @@ use primitives::{Id, ReceivedOutput, Block};
 // TODO: Localize to ScanDb?
 use crate::{db::ScannerDb, ScannerFeed, ContinuallyRan};
 
+// Construct an InInstruction from an external output.
+//
+// Also returns the address to refund the coins to upon error.
+fn in_instruction_from_output<K: GroupEncoding, A>(
+  output: &impl ReceivedOutput<K, A>,
+) -> (Option<ExternalAddress>, Option<InInstruction>) {
+  assert_eq!(output.kind(), OutputType::External);
+
+  let presumed_origin = output.presumed_origin();
+
+  let mut data = output.data();
+  let max_data_len = usize::try_from(MAX_DATA_LEN).unwrap();
+  if data.len() > max_data_len {
+    error!(
+      "data in output {} exceeded MAX_DATA_LEN ({MAX_DATA_LEN}): {}. skipping",
+      hex::encode(output.id()),
+      data.len(),
+    );
+    return (presumed_origin, None);
+  }
+
+  let shorthand = match Shorthand::decode(&mut data) {
+    Ok(shorthand) => shorthand,
+    Err(e) => {
+      info!("data in output {} wasn't valid shorthand: {e:?}", hex::encode(output.id()));
+      return (presumed_origin, None);
+    }
+  };
+  let instruction = match RefundableInInstruction::try_from(shorthand) {
+    Ok(instruction) => instruction,
+    Err(e) => {
+      info!(
+        "shorthand in output {} wasn't convertible to a RefundableInInstruction: {e:?}",
+        hex::encode(output.id())
+      );
+      return (presumed_origin, None);
+    }
+  };
+
+  (
+    instruction.origin.and_then(|addr| A::try_from(addr).ok()).or(presumed_origin),
+    Some(instruction.instruction),
+  )
+}
+
 struct ScanForOutputsTask<D: Db, S: ScannerFeed> {
   db: D,
   feed: S,
@@ -42,29 +87,79 @@ impl<D: Db, S: ScannerFeed> ContinuallyRan for ScanForOutputsTask<D, S> {
 
       log::info!("scanning block: {} ({b})", hex::encode(block.id()));
 
-      let mut keys =
-        ScannerDb::<S>::keys(&self.db).expect("scanning for a blockchain without any keys set");
-      // Remove all the retired keys
-      while let Some(retire_at) = keys[0].retirement_block_number {
-        if retire_at <= b {
-          keys.remove(0);
-        }
-      }
-      assert!(keys.len() <= 2);
+      assert_eq!(ScannerDb::<S>::next_to_scan_for_outputs_block(&self.db).unwrap(), b);
+      let mut keys = ScannerDb::<S>::active_keys_as_of_next_to_scan_for_outputs_block(&self.db)
+        .expect("scanning for a blockchain without any keys set");
 
-      let mut outputs = vec![];
+      let mut in_instructions = vec![];
       // Scan for each key
       for key in keys {
-        // If this key has yet to active, skip it
-        if key.activation_block_number > b {
-          continue;
-        }
+        for output in block.scan_for_outputs(key.key) {
+          assert_eq!(output.key(), key.key);
 
-        for output in block.scan_for_outputs(key.key.0) {
-          assert_eq!(output.key(), key.key.0);
+          /*
+            The scan task runs ahead of time, obtaining ordering on the external network's blocks
+            with relation to events on the Serai network. This is done via publishing a Batch which
+            contains the InInstructions from External outputs. Accordingly, the scan process only
+            has to yield External outputs.
+
+            It'd appear to make sense to scan for all outputs, and after scanning for all outputs,
+            yield all outputs. The issue is we can't identify outputs we created here. We can only
+            identify the outputs we receive and their *declared intention*.
+
+            We only want to handle Change/Branch/Forwarded outputs we made ourselves. For
+            Forwarded, the reasoning is obvious (retiring multisigs should only downsize, yet
+            accepting new outputs solely because they claim to be Forwarded would increase the size
+            of the multisig). For Change/Branch, it's because such outputs which aren't ours are
+            pointless. They wouldn't hurt to accumulate though.
+
+            The issue is they would hurt to accumulate. We want to filter outputs which are less
+            than their cost to aggregate, a variable itself variable to the current blockchain. We
+            can filter such outputs here, yet if we drop a Change output, we create an insolvency.
+            We'd need to track the loss and offset it later. That means we can't filter such
+            outputs, as we expect any Change output we make.
+
+            The issue is the Change outputs we don't make. Someone can create an output declaring
+            to be Change, yet not actually Change. If we don't filter it, it'd be queued for
+            accumulation, yet it may cost more to accumulate than it's worth.
+
+            The solution is to let the Eventuality task, which does know if we made an output or
+            not (or rather, if a transaction is identical to a transaction which should exist
+            regarding effects) decide to keep/yield the outputs which we should only keep if we
+            made them (as Serai itself should not make worthless outputs, so we can assume they're
+            worthwhile, and even if they're not economically, they are technically).
+
+            The alternative, we drop outputs here with a generic filter rule and then report back
+            the insolvency created, still doesn't work as we'd only be creating if an insolvency if
+            the output was actually made by us (and not simply someone else sending in). We can
+            have the Eventuality task report the insolvency, yet that requires the scanner be
+            responsible for such filter logic. It's more flexible, and has a cleaner API,
+            to do so at a higher level.
+          */
+          if output.kind() != OutputType::External {
+            continue;
+          }
+
+          // Drop External outputs if they're to a multisig which won't report them
+          // This means we should report any External output we save to disk here
+          #[allow(clippy::match_same_arms)]
+          match key.stage {
+            // TODO: Delay External outputs
+            LifetimeStage::ActiveYetNotReporting => todo!("TODO"),
+            // We should report External outputs in these cases
+            LifetimeStage::Active | LifetimeStage::UsingNewForChange => {}
+            // We should report External outputs only once forwarded, where they'll appear as
+            // OutputType::Forwarded
+            LifetimeStage::Forwarding => todo!("TODO"),
+            // We should drop these as we should not be handling new External outputs at this
+            // time
+            LifetimeStage::Finishing => {
+              continue;
+            }
+          }
 
           // Check this isn't dust
-          {
+          let balance_to_use = {
             let mut balance = output.balance();
             // First, subtract 2 * the cost to aggregate, as detailed in
             // `spec/processor/UTXO Management.md`
@@ -79,15 +174,26 @@ impl<D: Db, S: ScannerFeed> ContinuallyRan for ScanForOutputsTask<D, S> {
             if balance.amount.0 < self.feed.dust(balance.coin).0 {
               continue;
             }
-          }
+          };
 
-          outputs.push(output);
+          // Decode and save the InInstruction/refund addr for this output
+          match in_instruction_from_output::<S::Key, S::Address>(output) {
+            (refund_addr, Some(instruction)) => {
+              let instruction = InInstructionWithBalance { instruction, balance: balance_to_use };
+              // TODO: Make a proper struct out of this
+              in_instructions.push((output.id(), refund_addr, instruction));
+              todo!("TODO: Save to be reported")
+            }
+            (Some(refund_addr), None) => todo!("TODO: Queue refund"),
+            // Since we didn't receive an instruction nor can we refund this, accumulate it
+            (None, None) => {}
+          }
         }
       }
 
       let mut txn = self.db.txn();
-      // Save the outputs
-      ScannerDb::<S>::set_outputs(&mut txn, b, outputs);
+      // Save the in instructions
+      ScannerDb::<S>::set_in_instructions(&mut txn, b, in_instructions);
       // Update the next to scan block
       ScannerDb::<S>::set_next_to_scan_for_outputs_block(&mut txn, b + 1);
       txn.commit();