Report a Change Output with every Eventuality to ensure we don't fall out of synchrony

processor/ethereum/src/primitives/block.rs

@@ -61,7 +61,29 @@ impl primitives::Block for FullEpoch {
     // Associate all outputs with the latest active key
     // We don't associate these with the current key within the SC as that'll cause outputs to be
     // marked for forwarding if the SC is delayed to actually rotate
-    self.instructions.iter().cloned().map(|instruction| Output { key, instruction }).collect()
+    let mut outputs: Vec<_> = self
+      .instructions
+      .iter()
+      .cloned()
+      .map(|instruction| Output::Output { key, instruction })
+      .collect();
+
+    /*
+      The scanner requires a change output be associated with every Eventuality that came from
+      fulfilling payments, unless said Eventuality descends from an Eventuality meeting that
+      requirement from the same fulfillment. This ensures we have a fully populated Eventualities
+      set by the time we process the block which has an Eventuality.
+
+      Accordingly, for any block with an Eventuality completion, we claim there's a Change output
+      so that the block is flagged. Ethereum doesn't actually have Change outputs, yet the scanner
+      won't report them to Substrate, and the Smart Contract scheduler will drop any/all outputs
+      passed to it (handwaving their balances as present within the Smart Contract).
+    */
+    if !self.executed.is_empty() {
+      outputs.push(Output::Eventuality { key, nonce: self.executed.first().unwrap().nonce() });
+    }
+
+    outputs
   }
 
   #[allow(clippy::type_complexity)]
@@ -85,15 +107,17 @@ impl primitives::Block for FullEpoch {
         "Router emitted distinct event for nonce {}",
         executed.nonce()
       );
+
       /*
         The transaction ID is used to determine how internal outputs from this transaction should
         be handled (if they were actually internal or if they were just to an internal address).
-        The Ethereum integration doesn't have internal addresses, and this transaction wasn't made
-        by Serai. It was simply authorized by Serai yet may or may not be associated with other
-        actions we don't want to flag as our own.
+        The Ethereum integration doesn't use internal addresses, and only uses internal outputs to
+        flag a block as having an Eventuality. Those internal outputs will always be scanned, and
+        while they may be dropped/kept by this ID, the scheduler will then always drop them.
+        Accordingly, we have free reign as to what to set the transaction ID to.
 
-        Accordingly, we set the transaction ID to the nonce. This is unique barring someone finding
-        the preimage which hashes to this nonce, and won't cause any other data to be associated.
+        We set the ID to the nonce as it's the most helpful value and unique barring someone
+        finding the premise for this as a hash.
       */
       let mut tx_id = [0; 32];
       tx_id[.. 8].copy_from_slice(executed.nonce().to_le_bytes().as_slice());

processor/ethereum/src/primitives/mod.rs

@@ -1,3 +1,5 @@
+use serai_client::primitives::Amount;
+
 pub(crate) mod output;
 pub(crate) mod transaction;
 pub(crate) mod machine;
@@ -10,3 +12,10 @@ pub(crate) const DAI: [u8; 20] =
   };
 
 pub(crate) const TOKENS: [[u8; 20]; 1] = [DAI];
+
+// 8 decimals, so 1_000_000_00 would be 1 ETH. This is 0.0015 ETH (5 USD if Ether is ~3300 USD).
+#[allow(clippy::inconsistent_digit_grouping)]
+pub(crate) const ETHER_DUST: Amount = Amount(1_500_00);
+// 5 DAI
+#[allow(clippy::inconsistent_digit_grouping)]
+pub(crate) const DAI_DUST: Amount = Amount(5_000_000_00);

processor/ethereum/src/primitives/output.rs

@@ -15,7 +15,7 @@ use serai_client::{
 use primitives::{OutputType, ReceivedOutput};
 use ethereum_router::{Coin as EthereumCoin, InInstruction as EthereumInInstruction};
 
-use crate::DAI;
+use crate::{DAI, ETHER_DUST};
 
 fn coin_to_serai_coin(coin: &EthereumCoin) -> Option<Coin> {
   match coin {
@@ -59,58 +59,122 @@ impl AsMut<[u8]> for OutputId {
 }
 
 #[derive(Clone, PartialEq, Eq, Debug)]
-pub(crate) struct Output {
-  pub(crate) key: <Secp256k1 as Ciphersuite>::G,
-  pub(crate) instruction: EthereumInInstruction,
+pub(crate) enum Output {
+  Output { key: <Secp256k1 as Ciphersuite>::G, instruction: EthereumInInstruction },
+  Eventuality { key: <Secp256k1 as Ciphersuite>::G, nonce: u64 },
 }
 impl ReceivedOutput<<Secp256k1 as Ciphersuite>::G, Address> for Output {
   type Id = OutputId;
   type TransactionId = [u8; 32];
 
-  // We only scan external outputs as we don't have branch/change/forwards
   fn kind(&self) -> OutputType {
-    OutputType::External
+    match self {
+      // All outputs received are External
+      Output::Output { .. } => OutputType::External,
+      // Yet upon Eventuality completions, we report a Change output to ensure synchrony per the
+      // scanner's documented bounds
+      Output::Eventuality { .. } => OutputType::Change,
+    }
   }
 
   fn id(&self) -> Self::Id {
-    let mut id = [0; 40];
-    id[.. 32].copy_from_slice(&self.instruction.id.0);
-    id[32 ..].copy_from_slice(&self.instruction.id.1.to_le_bytes());
-    OutputId(id)
+    match self {
+      Output::Output { key: _, instruction } => {
+        let mut id = [0; 40];
+        id[.. 32].copy_from_slice(&instruction.id.0);
+        id[32 ..].copy_from_slice(&instruction.id.1.to_le_bytes());
+        OutputId(id)
+      }
+      // Yet upon Eventuality completions, we report a Change output to ensure synchrony per the
+      // scanner's documented bounds
+      Output::Eventuality { key: _, nonce } => {
+        let mut id = [0; 40];
+        id[.. 8].copy_from_slice(&nonce.to_le_bytes());
+        OutputId(id)
+      }
+    }
   }
 
   fn transaction_id(&self) -> Self::TransactionId {
-    self.instruction.id.0
+    match self {
+      Output::Output { key: _, instruction } => instruction.id.0,
+      Output::Eventuality { key: _, nonce } => {
+        let mut id = [0; 32];
+        id[.. 8].copy_from_slice(&nonce.to_le_bytes());
+        id
+      }
+    }
   }
 
   fn key(&self) -> <Secp256k1 as Ciphersuite>::G {
-    self.key
+    match self {
+      Output::Output { key, .. } | Output::Eventuality { key, .. } => *key,
+    }
   }
 
   fn presumed_origin(&self) -> Option<Address> {
-    Some(Address::from(self.instruction.from))
+    match self {
+      Output::Output { key: _, instruction } => Some(Address::from(instruction.from)),
+      Output::Eventuality { .. } => None,
+    }
   }
 
   fn balance(&self) -> Balance {
-    let coin = coin_to_serai_coin(&self.instruction.coin).unwrap_or_else(|| {
-      panic!(
-        "mapping coin from an EthereumInInstruction with coin {}, which we don't handle.",
-        "this never should have been yielded"
-      )
-    });
-    Balance { coin, amount: amount_to_serai_amount(coin, self.instruction.amount) }
+    match self {
+      Output::Output { key: _, instruction } => {
+        let coin = coin_to_serai_coin(&instruction.coin).unwrap_or_else(|| {
+          panic!(
+            "mapping coin from an EthereumInInstruction with coin {}, which we don't handle.",
+            "this never should have been yielded"
+          )
+        });
+        Balance { coin, amount: amount_to_serai_amount(coin, instruction.amount) }
+      }
+      Output::Eventuality { .. } => Balance { coin: Coin::Ether, amount: ETHER_DUST },
+    }
   }
   fn data(&self) -> &[u8] {
-    &self.instruction.data
+    match self {
+      Output::Output { key: _, instruction } => &instruction.data,
+      Output::Eventuality { .. } => &[],
+    }
   }
 
   fn write<W: io::Write>(&self, writer: &mut W) -> io::Result<()> {
-    writer.write_all(self.key.to_bytes().as_ref())?;
-    self.instruction.write(writer)
+    match self {
+      Output::Output { key, instruction } => {
+        writer.write_all(&[0])?;
+        writer.write_all(key.to_bytes().as_ref())?;
+        instruction.write(writer)
+      }
+      Output::Eventuality { key, nonce } => {
+        writer.write_all(&[1])?;
+        writer.write_all(key.to_bytes().as_ref())?;
+        writer.write_all(&nonce.to_le_bytes())
+      }
+    }
   }
   fn read<R: io::Read>(reader: &mut R) -> io::Result<Self> {
-    let key = Secp256k1::read_G(reader)?;
-    let instruction = EthereumInInstruction::read(reader)?;
-    Ok(Self { key, instruction })
+    let mut kind = [0xff];
+    reader.read_exact(&mut kind)?;
+    if kind[0] >= 2 {
+      Err(io::Error::other("unknown Output type"))?;
+    }
+
+    Ok(match kind[0] {
+      0 => {
+        let key = Secp256k1::read_G(reader)?;
+        let instruction = EthereumInInstruction::read(reader)?;
+        Self::Output { key, instruction }
+      }
+      1 => {
+        let key = Secp256k1::read_G(reader)?;
+        let mut nonce = [0; 8];
+        reader.read_exact(&mut nonce)?;
+        let nonce = u64::from_le_bytes(nonce);
+        Self::Eventuality { key, nonce }
+      }
+      _ => unreachable!(),
+    })
   }
 }

processor/scanner/src/lib.rs

@@ -321,7 +321,9 @@ pub trait Scheduler<S: ScannerFeed>: 'static + Send {
   ///
   /// Any Eventualities returned by this function must include an output-to-Serai (such as a Branch
   /// or Change), unless they descend from a transaction returned by this function which satisfies
-  /// that requirement.
+  /// that requirement. This ensures when we scan outputs from transactions we made, we report the
+  /// block up to Substrate, and obtain synchrony on all prior blocks (allowing us to identify our
+  /// own transactions, which we may be prior unaware of due to a lagging view of Substrate).
   ///
   /// `active_keys` is the list of active keys, potentially including a key for which we've already
   /// called `retire_key` on. If so, its stage will be `Finishing` and no further operations will