Use a single transaction to generate exchange amounts

TODO.md

@@ -1,5 +1,6 @@
-- [ ] Add a locking mechanism, to either a start a tx per account, or use the storage engine as a lock mechanism (to lock the utxos)
 - [ ] Optimize `select_inputs_from_accounts` to return a single change operation instead of a vector
+- [ ] Improve read performance with SQLite
+- [ ] Add a locking mechanism, to either a start a tx per account, or use the storage engine as a lock mechanism (to lock the utxos)
 - [ ] Add ability to query accounts in a point in time
 - [ ] Write other servers, other than the restful server
 - [ ] Add caching layer: This cache layer can built on top of the utxo::ledger, because all operations can be safely cached until a new transaction referencing their account is issued, by that point, all the caches related to anaccount can be evicted

utxo/src/ledger.rs

@@ -41,20 +41,26 @@ where
         &self.asset_manager
     }
 
-    /// Selects all inputs to be used in a transaction. The inputs are selected
-    /// from each account in a ascendent order.
+    /// The internal usage is to select unspent payments for each account to
+    /// create new transactions. The external API however does not expose that
+    /// level of usage, instead it exposes a simple API to move funds using
+    /// accounts to debit from and accounts to credit to. A single transaction
+    /// can use multiple accounts to debit and credit, instead of a single
+    /// account.
     ///
-    /// The returned inputs to be used matched exactly with the amounts to be
-    /// spent. Optionally a vector of transactions to be executed before are
-    /// returned. These transactions are `exchange` transactions, and settle
-    /// immediately, because they are internal transactions needed to be sure the
-    /// inputs to be used as input matches exactly the amounts to be spent, to
-    /// avoid locking any exchange amount to the duration of the transaction
-    /// (which is unknown)
-    async fn select_inputs_from_accounts(
+    /// This function selects the unspent payments to be used in a transaction,
+    /// in a descending order (making sure to include any negative deposit).
+    ///
+    /// This function returns a vector of payments to be used as inputs and
+    /// optionally a dependent transaction to be executed first. This
+    /// transaction is an internal transaction and it settles immediately. It is
+    /// used to split an existing payment into two payments, one to be used as
+    /// input and the other to be used as change. This is done to avoid locking
+    /// any change amount until the main transaction settles.
+    async fn select_payments_from_accounts(
         &self,
         payments: Vec<(AccountId, Amount)>,
-    ) -> Result<(Vec<Transaction>, Vec<Payment>), Error> {
+    ) -> Result<(Option<Transaction>, Vec<Payment>), Error> {
         let mut to_spend = HashMap::new();
 
         for (account_id, amount) in payments.into_iter() {
@@ -66,7 +72,8 @@ where
             }
         }
 
-        let mut change_transactions = vec![];
+        let mut change_input = vec![];
+        let mut change_output = vec![];
         let mut payments: Vec<Payment> = vec![];
 
         for ((account, asset), mut to_spend_cents) in to_spend.into_iter() {
@@ -87,42 +94,22 @@ where
                         // There is a change amount, we need to split the last
                         // input into two payment_ids into the same accounts in
                         // a transaction that will settle immediately, otherwise
-                        // the change amount will be unspentable until this
+                        // the change amount will be unspendable until this
                         // transaction settles. By doing so the current
                         // operation will have no change and it can safely take
                         // its time to settle without making any change amount
-                        // unspentable.
+                        // unspendable.
                         let to_spend_cents = to_spend_cents.abs();
                         let input = payments
                             .pop()
                             .ok_or(Error::InsufficientBalance(account.clone(), asset.id))?;
-                        let split_input = Transaction::new(
-                            "Exchange transaction".to_owned(),
-                            // Set the change transaction as settled. This is an
-                            // internal transaction to split an existing payment
-                            // into two. Since this is an internal transaction it
-                            // can be settled immediately.
-                            //
-                            // Because this internal transaction is being settled
-                            // immediately, the other payment can be used right away,
-                            // otherwise it would be locked until the main
-                            // transaction settles.
-                            Status::Settled,
-                            Type::Exchange,
-                            vec![input],
-                            vec![
-                                (account.clone(), asset.new_amount(cents - to_spend_cents)),
-                                (account.clone(), asset.new_amount(to_spend_cents)),
-                            ],
-                        )
-                        .await?;
-                        // Spend the new payment
-                        payments.push(split_input.creates()[0].clone());
-                        // Return the split payment transaction to be executed
-                        // later as a pre-requisite for the new transaction
-                        change_transactions.push(split_input);
 
-                        // Go to the next payment input or exit the loop
+                        change_input.push(input);
+                        change_output
+                            .push((account.clone(), asset.new_amount(cents - to_spend_cents)));
+                        change_output.push((account.clone(), asset.new_amount(to_spend_cents)));
+
+                        // Go to the next payment
                         break;
                     }
                     _ => {
@@ -139,7 +126,31 @@ where
             }
         }
 
-        Ok((change_transactions, payments))
+        let exchange_tx = if change_input.is_empty() {
+            None
+        } else {
+            let total = change_input.len();
+            let split_input = Transaction::new(
+                "Exchange transaction".to_owned(),
+                // Set the change transaction as settled. This is an
+                // internal transaction to split existing payments
+                // into exact new payments, so the main transaction has no
+                // change.
+                Status::Settled,
+                Type::Exchange,
+                change_input,
+                change_output,
+            )
+            .await?;
+
+            for i in 0..total {
+                // Spend the new payment
+                payments.push(split_input.creates()[i * 2].clone());
+            }
+            Some(split_input)
+        };
+
+        Ok((exchange_tx, payments))
     }
 
     /// Creates a new transaction and returns it.
@@ -169,12 +180,10 @@ where
         from: Vec<(AccountId, Amount)>,
         to: Vec<(AccountId, Amount)>,
     ) -> Result<Transaction, Error> {
-        let (change_transactions, payments) = self.select_inputs_from_accounts(from).await?;
-
-        for mut change_tx in change_transactions.into_iter() {
+        let (change_transaction, payments) = self.select_payments_from_accounts(from).await?;
+        if let Some(mut change_tx) = change_transaction {
             change_tx.persist(&self.storage).await?;
         }
-
         let mut transaction =
             Transaction::new(reference, status, Type::Transaction, payments, to).await?;
         transaction.persist(&self.storage).await?;
@@ -224,7 +233,7 @@ where
         reference: String,
     ) -> Result<Transaction, Error> {
         let (change_transactions, payments) = self
-            .select_inputs_from_accounts(vec![(account.clone(), amount)])
+            .select_payments_from_accounts(vec![(account.clone(), amount)])
             .await?;
         for mut change_tx in change_transactions.into_iter() {
             change_tx.persist(&self.storage).await?;

utxo/src/sqlite/mod.rs

@@ -72,7 +72,7 @@ impl<'a> SQLite<'a> {
                 "updated_at" DATETIME DEFAULT CURRENT_TIMESTAMP,
                 PRIMARY KEY("account_id", "transaction_id")
             );
-            CREATE INDEX IF NOT EXISTS "type" ON "transaction_accounts" ("account_id", "type", "created_at");
+            CREATE INDEX IF NOT EXISTS "type" ON "transaction_accounts" ("account_id", "type", "created_at" DESC);
             CREATE TABLE IF NOT EXISTS "transaction_input_payments" (
                 "transaction_id" VARCHAR(66) NOT NULL,
                 "payment_transaction_id" VARCHAR(66) NOT NULL,
@@ -206,7 +206,6 @@ impl<'a> SQLite<'a> {
             .map_err(|_| Error::Storage("Invalid spent_by_status".to_string()))?;
 
         if spent_by.is_some() != spent_status.is_some() {
-            panic!("{:?} {:?}", spent_by, spent_status);
             return Err(Error::Storage(
                 "Invalid spent_by and spent_by_status combination".to_string(),
             ));

utxo/src/tests/mod.rs

@@ -5,6 +5,30 @@ use crate::{
 };
 use sqlx::sqlite::SqlitePoolOptions;
 
+pub async fn get_persistance_instance(
+    name: &str,
+) -> (
+    AssetManager,
+    Ledger<'static, Batch<'static>, SQLite<'static>>,
+) {
+    let pool = SqlitePoolOptions::new()
+        .max_connections(1)
+        .idle_timeout(None)
+        .max_lifetime(None)
+        .connect(format!("sqlite:///tmp/{}.db", name).as_str())
+        .await
+        .expect("pool");
+
+    let assets = AssetManager::new(vec![
+        AssetDefinition::new(1, "BTC", 8),
+        AssetDefinition::new(2, "USD", 4),
+    ]);
+
+    let db = SQLite::new(pool, assets.clone());
+    db.setup().await.expect("setup");
+    (assets.clone(), Ledger::new(db, assets))
+}
+
 pub async fn get_instance() -> (
     AssetManager,
     Ledger<'static, Batch<'static>, SQLite<'static>>,
@@ -14,7 +38,6 @@ pub async fn get_instance() -> (
         .idle_timeout(None)
         .max_lifetime(None)
         .connect(":memory:")
-        //.connect("sqlite:///tmp/test.db")
         .await
         .expect("pool");
 
@@ -56,4 +79,5 @@ pub async fn deposit(
 
 mod deposit;
 mod negative_deposit;
+mod tx;
 mod withdrawal;

utxo/src/tests/withdrawal.rs

@@ -2,7 +2,7 @@ use super::{deposit, get_instance, withdrawal};
 use crate::{AccountId, Status};
 
 #[tokio::test]
-async fn deposit_and_transfer_and_withdrawal() {
+async fn deposit_transfer_and_withdrawal() {
     let source = "account1".parse::<AccountId>().expect("account");
     let dest = "account2".parse::<AccountId>().expect("account");
     let fee = "fee".parse::<AccountId>().expect("account");