From 155ad48f4cbe5dee2de99797d846158870c031a6 Mon Sep 17 00:00:00 2001
From: Luke Parker <lukeparker5132@gmail.com>
Date: Tue, 20 Aug 2024 18:20:28 -0400
Subject: [PATCH] Handle dust

---
 Cargo.lock                      |  1 +
 processor/primitives/src/lib.rs | 36 ++++++++++++++++-----------------
 processor/scanner/Cargo.toml    |  2 ++
 processor/scanner/src/lib.rs    | 15 ++++++++++++++
 processor/scanner/src/scan.rs   | 20 +++++++++++++++++-
 5 files changed, 55 insertions(+), 19 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index 230ed22f..e3e6f378 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -8669,6 +8669,7 @@ dependencies = [
  "log",
  "parity-scale-codec",
  "serai-db",
+ "serai-primitives",
  "serai-processor-messages",
  "serai-processor-primitives",
  "thiserror",
diff --git a/processor/primitives/src/lib.rs b/processor/primitives/src/lib.rs
index 45f02571..744aae47 100644
--- a/processor/primitives/src/lib.rs
+++ b/processor/primitives/src/lib.rs
@@ -34,6 +34,24 @@ pub trait Id:
 }
 impl<const N: usize> Id for [u8; N] where [u8; N]: Default {}
 
+/// A wrapper for a group element which implements the borsh traits.
+#[derive(Clone, Copy, PartialEq, Eq, Debug)]
+pub struct BorshG<G: GroupEncoding>(pub G);
+impl<G: GroupEncoding> BorshSerialize for BorshG<G> {
+  fn serialize<W: borsh::io::Write>(&self, writer: &mut W) -> borsh::io::Result<()> {
+    writer.write_all(self.0.to_bytes().as_ref())
+  }
+}
+impl<G: GroupEncoding> BorshDeserialize for BorshG<G> {
+  fn deserialize_reader<R: borsh::io::Read>(reader: &mut R) -> borsh::io::Result<Self> {
+    let mut repr = G::Repr::default();
+    reader.read_exact(repr.as_mut())?;
+    Ok(Self(
+      Option::<G>::from(G::from_bytes(&repr)).ok_or(borsh::io::Error::other("invalid point"))?,
+    ))
+  }
+}
+
 /// The type of the output.
 #[derive(Clone, Copy, PartialEq, Eq, Hash, Debug)]
 pub enum OutputType {
@@ -171,21 +189,3 @@ pub trait Block: Send + Sync + Sized + Clone + Debug {
   /// Scan all outputs within this block to find the outputs spendable by this key.
   fn scan_for_outputs(&self, key: Self::Key) -> Vec<Self::Output>;
 }
-
-/// A wrapper for a group element which implements the borsh traits.
-#[derive(Clone, Copy, PartialEq, Eq, Debug)]
-pub struct BorshG<G: GroupEncoding>(pub G);
-impl<G: GroupEncoding> BorshSerialize for BorshG<G> {
-  fn serialize<W: borsh::io::Write>(&self, writer: &mut W) -> borsh::io::Result<()> {
-    writer.write_all(self.0.to_bytes().as_ref())
-  }
-}
-impl<G: GroupEncoding> BorshDeserialize for BorshG<G> {
-  fn deserialize_reader<R: borsh::io::Read>(reader: &mut R) -> borsh::io::Result<Self> {
-    let mut repr = G::Repr::default();
-    reader.read_exact(repr.as_mut())?;
-    Ok(Self(
-      Option::<G>::from(G::from_bytes(&repr)).ok_or(borsh::io::Error::other("invalid point"))?,
-    ))
-  }
-}
diff --git a/processor/scanner/Cargo.toml b/processor/scanner/Cargo.toml
index 670581d9..82de4de1 100644
--- a/processor/scanner/Cargo.toml
+++ b/processor/scanner/Cargo.toml
@@ -35,5 +35,7 @@ tokio = { version = "1", default-features = false, features = ["rt-multi-thread"
 
 serai-db = { path = "../../common/db" }
 
+serai-primitives = { path = "../../substrate/primitives", default-features = false, features = ["std"] }
+
 messages = { package = "serai-processor-messages", path = "../messages" }
 primitives = { package = "serai-processor-primitives", path = "../primitives" }
diff --git a/processor/scanner/src/lib.rs b/processor/scanner/src/lib.rs
index addebb60..02c88599 100644
--- a/processor/scanner/src/lib.rs
+++ b/processor/scanner/src/lib.rs
@@ -2,6 +2,7 @@ use core::{fmt::Debug, time::Duration};
 
 use tokio::sync::mpsc;
 
+use serai_primitives::{Coin, Amount};
 use primitives::{ReceivedOutput, BlockHeader, Block};
 
 mod db;
@@ -57,6 +58,20 @@ pub trait ScannerFeed: Send + Sync {
 
   /// Fetch a block by its number.
   async fn block_by_number(&self, number: u64) -> Result<Self::Block, Self::EphemeralError>;
+
+  /// The cost to aggregate an input as of the specified block.
+  ///
+  /// This is defined as the transaction fee for a 2-input, 1-output transaction.
+  async fn cost_to_aggregate(
+    &self,
+    coin: Coin,
+    block_number: u64,
+  ) -> Result<Amount, Self::EphemeralError>;
+
+  /// The dust threshold for the specified coin.
+  ///
+  /// This should be a value worth handling at a human level.
+  fn dust(&self, coin: Coin) -> Amount;
 }
 
 type BlockIdFor<S> = <<<S as ScannerFeed>::Block as Block>::Header as BlockHeader>::Id;
diff --git a/processor/scanner/src/scan.rs b/processor/scanner/src/scan.rs
index 6743d950..6058c7da 100644
--- a/processor/scanner/src/scan.rs
+++ b/processor/scanner/src/scan.rs
@@ -62,7 +62,25 @@ impl<D: Db, S: ScannerFeed> ContinuallyRan for ScanForOutputsTask<D, S> {
 
         for output in block.scan_for_outputs(key.key.0) {
           assert_eq!(output.key(), key.key.0);
-          // TODO: Check for dust
+
+          // Check this isn't dust
+          {
+            let mut balance = output.balance();
+            // First, subtract 2 * the cost to aggregate, as detailed in
+            // `spec/processor/UTXO Management.md`
+            // TODO: Cache this
+            let cost_to_aggregate =
+              self.feed.cost_to_aggregate(balance.coin, b).await.map_err(|e| {
+                format!("couldn't fetch cost to aggregate {:?} at {b}: {e:?}", balance.coin)
+              })?;
+            balance.amount.0 -= 2 * cost_to_aggregate.0;
+
+            // Now, check it's still past the dust threshold
+            if balance.amount.0 < self.feed.dust(balance.coin).0 {
+              continue;
+            }
+          }
+
           outputs.push(output);
         }
       }