Outline of the transaction-chaining scheduler

Cargo.lock

@@ -8656,6 +8656,7 @@ dependencies = [
  "group",
  "log",
  "parity-scale-codec",
+ "serai-coins-primitives",
  "serai-primitives",
  "tokio",
 ]
@@ -8674,9 +8675,7 @@ dependencies = [
  "serai-db",
  "serai-in-instructions-primitives",
  "serai-primitives",
- "serai-processor-messages",
  "serai-processor-primitives",
- "thiserror",
  "tokio",
 ]
 
@@ -8712,6 +8711,17 @@ dependencies = [
 [[package]]
 name = "serai-processor-transaction-chaining-scheduler"
 version = "0.1.0"
+dependencies = [
+ "borsh",
+ "group",
+ "parity-scale-codec",
+ "serai-coins-primitives",
+ "serai-db",
+ "serai-primitives",
+ "serai-processor-primitives",
+ "serai-processor-scanner",
+ "serai-processor-utxo-scheduler-primitives",
+]
 
 [[package]]
 name = "serai-processor-utxo-scheduler-primitives"

processor/primitives/Cargo.toml

@@ -22,6 +22,7 @@ async-trait = { version = "0.1", default-features = false }
 group = { version = "0.13", default-features = false }
 
 serai-primitives = { path = "../../substrate/primitives", default-features = false, features = ["std"] }
+serai-coins-primitives = { path = "../../substrate/coins/primitives", default-features = false, features = ["std"] }
 
 scale = { package = "parity-scale-codec", version = "3", default-features = false, features = ["std"] }
 borsh = { version = "1", default-features = false, features = ["std", "derive", "de_strict_order"] }

processor/primitives/src/lib.rs

@@ -21,6 +21,9 @@ pub use eventuality::*;
 mod block;
 pub use block::*;
 
+mod payment;
+pub use payment::*;
+
 /// An ID for an output/transaction/block/etc.
 ///
 /// IDs don't need to implement `Copy`, enabling `[u8; 33]`, `[u8; 64]` to be used. IDs are still

processor/primitives/src/payment.rs

@@ -0,0 +1,42 @@
+use serai_primitives::{Balance, Data};
+use serai_coins_primitives::OutInstructionWithBalance;
+
+use crate::Address;
+
+/// A payment to fulfill.
+#[derive(Clone)]
+pub struct Payment<A: Address> {
+  address: A,
+  balance: Balance,
+  data: Option<Vec<u8>>,
+}
+
+impl<A: Address> TryFrom<OutInstructionWithBalance> for Payment<A> {
+  type Error = ();
+  fn try_from(out_instruction_with_balance: OutInstructionWithBalance) -> Result<Self, ()> {
+    Ok(Payment {
+      address: out_instruction_with_balance.instruction.address.try_into().map_err(|_| ())?,
+      balance: out_instruction_with_balance.balance,
+      data: out_instruction_with_balance.instruction.data.map(Data::consume),
+    })
+  }
+}
+
+impl<A: Address> Payment<A> {
+  /// Create a new Payment.
+  pub fn new(address: A, balance: Balance, data: Option<Vec<u8>>) -> Self {
+    Payment { address, balance, data }
+  }
+  /// The address to pay.
+  pub fn address(&self) -> &A {
+    &self.address
+  }
+  /// The balance to transfer.
+  pub fn balance(&self) -> Balance {
+    self.balance
+  }
+  /// The data to associate with this payment.
+  pub fn data(&self) -> &Option<Vec<u8>> {
+    &self.data
+  }
+}

processor/scanner/Cargo.toml

@@ -19,7 +19,6 @@ workspace = true
 [dependencies]
 # Macros
 async-trait = { version = "0.1", default-features = false }
-thiserror = { version = "1", default-features = false }
 
 # Encoders
 hex = { version = "0.4", default-features = false, features = ["std"] }
@@ -37,7 +36,6 @@ serai-db = { path = "../../common/db" }
 
 serai-primitives = { path = "../../substrate/primitives", default-features = false, features = ["std"] }
 serai-in-instructions-primitives = { path = "../../substrate/in-instructions/primitives", default-features = false, features = ["std"] }
-serai-coins-primitives = { path = "../../substrate/coins/primitives", default-features = false, features = ["std"] }
+serai-coins-primitives = { path = "../../substrate/coins/primitives", default-features = false, features = ["std", "borsh"] }
 
-messages = { package = "serai-processor-messages", path = "../messages" }
 primitives = { package = "serai-processor-primitives", path = "../primitives" }

processor/scanner/src/eventuality/mod.rs

@@ -4,7 +4,7 @@ use group::GroupEncoding;
 
 use serai_db::{Get, DbTxn, Db};
 
-use primitives::{task::ContinuallyRan, OutputType, ReceivedOutput, Eventuality, Block};
+use primitives::{task::ContinuallyRan, OutputType, ReceivedOutput, Eventuality, Block, Payment};
 
 use crate::{
   lifetime::LifetimeStage,
@@ -12,7 +12,7 @@ use crate::{
     SeraiKey, OutputWithInInstruction, ReceiverScanData, ScannerGlobalDb, SubstrateToEventualityDb,
     ScanToEventualityDb,
   },
-  BlockExt, ScannerFeed, KeyFor, OutputFor, EventualityFor, Payment, SchedulerUpdate, Scheduler,
+  BlockExt, ScannerFeed, KeyFor, AddressFor, OutputFor, EventualityFor, SchedulerUpdate, Scheduler,
   sort_outputs,
   scan::{next_to_scan_for_outputs_block, queue_output_until_block},
 };
@@ -168,7 +168,10 @@ impl<D: Db, S: ScannerFeed, Sch: Scheduler<S>> EventualityTask<D, S, Sch> {
         let new_eventualities = self.scheduler.fulfill(
           &mut txn,
           &keys_with_stages,
-          burns.into_iter().filter_map(|burn| Payment::try_from(burn).ok()).collect(),
+          burns
+            .into_iter()
+            .filter_map(|burn| Payment::<AddressFor<S>>::try_from(burn).ok())
+            .collect(),
         );
         intake_eventualities::<S>(&mut txn, new_eventualities);
       }

processor/scanner/src/lib.rs

@@ -5,11 +5,11 @@ use group::GroupEncoding;
 
 use serai_db::{Get, DbTxn, Db};
 
-use serai_primitives::{NetworkId, Coin, Amount, Balance, Data};
+use serai_primitives::{NetworkId, Coin, Amount};
 use serai_in_instructions_primitives::Batch;
 use serai_coins_primitives::OutInstructionWithBalance;
 
-use primitives::{task::*, Address, ReceivedOutput, Block};
+use primitives::{task::*, Address, ReceivedOutput, Block, Payment};
 
 // Logic for deciding where in its lifetime a multisig is.
 mod lifetime;
@@ -195,6 +195,16 @@ impl<S: ScannerFeed> Return<S> {
     let output = OutputFor::<S>::read(reader)?;
     Ok(Return { address, output })
   }
+
+  /// The address to return the output to.
+  pub fn address(&self) -> &AddressFor<S> {
+    &self.address
+  }
+
+  /// The output to return.
+  pub fn output(&self) -> &OutputFor<S> {
+    &self.output
+  }
 }
 
 /// An update for the scheduler.
@@ -219,40 +229,6 @@ impl<S: ScannerFeed> SchedulerUpdate<S> {
   }
 }
 
-/// A payment to fulfill.
-#[derive(Clone)]
-pub struct Payment<S: ScannerFeed> {
-  address: AddressFor<S>,
-  balance: Balance,
-  data: Option<Vec<u8>>,
-}
-
-impl<S: ScannerFeed> TryFrom<OutInstructionWithBalance> for Payment<S> {
-  type Error = ();
-  fn try_from(out_instruction_with_balance: OutInstructionWithBalance) -> Result<Self, ()> {
-    Ok(Payment {
-      address: out_instruction_with_balance.instruction.address.try_into().map_err(|_| ())?,
-      balance: out_instruction_with_balance.balance,
-      data: out_instruction_with_balance.instruction.data.map(Data::consume),
-    })
-  }
-}
-
-impl<S: ScannerFeed> Payment<S> {
-  /// The address to pay.
-  pub fn address(&self) -> &AddressFor<S> {
-    &self.address
-  }
-  /// The balance to transfer.
-  pub fn balance(&self) -> Balance {
-    self.balance
-  }
-  /// The data to associate with this payment.
-  pub fn data(&self) -> &Option<Vec<u8>> {
-    &self.data
-  }
-}
-
 /// The object responsible for accumulating outputs and planning new transactions.
 pub trait Scheduler<S: ScannerFeed>: 'static + Send {
   /// Activate a key.
@@ -327,7 +303,7 @@ pub trait Scheduler<S: ScannerFeed>: 'static + Send {
     &mut self,
     txn: &mut impl DbTxn,
     active_keys: &[(KeyFor<S>, LifetimeStage)],
-    payments: Vec<Payment<S>>,
+    payments: Vec<Payment<AddressFor<S>>>,
   ) -> HashMap<Vec<u8>, Vec<EventualityFor<S>>>;
 }
 

processor/scanner/src/lifetime.rs

@@ -6,7 +6,7 @@ use crate::ScannerFeed;
 /// 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).
-#[derive(Clone, Copy, PartialEq)]
+#[derive(Clone, Copy, PartialEq, Debug)]
 pub 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

processor/scheduler/utxo/primitives/src/lib.rs

@@ -6,12 +6,12 @@ use core::fmt::Debug;
 
 use serai_primitives::{Coin, Amount};
 
-use primitives::ReceivedOutput;
-use scanner::{Payment, ScannerFeed, AddressFor, OutputFor};
+use primitives::{ReceivedOutput, Payment};
+use scanner::{ScannerFeed, KeyFor, AddressFor, OutputFor};
 
 /// An object able to plan a transaction.
 #[async_trait::async_trait]
-pub trait TransactionPlanner<S: ScannerFeed> {
+pub trait TransactionPlanner<S: ScannerFeed>: 'static + Send + Sync {
   /// An error encountered when determining the fee rate.
   ///
   /// This MUST be an ephemeral error. Retrying fetching data from the blockchain MUST eventually
@@ -33,17 +33,22 @@ pub trait TransactionPlanner<S: ScannerFeed> {
     coin: Coin,
   ) -> Result<Self::FeeRate, Self::EphemeralError>;
 
+  /// The branch address for this key of Serai's.
+  fn branch_address(key: KeyFor<S>) -> AddressFor<S>;
+  /// The change address for this key of Serai's.
+  fn change_address(key: KeyFor<S>) -> AddressFor<S>;
+  /// The forwarding address for this key of Serai's.
+  fn forwarding_address(key: KeyFor<S>) -> AddressFor<S>;
+
   /// Calculate the for a tansaction with this structure.
   ///
   /// The fee rate, inputs, and payments, will all be for the same coin. The returned fee is
   /// denominated in this coin.
   fn calculate_fee(
-    &self,
-    block_number: u64,
     fee_rate: Self::FeeRate,
     inputs: Vec<OutputFor<S>>,
-    payments: Vec<Payment<S>>,
-    change: Option<AddressFor<S>>,
+    payments: Vec<Payment<AddressFor<S>>>,
+    change: Option<KeyFor<S>>,
   ) -> Amount;
 
   /// Plan a transaction.
@@ -53,12 +58,10 @@ pub trait TransactionPlanner<S: ScannerFeed> {
   ///
   /// `change` will always be an address belonging to the Serai network.
   fn plan(
-    &self,
-    block_number: u64,
     fee_rate: Self::FeeRate,
     inputs: Vec<OutputFor<S>>,
-    payments: Vec<Payment<S>>,
-    change: Option<AddressFor<S>>,
+    payments: Vec<Payment<AddressFor<S>>>,
+    change: Option<KeyFor<S>>,
   ) -> Self::PlannedTransaction;
 
   /// Obtain a PlannedTransaction via amortizing the fee over the payments.
@@ -69,13 +72,11 @@ pub trait TransactionPlanner<S: ScannerFeed> {
   ///
   /// Returns `None` if the fee exceeded the inputs, or `Some` otherwise.
   fn plan_transaction_with_fee_amortization(
-    &self,
     operating_costs: &mut u64,
-    block_number: u64,
     fee_rate: Self::FeeRate,
     inputs: Vec<OutputFor<S>>,
-    mut payments: Vec<Payment<S>>,
-    change: Option<AddressFor<S>>,
+    mut payments: Vec<Payment<AddressFor<S>>>,
+    mut change: Option<KeyFor<S>>,
   ) -> Option<Self::PlannedTransaction> {
     // Sanity checks
     {
@@ -102,9 +103,7 @@ pub trait TransactionPlanner<S: ScannerFeed> {
       // Sort payments from high amount to low amount
       payments.sort_by(|a, b| a.balance().amount.0.cmp(&b.balance().amount.0).reverse());
 
-      let mut fee = self
-        .calculate_fee(block_number, fee_rate, inputs.clone(), payments.clone(), change.clone())
-        .0;
+      let mut fee = Self::calculate_fee(fee_rate, inputs.clone(), payments.clone(), change).0;
       let mut amortized = 0;
       while !payments.is_empty() {
         // We need to pay the fee, and any accrued operating costs, minus what we've already
@@ -124,9 +123,7 @@ pub trait TransactionPlanner<S: ScannerFeed> {
         if payments.last().unwrap().balance().amount.0 <= (per_payment_fee + S::dust(coin).0) {
           amortized += payments.pop().unwrap().balance().amount.0;
           // Recalculate the fee and try again
-          fee = self
-            .calculate_fee(block_number, fee_rate, inputs.clone(), payments.clone(), change.clone())
-            .0;
+          fee = Self::calculate_fee(fee_rate, inputs.clone(), payments.clone(), change).0;
           continue;
         }
         // Break since all of these payments shouldn't be dropped
@@ -167,6 +164,15 @@ pub trait TransactionPlanner<S: ScannerFeed> {
           amortized += per_payment_fee;
         }
         assert!(amortized >= (*operating_costs + fee));
+
+        // If the change is less than the dust, drop it
+        let would_be_change = inputs.iter().map(|input| input.balance().amount.0).sum::<u64>() -
+          payments.iter().map(|payment| payment.balance().amount.0).sum::<u64>() -
+          fee;
+        if would_be_change < S::dust(coin).0 {
+          change = None;
+          *operating_costs += would_be_change;
+        }
       }
 
       // Update the amount of operating costs
@@ -174,6 +180,6 @@ pub trait TransactionPlanner<S: ScannerFeed> {
     }
 
     // Because we amortized, or accrued as operating costs, the fee, make the transaction
-    Some(self.plan(block_number, fee_rate, inputs, payments, change))
+    Some(Self::plan(fee_rate, inputs, payments, change))
   }
 }

processor/scheduler/utxo/transaction-chaining/Cargo.toml

@@ -1,9 +1,9 @@
 [package]
 name = "serai-processor-transaction-chaining-scheduler"
 version = "0.1.0"
-description = "Scheduler for networks with transaction chaining for the Serai processor"
+description = "Scheduler for UTXO networks with transaction chaining for the Serai processor"
 license = "AGPL-3.0-only"
-repository = "https://github.com/serai-dex/serai/tree/develop/processor/scheduler/transaction-chaining"
+repository = "https://github.com/serai-dex/serai/tree/develop/processor/scheduler/utxo/transaction-chaining"
 authors = ["Luke Parker <lukeparker5132@gmail.com>"]
 keywords = []
 edition = "2021"
@@ -14,9 +14,22 @@ all-features = true
 rustdoc-args = ["--cfg", "docsrs"]
 
 [package.metadata.cargo-machete]
-ignored = ["scale"]
+ignored = ["scale", "borsh"]
 
 [lints]
 workspace = true
 
 [dependencies]
+group = { version = "0.13", default-features = false }
+
+scale = { package = "parity-scale-codec", version = "3", default-features = false, features = ["std"] }
+borsh = { version = "1", default-features = false, features = ["std", "derive", "de_strict_order"] }
+
+serai-primitives = { path = "../../../../substrate/primitives", default-features = false, features = ["std"] }
+serai-coins-primitives = { path = "../../../../substrate/coins/primitives", default-features = false, features = ["std"] }
+
+serai-db = { path = "../../../../common/db" }
+
+primitives = { package = "serai-processor-primitives", path = "../../../primitives" }
+scheduler-primitives = { package = "serai-processor-utxo-scheduler-primitives", path = "../primitives" }
+scanner = { package = "serai-processor-scanner", path = "../../../scanner" }

processor/scheduler/utxo/transaction-chaining/LICENSE

@@ -1,6 +1,6 @@
 AGPL-3.0-only license
 
-Copyright (c) 2022-2024 Luke Parker
+Copyright (c) 2024 Luke Parker
 
 This program is free software: you can redistribute it and/or modify
 it under the terms of the GNU Affero General Public License Version 3 as

processor/scheduler/utxo/transaction-chaining/src/db.rs

@@ -0,0 +1,49 @@
+use core::marker::PhantomData;
+
+use group::GroupEncoding;
+
+use serai_primitives::Coin;
+
+use serai_db::{Get, DbTxn, create_db};
+
+use primitives::ReceivedOutput;
+use scanner::{ScannerFeed, KeyFor, OutputFor};
+
+create_db! {
+  TransactionChainingScheduler {
+    SerializedOutputs: (key: &[u8], coin: Coin) -> Vec<u8>,
+  }
+}
+
+pub(crate) struct Db<S: ScannerFeed>(PhantomData<S>);
+impl<S: ScannerFeed> Db<S> {
+  pub(crate) fn outputs(
+    getter: &impl Get,
+    key: KeyFor<S>,
+    coin: Coin,
+  ) -> Option<Vec<OutputFor<S>>> {
+    let buf = SerializedOutputs::get(getter, key.to_bytes().as_ref(), coin)?;
+    let mut buf = buf.as_slice();
+
+    let mut res = Vec::with_capacity(buf.len() / 128);
+    while !buf.is_empty() {
+      res.push(OutputFor::<S>::read(&mut buf).unwrap());
+    }
+    Some(res)
+  }
+  pub(crate) fn set_outputs(
+    txn: &mut impl DbTxn,
+    key: KeyFor<S>,
+    coin: Coin,
+    outputs: &[OutputFor<S>],
+  ) {
+    let mut buf = Vec::with_capacity(outputs.len() * 128);
+    for output in outputs {
+      output.write(&mut buf).unwrap();
+    }
+    SerializedOutputs::set(txn, key.to_bytes().as_ref(), coin, &buf);
+  }
+  pub(crate) fn del_outputs(txn: &mut impl DbTxn, key: KeyFor<S>, coin: Coin) {
+    SerializedOutputs::del(txn, key.to_bytes().as_ref(), coin);
+  }
+}

processor/scheduler/utxo/transaction-chaining/src/lib.rs

@@ -1,3 +1,151 @@
 #![cfg_attr(docsrs, feature(doc_auto_cfg))]
 #![doc = include_str!("../README.md")]
 #![deny(missing_docs)]
+
+use core::marker::PhantomData;
+use std::collections::HashMap;
+
+use serai_primitives::Coin;
+
+use serai_db::DbTxn;
+
+use primitives::{ReceivedOutput, Payment};
+use scanner::{
+  LifetimeStage, ScannerFeed, KeyFor, AddressFor, OutputFor, EventualityFor, SchedulerUpdate,
+  Scheduler as SchedulerTrait,
+};
+use scheduler_primitives::*;
+
+mod db;
+use db::Db;
+
+/// A planned transaction.
+pub struct PlannedTransaction<S: ScannerFeed, T> {
+  /// The signable transaction.
+  signable: T,
+  /// The outputs we'll receive from this.
+  effected_received_outputs: OutputFor<S>,
+  /// The Evtnuality to watch for.
+  eventuality: EventualityFor<S>,
+}
+
+/// A scheduler of transactions for networks premised on the UTXO model which support
+/// transaction chaining.
+pub struct Scheduler<
+  S: ScannerFeed,
+  T,
+  P: TransactionPlanner<S, PlannedTransaction = PlannedTransaction<S, T>>,
+>(PhantomData<S>, PhantomData<T>, PhantomData<P>);
+
+impl<S: ScannerFeed, T, P: TransactionPlanner<S, PlannedTransaction = PlannedTransaction<S, T>>>
+  Scheduler<S, T, P>
+{
+  fn accumulate_outputs(txn: &mut impl DbTxn, key: KeyFor<S>, outputs: &[OutputFor<S>]) {
+    // Accumulate them in memory
+    let mut outputs_by_coin = HashMap::with_capacity(1);
+    for output in outputs.iter().filter(|output| output.key() == key) {
+      let coin = output.balance().coin;
+      if let std::collections::hash_map::Entry::Vacant(e) = outputs_by_coin.entry(coin) {
+        e.insert(Db::<S>::outputs(txn, key, coin).unwrap());
+      }
+      outputs_by_coin.get_mut(&coin).unwrap().push(output.clone());
+    }
+
+    // Flush them to the database
+    for (coin, outputs) in outputs_by_coin {
+      Db::<S>::set_outputs(txn, key, coin, &outputs);
+    }
+  }
+}
+
+impl<
+    S: ScannerFeed,
+    T: 'static + Send + Sync,
+    P: TransactionPlanner<S, PlannedTransaction = PlannedTransaction<S, T>>,
+  > SchedulerTrait<S> for Scheduler<S, T, P>
+{
+  fn activate_key(&mut self, txn: &mut impl DbTxn, key: KeyFor<S>) {
+    for coin in S::NETWORK.coins() {
+      Db::<S>::set_outputs(txn, key, *coin, &vec![]);
+    }
+  }
+
+  fn flush_key(&mut self, txn: &mut impl DbTxn, retiring_key: KeyFor<S>, new_key: KeyFor<S>) {
+    todo!("TODO")
+  }
+
+  fn retire_key(&mut self, txn: &mut impl DbTxn, key: KeyFor<S>) {
+    for coin in S::NETWORK.coins() {
+      assert!(Db::<S>::outputs(txn, key, *coin).is_none());
+      Db::<S>::del_outputs(txn, key, *coin);
+    }
+  }
+
+  fn update(
+    &mut self,
+    txn: &mut impl DbTxn,
+    active_keys: &[(KeyFor<S>, LifetimeStage)],
+    update: SchedulerUpdate<S>,
+  ) -> HashMap<Vec<u8>, Vec<EventualityFor<S>>> {
+    // Accumulate all the outputs
+    for key in active_keys {
+      Self::accumulate_outputs(txn, key.0, update.outputs());
+    }
+
+    let mut fee_rates: HashMap<Coin, _> = todo!("TODO");
+
+    // Create the transactions for the forwards/burns
+    {
+      let mut planned_txs = vec![];
+      for forward in update.forwards() {
+        let forward_to_key = active_keys.last().unwrap();
+        assert_eq!(forward_to_key.1, LifetimeStage::Active);
+
+        let Some(plan) = P::plan_transaction_with_fee_amortization(
+          // This uses 0 for the operating costs as we don't incur any here
+          &mut 0,
+          fee_rates[&forward.balance().coin],
+          vec![forward.clone()],
+          vec![Payment::new(P::forwarding_address(forward_to_key.0), forward.balance(), None)],
+          None,
+        ) else {
+          continue;
+        };
+        planned_txs.push(plan);
+      }
+      for to_return in update.returns() {
+        let out_instruction =
+          Payment::new(to_return.address().clone(), to_return.output().balance(), None);
+        let Some(plan) = P::plan_transaction_with_fee_amortization(
+          // This uses 0 for the operating costs as we don't incur any here
+          &mut 0,
+          fee_rates[&out_instruction.balance().coin],
+          vec![to_return.output().clone()],
+          vec![out_instruction],
+          None,
+        ) else {
+          continue;
+        };
+        planned_txs.push(plan);
+      }
+
+      // TODO: Send the transactions off for signing
+      // TODO: Return the eventualities
+      todo!("TODO")
+    }
+  }
+
+  fn fulfill(
+    &mut self,
+    txn: &mut impl DbTxn,
+    active_keys: &[(KeyFor<S>, LifetimeStage)],
+    payments: Vec<Payment<AddressFor<S>>>,
+  ) -> HashMap<Vec<u8>, Vec<EventualityFor<S>>> {
+    // TODO: Find the key to use for fulfillment
+    // TODO: Sort outputs and payments by amount
+    // TODO: For as long as we don't have sufficiently aggregated inputs to handle all payments,
+    // aggregate
+    // TODO: Create the tree for the payments
+    todo!("TODO")
+  }
+}