Merge branch 'blockchain-overlay'

Cargo.lock

@@ -1157,6 +1157,7 @@ dependencies = [
  "serde_json",
  "simplelog",
  "sled",
+ "sled-overlay",
  "smol",
  "socket2",
  "sqlx",
@@ -3837,6 +3838,15 @@ dependencies = [
  "parking_lot 0.11.2",
 ]
 
+[[package]]
+name = "sled-overlay"
+version = "0.0.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "562d0d1dc5940eae7db86d162f4682db65cdcd8b838d8e05163a70774d52ea7c"
+dependencies = [
+ "sled",
+]
+
 [[package]]
 name = "slice-group-by"
 version = "0.3.0"

Cargo.toml

@@ -129,6 +129,7 @@ sqlx = {version = "0.6.2", features = ["runtime-async-std-rustls", "sqlite"], op
 
 # Blockchain store
 sled = {version = "0.34.7", optional = true}
+sled-overlay = {version = "0.0.3", optional = true}
 
 [dev-dependencies]
 clap = {version = "4.1.4", features = ["derive"]}
@@ -156,6 +157,7 @@ blockchain = [
     "lazy_static",
     "rand",
     "sled",
+    "sled-overlay",
     "sqlx",
     "url",
 

bin/darkfid/src/main.rs

@@ -222,6 +222,9 @@ impl RequestHandler for Darkfid {
             Some("blockchain.subscribe_blocks") => {
                 return self.blockchain_subscribe_blocks(req.id, params).await
             }
+            Some("blockchain.subscribe_err_txs") => {
+                return self.blockchain_subscribe_err_txs(req.id, params).await
+            }
             Some("blockchain.lookup_zkas") => {
                 return self.blockchain_lookup_zkas(req.id, params).await
             }

bin/darkfid/src/rpc_blockchain.rs

@@ -144,6 +144,24 @@ impl Darkfid {
         JsonSubscriber::new(blocks_subscriber).into()
     }
 
+    // RPCAPI:
+    // Initializes a subscription to erroneous transactions notifications.
+    // Once a subscription is established, `darkfid` will send JSON-RPC notifications of
+    // erroneous transactions to the subscriber.
+    //
+    // --> {"jsonrpc": "2.0", "method": "blockchain.subscribe_err_txs", "params": [], "id": 1}
+    // <-- {"jsonrpc": "2.0", "method": "blockchain.subscribe_err_txs", "params": [`tx_hash`]}
+    pub async fn blockchain_subscribe_err_txs(&self, id: Value, params: &[Value]) -> JsonResult {
+        if !params.is_empty() {
+            return JsonError::new(InvalidParams, None, id).into()
+        }
+
+        let err_txs_subscriber =
+            self.validator_state.read().await.subscribers.get("err_txs").unwrap().clone();
+
+        JsonSubscriber::new(err_txs_subscriber).into()
+    }
+
     // RPCAPI:
     // Performs a lookup of zkas bincodes for a given contract ID and returns all of
     // them, including their namespace.

bin/darkfid/src/rpc_tx.rs

@@ -64,9 +64,17 @@ impl Darkfid {
         };
 
         // Simulate state transition
-        if let Err(e) = self.validator_state.read().await.verify_transactions(&[tx], false).await {
-            error!("[RPC] tx.broadcast: Failed to validate state transition: {}", e);
-            return server_error(RpcError::TxSimulationFail, id, None)
+        match self.validator_state.read().await.verify_transactions(&[tx], false).await {
+            Ok(erroneous_txs) => {
+                if !erroneous_txs.is_empty() {
+                    error!("[RPC] tx.simulate: invalid transaction provided");
+                    return server_error(RpcError::TxSimulationFail, id, None)
+                }
+            }
+            Err(e) => {
+                error!("[RPC] tx.simulate: Failed to validate state transition: {}", e);
+                return server_error(RpcError::TxSimulationFail, id, None)
+            }
         };
 
         JsonResponse::new(json!(true), id).into()
@@ -116,11 +124,18 @@ impl Darkfid {
             }
         } else {
             // We'll perform the state transition check here.
-            if let Err(e) =
-                self.validator_state.read().await.verify_transactions(&[tx.clone()], false).await
+            match self.validator_state.read().await.verify_transactions(&[tx.clone()], false).await
             {
-                error!("[RPC] tx.broadcast: Failed to validate state transition: {}", e);
-                return server_error(RpcError::TxSimulationFail, id, None)
+                Ok(erroneous_txs) => {
+                    if !erroneous_txs.is_empty() {
+                        error!("[RPC] tx.broadcast: invalid transaction provided");
+                        return server_error(RpcError::TxSimulationFail, id, None)
+                    }
+                }
+                Err(e) => {
+                    error!("[RPC] tx.broadcast: Failed to validate state transition: {}", e);
+                    return server_error(RpcError::TxSimulationFail, id, None)
+                }
             };
         }
 

bin/darkfid/src/rpc_wallet.rs

@@ -166,6 +166,24 @@ impl Darkfid {
                     continue
                 }
 
+                QueryType::Text => {
+                    let Some(ref row) = row else {
+                        error!("[RPC] wallet.query_row_single: Got None for QueryType::Text");
+                        return server_error(RpcError::NoRowsFoundInWallet, id, None)
+                    };
+
+                    let value: String = match row.try_get(col) {
+                        Ok(v) => v,
+                        Err(e) => {
+                            error!("[RPC] wallet.query_row_single: {}", e);
+                            return JsonError::new(ParseError, None, id).into()
+                        }
+                    };
+
+                    ret.push(json!(value));
+                    continue
+                }
+
                 _ => unreachable!(),
             }
         }
@@ -275,6 +293,18 @@ impl Darkfid {
                         row_ret.push(json!(value));
                     }
 
+                    QueryType::Text => {
+                        let value: String = match row.try_get(col) {
+                            Ok(v) => v,
+                            Err(e) => {
+                                error!("[RPC] wallet.query_row_multi: {}", e);
+                                return JsonError::new(ParseError, None, id).into()
+                            }
+                        };
+
+                        row_ret.push(json!(value));
+                    }
+
                     _ => unreachable!(),
                 }
             }
@@ -322,6 +352,7 @@ impl Darkfid {
 
                     query = query.bind(val);
                 }
+
                 QueryType::Blob => {
                     let val: Vec<u8> = match serde_json::from_value(pair[1].clone()) {
                         Ok(v) => v,
@@ -361,6 +392,18 @@ impl Darkfid {
                     query = query.bind(val);
                 }
 
+                QueryType::Text => {
+                    let val: String = match serde_json::from_value(pair[1].clone()) {
+                        Ok(v) => v,
+                        Err(e) => {
+                            error!("[RPC] wallet.exec_sql: Failed casting value to String: {}", e);
+                            return JsonError::new(ParseError, None, id).into()
+                        }
+                    };
+
+                    query = query.bind(val);
+                }
+
                 _ => return JsonError::new(InvalidParams, None, id).into(),
             }
         }

bin/drk/src/main.rs

@@ -70,6 +70,9 @@ mod rpc_blockchain;
 mod cli_util;
 use cli_util::{parse_token_pair, parse_value_pair};
 
+/// Wallet functionality related to drk operations
+mod wallet;
+
 /// Wallet functionality related to DAO
 mod wallet_dao;
 use wallet_dao::DaoParams;
@@ -80,6 +83,9 @@ mod wallet_money;
 /// Wallet functionality related to arbitrary tokens
 mod wallet_token;
 
+/// Wallet functionality related to transactions history
+mod wallet_txs_history;
+
 #[derive(Parser)]
 #[command(about = cli_desc!())]
 struct Args {
@@ -189,13 +195,9 @@ enum Subcmd {
     /// Read a transaction from stdin and broadcast it
     Broadcast,
 
-    /// Subscribe to incoming blocks from darkfid
-    ///
-    /// This subscription will listen for incoming blocks from darkfid and look
-    /// through their transactions to see if there's any that interest us.
-    /// With `drk` we look at transactions calling the money contract so we can
-    /// find coins sent to us and fill our wallet with the necessary metadata.
-    Subscribe,
+    /// Subscribe to incoming notifications from darkfid
+    #[command(subcommand)]
+    Subscribe(SubscribeSubcmd),
 
     /// DAO functionalities
     #[command(subcommand)]
@@ -354,10 +356,29 @@ enum ExplorerSubcmd {
     FetchTx {
         /// Transaction hash
         tx_hash: String,
+
+        #[arg(long)]
+        /// Print the full transaction information
+        full: bool,
+
+        #[arg(long)]
+        /// Encode transaction to base58
+        encode: bool,
     },
 
     /// Read a transaction from stdin and simulate it
     SimulateTx,
+
+    /// Fetch broadcasted transactions history
+    TxsHistory {
+        /// Fetch specific history record (optional)
+        tx_hash: Option<String>,
+
+        #[arg(long)]
+        /// Encode specific history record transaction
+        /// to base58.
+        encode: bool,
+    },
 }
 
 #[derive(Subcommand)]
@@ -420,6 +441,19 @@ enum TokenSubcmd {
     },
 }
 
+#[derive(Subcommand)]
+enum SubscribeSubcmd {
+    /// This subscription will listen for incoming blocks from darkfid and look
+    /// through their transactions to see if there's any that interest us.
+    /// With `drk` we look at transactions calling the money contract so we can
+    /// find coins sent to us and fill our wallet with the necessary metadata.
+    Blocks,
+
+    /// This subscription will listen for erroneous transactions that got
+    /// removed from darkfid mempool.
+    Transactions,
+}
+
 pub struct Drk {
     pub rpc_client: RpcClient,
 }
@@ -492,6 +526,7 @@ async fn main() -> Result<()> {
             let drk = Drk::new(args.endpoint).await?;
 
             if initialize {
+                drk.initialize_wallet().await?;
                 drk.initialize_money().await?;
                 drk.initialize_dao().await?;
                 return Ok(())
@@ -799,15 +834,27 @@ async fn main() -> Result<()> {
             Ok(())
         }
 
-        Subcmd::Subscribe => {
-            let drk = Drk::new(args.endpoint.clone()).await?;
+        Subcmd::Subscribe(cmd) => match cmd {
+            SubscribeSubcmd::Blocks => {
+                let drk = Drk::new(args.endpoint.clone()).await?;
 
-            drk.subscribe_blocks(args.endpoint)
-                .await
-                .with_context(|| "Block subscription failed")?;
+                drk.subscribe_blocks(args.endpoint.clone())
+                    .await
+                    .with_context(|| "Block subscription failed")?;
 
-            Ok(())
-        }
+                Ok(())
+            }
+
+            SubscribeSubcmd::Transactions => {
+                let drk = Drk::new(args.endpoint.clone()).await?;
+
+                drk.subscribe_err_txs(args.endpoint)
+                    .await
+                    .with_context(|| "Erroneous transactions subscription failed")?;
+
+                Ok(())
+            }
+        },
 
         Subcmd::Scan { reset, list, checkpoint } => {
             let drk = Drk::new(args.endpoint).await?;
@@ -1037,7 +1084,7 @@ async fn main() -> Result<()> {
         },
 
         Subcmd::Explorer(cmd) => match cmd {
-            ExplorerSubcmd::FetchTx { tx_hash } => {
+            ExplorerSubcmd::FetchTx { tx_hash, full, encode } => {
                 let tx_hash = blake3::Hash::from_hex(&tx_hash)?;
 
                 let drk = Drk::new(args.endpoint).await?;
@@ -1047,14 +1094,22 @@ async fn main() -> Result<()> {
                 {
                     tx
                 } else {
-                    eprintln!("tx not found");
+                    eprintln!("Transaction was not found");
                     exit(1);
                 };
 
                 // Make sure the tx is correct
                 assert_eq!(tx.hash(), tx_hash);
 
+                if encode {
+                    println!("{}", bs58::encode(&serialize(&tx)).into_string());
+                    exit(1)
+                }
+
                 println!("Transaction ID: {}", tx_hash);
+                if full {
+                    println!("{:?}", tx);
+                }
 
                 Ok(())
             }
@@ -1076,6 +1131,43 @@ async fn main() -> Result<()> {
 
                 Ok(())
             }
+
+            ExplorerSubcmd::TxsHistory { tx_hash, encode } => {
+                let drk = Drk::new(args.endpoint).await?;
+
+                if let Some(c) = tx_hash {
+                    let (tx_hash, status, tx) = drk.get_tx_history_record(&c).await?;
+
+                    if encode {
+                        println!("{}", bs58::encode(&serialize(&tx)).into_string());
+                        exit(1)
+                    }
+
+                    println!("Transaction ID: {}", tx_hash);
+                    println!("Status: {}", status);
+                    println!("{:?}", tx);
+
+                    return Ok(())
+                }
+
+                let map = drk.get_txs_history().await?;
+
+                // Create a prettytable with the new data:
+                let mut table = Table::new();
+                table.set_format(*format::consts::FORMAT_NO_BORDER_LINE_SEPARATOR);
+                table.set_titles(row!["Transaction Hash", "Status"]);
+                for (txs_hash, status) in map.iter() {
+                    table.add_row(row![txs_hash, status]);
+                }
+
+                if table.is_empty() {
+                    println!("No transactions found");
+                } else {
+                    println!("{}", table);
+                }
+
+                Ok(())
+            }
         },
 
         Subcmd::Alias(cmd) => match cmd {

bin/drk/src/rpc_blockchain.rs

@@ -94,6 +94,7 @@ impl Drk {
                     eprintln!("Deserialized successfully. Scanning block...");
                     self.scan_block_money(&block_data).await?;
                     self.scan_block_dao(&block_data).await?;
+                    self.update_tx_history_records_status(&block_data.txs, "Finalized").await?;
                 }
 
                 JsonResult::Error(e) => {
@@ -169,13 +170,8 @@ impl Drk {
 
         let txid = serde_json::from_value(rep)?;
 
-        // At this point the tx is successfully broadcasted. We can add the
-        // temp data into the wallet. Once scanned, it should mean that the
-        // transaction was finalized, so at that point we actually add the
-        // missing data. For now it'll be in an "unconfirmed" state.
-        // TODO: Do the same for Money::*
-        //self.wallet_apply_unconfirmed_dao_data(tx).await?;
-        //self.wallet_apply_unconfirmed_money_data(tx).await?;
+        // Store transactions history record
+        self.insert_tx_history_record(tx).await?;
 
         Ok(txid)
     }
@@ -235,6 +231,7 @@ impl Drk {
             self.reset_daos().await?;
             self.reset_dao_proposals().await?;
             self.reset_dao_votes().await?;
+            self.update_all_tx_history_records_status("Rejected").await?;
             0
         } else {
             self.last_scanned_slot().await?
@@ -280,6 +277,7 @@ impl Drk {
                 eprintln!("Found");
                 self.scan_block_money(&block).await?;
                 self.scan_block_dao(&block).await?;
+                self.update_tx_history_records_status(&block.txs, "Finalized").await?;
             } else {
                 eprintln!("Not found");
                 // Write down the slot number into back to the wallet
@@ -299,4 +297,59 @@ impl Drk {
 
         Ok(())
     }
+
+    /// Subscribes to darkfid's JSON-RPC notification endpoint that serves
+    /// erroneous transactions rejections.
+    pub async fn subscribe_err_txs(&self, endpoint: Url) -> Result<()> {
+        eprintln!("Subscribing to receive notifications of erroneous transactions");
+        let subscriber = Subscriber::new();
+        let subscription = subscriber.clone().subscribe().await;
+
+        let rpc_client = RpcClient::new(endpoint).await?;
+
+        let req = JsonRequest::new("blockchain.subscribe_err_txs", json!([]));
+        task::spawn(async move { rpc_client.subscribe(req, subscriber).await.unwrap() });
+        eprintln!("Detached subscription to background");
+        eprintln!("All is good. Waiting for erroneous transactions notifications...");
+
+        let e = loop {
+            match subscription.receive().await {
+                JsonResult::Notification(n) => {
+                    eprintln!("Got erroneous transaction notification from darkfid subscription");
+                    if n.method != "blockchain.subscribe_err_txs" {
+                        break anyhow!("Got foreign notification from darkfid: {}", n.method)
+                    }
+
+                    let Some(params) = n.params.as_array() else {
+                        break anyhow!("Received notification params are not an array")
+                    };
+
+                    if params.len() != 1 {
+                        break anyhow!("Notification parameters are not len 1")
+                    }
+
+                    let params = n.params.as_array().unwrap()[0].as_str().unwrap();
+                    let bytes = bs58::decode(params).into_vec()?;
+
+                    let tx_hash: String = deserialize(&bytes)?;
+                    eprintln!("===================================");
+                    eprintln!("Erroneous transaction: {}", tx_hash);
+                    eprintln!("===================================");
+                    self.update_tx_history_record_status(&tx_hash, "Rejected").await?;
+                }
+
+                JsonResult::Error(e) => {
+                    // Some error happened in the transmission
+                    break anyhow!("Got error from JSON-RPC: {:?}", e)
+                }
+
+                x => {
+                    // And this is weird
+                    break anyhow!("Got unexpected data from JSON-RPC: {:?}", x)
+                }
+            }
+        };
+
+        Err(e)
+    }
 }

bin/drk/src/wallet.rs

@@ -0,0 +1,43 @@
+/* This file is part of DarkFi (https://dark.fi)
+ *
+ * Copyright (C) 2020-2023 Dyne.org foundation
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+use anyhow::Result;
+use darkfi::rpc::jsonrpc::JsonRequest;
+use serde_json::json;
+
+use super::Drk;
+
+impl Drk {
+    /// Initialize wallet with tables for drk
+    pub async fn initialize_wallet(&self) -> Result<()> {
+        let wallet_schema = include_str!("../wallet.sql");
+
+        // We perform a request to darkfid with the schema to initialize
+        // the necessary tables in the wallet.
+        let req = JsonRequest::new("wallet.exec_sql", json!([wallet_schema]));
+        let rep = self.rpc_client.request(req).await?;
+
+        if rep == true {
+            eprintln!("Successfully initialized wallet schema for drk");
+        } else {
+            eprintln!("[initialize_wallet] Got unexpected reply from darkfid: {}", rep);
+        }
+
+        Ok(())
+    }
+}

bin/drk/src/wallet_txs_history.rs

@@ -0,0 +1,191 @@
+/* This file is part of DarkFi (https://dark.fi)
+ *
+ * Copyright (C) 2020-2023 Dyne.org foundation
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+use anyhow::{anyhow, Result};
+use darkfi::{rpc::jsonrpc::JsonRequest, tx::Transaction, wallet::walletdb::QueryType};
+use darkfi_serial::{deserialize, serialize};
+use serde_json::json;
+
+use super::Drk;
+
+// Wallet SQL table constant names. These have to represent the `wallet.sql`
+// SQL schema.
+const WALLET_TXS_HISTORY_TABLE: &str = "transactions_history";
+const WALLET_TXS_HISTORY_COL_TX_HASH: &str = "transaction_hash";
+const WALLET_TXS_HISTORY_COL_STATUS: &str = "status";
+const WALLET_TXS_HISTORY_COL_TX: &str = "tx";
+
+impl Drk {
+    /// Fetch all transactions history records, excluding bytes column.
+    pub async fn get_txs_history(&self) -> Result<Vec<(String, String)>> {
+        let mut ret = vec![];
+
+        let query = format!(
+            "SELECT {}, {} FROM {};",
+            WALLET_TXS_HISTORY_COL_TX_HASH, WALLET_TXS_HISTORY_COL_STATUS, WALLET_TXS_HISTORY_TABLE
+        );
+
+        let params = json!([
+            query,
+            QueryType::Text as u8,
+            WALLET_TXS_HISTORY_COL_TX_HASH,
+            QueryType::Text as u8,
+            WALLET_TXS_HISTORY_COL_STATUS,
+        ]);
+
+        let req = JsonRequest::new("wallet.query_row_multi", params);
+        let rep = self.rpc_client.request(req).await?;
+
+        let Some(rows) = rep.as_array() else {
+            return Err(anyhow!("[txs_history] Unexpected response from darkfid: {}", rep));
+        };
+
+        for row in rows {
+            let tx_hash: String = serde_json::from_value(row[0].clone())?;
+            let status: String = serde_json::from_value(row[1].clone())?;
+            ret.push((tx_hash, status));
+        }
+
+        Ok(ret)
+    }
+
+    /// Get a transaction history record.
+    pub async fn get_tx_history_record(
+        &self,
+        tx_hash: &str,
+    ) -> Result<(String, String, Transaction)> {
+        let query = format!(
+            "SELECT * FROM {} WHERE {} = {};",
+            WALLET_TXS_HISTORY_TABLE, WALLET_TXS_HISTORY_COL_TX_HASH, tx_hash
+        );
+
+        let params = json!([
+            query,
+            QueryType::Text as u8,
+            WALLET_TXS_HISTORY_COL_TX_HASH,
+            QueryType::Text as u8,
+            WALLET_TXS_HISTORY_COL_STATUS,
+            QueryType::Blob as u8,
+            WALLET_TXS_HISTORY_COL_TX,
+        ]);
+
+        let req = JsonRequest::new("wallet.query_row_single", params);
+        let rep = self.rpc_client.request(req).await?;
+
+        let Some(arr) = rep.as_array() else {
+            return Err(anyhow!("[get_tx_history_record] Unexpected response from darkfid: {}", rep));
+        };
+
+        if arr.len() != 3 {
+            return Err(anyhow!("Did not find transaction record with hash {}", tx_hash))
+        }
+
+        let tx_hash: String = serde_json::from_value(arr[0].clone())?;
+
+        let status: String = serde_json::from_value(arr[1].clone())?;
+
+        let tx_bytes: Vec<u8> = serde_json::from_value(arr[2].clone())?;
+        let tx: Transaction = deserialize(&tx_bytes)?;
+
+        Ok((tx_hash, status, tx))
+    }
+
+    /// Insert a [`Transaction`] history record into the wallet.
+    pub async fn insert_tx_history_record(&self, tx: &Transaction) -> Result<()> {
+        let query = format!(
+            "INSERT INTO {} ({}, {}, {}) VALUES (?1, ?2, ?3);",
+            WALLET_TXS_HISTORY_TABLE,
+            WALLET_TXS_HISTORY_COL_TX_HASH,
+            WALLET_TXS_HISTORY_COL_STATUS,
+            WALLET_TXS_HISTORY_COL_TX,
+        );
+
+        let params = json!([
+            query,
+            QueryType::Text as u8,
+            tx.hash().to_string(),
+            QueryType::Text as u8,
+            "Broadcasted",
+            QueryType::Blob as u8,
+            serialize(tx),
+        ]);
+
+        let req = JsonRequest::new("wallet.exec_sql", params);
+        let _ = self.rpc_client.request(req).await?;
+
+        Ok(())
+    }
+
+    /// Update a transactions history record status to the given one.
+    pub async fn update_tx_history_record_status(&self, tx_hash: &str, status: &str) -> Result<()> {
+        let query = format!(
+            "UPDATE {} SET {} = ?1 WHERE {} = ?2;",
+            WALLET_TXS_HISTORY_TABLE, WALLET_TXS_HISTORY_COL_STATUS, WALLET_TXS_HISTORY_COL_TX_HASH,
+        );
+
+        let params = json!([query, QueryType::Text as u8, status, QueryType::Text as u8, tx_hash,]);
+
+        let req = JsonRequest::new("wallet.exec_sql", params);
+        let _ = self.rpc_client.request(req).await?;
+
+        Ok(())
+    }
+
+    /// Update given transactions history record statuses to the given one.
+    pub async fn update_tx_history_records_status(
+        &self,
+        txs: &Vec<Transaction>,
+        status: &str,
+    ) -> Result<()> {
+        if txs.is_empty() {
+            return Ok(())
+        }
+
+        let txs_hashes: Vec<String> = txs.into_iter().map(|tx| tx.hash().to_string()).collect();
+        let txs_hashes_string = format!("{:?}", txs_hashes).replace("[", "(").replace("]", ")");
+        let query = format!(
+            "UPDATE {} SET {} = ?1 WHERE {} IN {};",
+            WALLET_TXS_HISTORY_TABLE,
+            WALLET_TXS_HISTORY_COL_STATUS,
+            WALLET_TXS_HISTORY_COL_TX_HASH,
+            txs_hashes_string
+        );
+
+        let params = json!([query, QueryType::Text as u8, status,]);
+
+        let req = JsonRequest::new("wallet.exec_sql", params);
+        let _ = self.rpc_client.request(req).await?;
+
+        Ok(())
+    }
+
+    /// Update all transaction history records statuses to the given one.
+    pub async fn update_all_tx_history_records_status(&self, status: &str) -> Result<()> {
+        let query = format!(
+            "UPDATE {} SET {} = ?1",
+            WALLET_TXS_HISTORY_TABLE, WALLET_TXS_HISTORY_COL_STATUS,
+        );
+
+        let params = json!([query, QueryType::Text as u8, status,]);
+
+        let req = JsonRequest::new("wallet.exec_sql", params);
+        let _ = self.rpc_client.request(req).await?;
+
+        Ok(())
+    }
+}

bin/drk/wallet.sql

@@ -0,0 +1,9 @@
+-- Wallet definitions for drk.
+-- We store data that is needed for wallet operations.
+
+-- Broadcasted transactions history
+CREATE TABLE IF NOT EXISTS transactions_history (
+    transaction_hash TEXT PRIMARY KEY NOT NULL,
+    status TEXT NOT NULL,
+	tx BLOB NOT NULL
+);

contrib/localnet/darkfid/README.md

@@ -35,8 +35,8 @@ work, we also need to subscribe to their RPC endpoints so we can scan
 incoming blocks and add them to our wallet.
 
 ```
-$ ./drk -e tcp://127.0.0.1:8440 subscribe
-$ ./drk -e tcp://127.0.0.1:8540 subscribe
+$ ./drk -e tcp://127.0.0.1:8440 subscribe blocks
+$ ./drk -e tcp://127.0.0.1:8540 subscribe blocks
 ```
 
 And now we can execute our airdrop calls:

doc/src/testnet/airdrop.md

@@ -19,7 +19,7 @@ On success, you should see a transaction ID. If successful,
 the airdrop transactions will how be in the consensus' mempool,
 waiting for inclusion in the next block. Depending on the network,
 finalization of the blocks could take some time. You'll have to wait
-for this to happen.  If your `drk subscribe` is running, then after
+for this to happen.  If your `drk subscribe blocks` is running, then after
 some time your balance should be in your wallet.
 
 ![pablo-waiting0](pablo0.jpg)
@@ -101,6 +101,6 @@ $ ./drk broadcast < mint_tx
 ```
 
 Now the transaction should be published to the network. If you have
-an active block subscription (which you can do with `drk subscribe`),
+an active block subscription (which you can do with `drk subscribe blocks`),
 then when the transaction is finalized, your wallet should have your
 new tokens listed when you request to see the balance.

doc/src/testnet/node.md

@@ -71,7 +71,7 @@ and then to subscribe to new blocks:
 
 ```
 $ ./drk scan
-$ ./drk subscribe
+$ ./drk subscribe blocks
 ```
 
 Now you can leave the subscriber running. In case you stop it, just

src/blockchain/contract_store.rs

@@ -15,6 +15,7 @@ r* This program is distributed in the hope that it will be useful,
  * You should have received a copy of the GNU Affero General Public License
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
+
 use std::io::Cursor;
 
 use darkfi_sdk::crypto::ContractId;
@@ -22,6 +23,7 @@ use darkfi_serial::{deserialize, serialize};
 use log::{debug, error};
 
 use crate::{
+    blockchain::SledDbOverlayPtr,
     runtime::vm_runtime::SMART_CONTRACT_ZKAS_DB_NAME,
     zk::{VerifyingKey, ZkCircuit},
     zkas::ZkBinary,
@@ -57,11 +59,23 @@ impl WasmStore {
 
         Err(Error::WasmBincodeNotFound)
     }
+}
+
+/// Overlay structure over a [`WasmStore`] instance.
+pub struct WasmStoreOverlay(SledDbOverlayPtr);
+
+impl WasmStoreOverlay {
+    pub fn new(overlay: SledDbOverlayPtr) -> Result<Self> {
+        overlay.lock().unwrap().open_tree(SLED_BINCODE_TREE)?;
+        Ok(Self(overlay))
+    }
 
     /// Inserts or replaces the bincode for a given ContractId
     pub fn insert(&self, contract_id: ContractId, bincode: &[u8]) -> Result<()> {
-        if let Err(e) = self.0.insert(serialize(&contract_id), bincode) {
-            error!(target: "blockchain::contractstore", "Failed to insert bincode to WasmStore: {}", e);
+        if let Err(e) =
+            self.0.lock().unwrap().insert(SLED_BINCODE_TREE, &serialize(&contract_id), bincode)
+        {
+            error!(target: "blockchain::contractstoreoverlay", "Failed to insert bincode to WasmStore: {}", e);
             return Err(e.into())
         }
 
@@ -89,58 +103,6 @@ impl ContractStateStore {
         Ok(Self(tree))
     }
 
-    /// Try to initialize a new contract state. Contracts can create a number
-    /// of trees, separated by `tree_name`, which they can then use from the
-    /// smart contract API. `init()` will look into the main `ContractStateStore`
-    /// tree to check if the smart contract was already deployed, and if so
-    /// it will fetch a vector of these states that were initialized. If the
-    /// state was already found, this function will return an error, because
-    /// in this case the handle should be fetched using `lookup()`.
-    /// If the tree was not initialized previously, it will be appended to
-    /// the main `ContractStateStore` tree and a `sled::Tree` handle will be
-    /// returned.
-    pub fn init(
-        &self,
-        db: &sled::Db,
-        contract_id: &ContractId,
-        tree_name: &str,
-    ) -> Result<sled::Tree> {
-        debug!(target: "blockchain::contractstore", "Initializing state tree for {}:{}", contract_id, tree_name);
-
-        let contract_id_bytes = serialize(contract_id);
-        let ptr = contract_id.hash_state_id(tree_name);
-
-        // See if there are existing state trees.
-        // If not, just start with an empty vector.
-        let mut state_pointers: Vec<[u8; 32]> = if self.0.contains_key(&contract_id_bytes)? {
-            let bytes = self.0.get(&contract_id_bytes)?.unwrap();
-            deserialize(&bytes)?
-        } else {
-            vec![]
-        };
-
-        // If the db was never initialized, it should not be in here.
-        if state_pointers.contains(&ptr) {
-            return Err(Error::ContractAlreadyInitialized)
-        }
-
-        // Now we add it so it's marked as initialized
-        state_pointers.push(ptr);
-
-        // We do this as a batch so in case of not being able to open the tree
-        // we don't write that it's initialized.
-        let mut batch = sled::Batch::default();
-        batch.insert(contract_id_bytes, serialize(&state_pointers));
-
-        // We open the tree and return its handle
-        let tree = db.open_tree(ptr)?;
-
-        // On success, apply the batch
-        self.0.apply_batch(batch)?;
-
-        Ok(tree)
-    }
-
     /// Do a lookup of an existing contract state. In order to succeed, the
     /// state must have been previously initialized with `init()`. If the
     /// state has been found, a handle to it will be returned. Otherwise, we
@@ -241,3 +203,83 @@ impl ContractStateStore {
         Ok((zkbin, vk))
     }
 }
+
+/// Overlay structure over a [`ContractStateStore`] instance.
+pub struct ContractStateStoreOverlay(SledDbOverlayPtr);
+
+impl ContractStateStoreOverlay {
+    pub fn new(overlay: SledDbOverlayPtr) -> Result<Self> {
+        overlay.lock().unwrap().open_tree(SLED_CONTRACTS_TREE)?;
+        Ok(Self(overlay))
+    }
+
+    /// Try to initialize a new contract state. Contracts can create a number
+    /// of trees, separated by `tree_name`, which they can then use from the
+    /// smart contract API. `init()` will look into the main `ContractStateStoreOverlay`
+    /// tree to check if the smart contract was already deployed, and if so
+    /// it will fetch a vector of these states that were initialized. If the
+    /// state was already found, this function will return an error, because
+    /// in this case the handle should be fetched using `lookup()`.
+    /// If the tree was not initialized previously, it will be appended to
+    /// the main `ContractStateStoreOverlay` tree and a handle to it will be
+    /// returned.
+    pub fn init(&self, contract_id: &ContractId, tree_name: &str) -> Result<[u8; 32]> {
+        debug!(target: "blockchain::contractstoreoverlay", "Initializing state overlay tree for {}:{}", contract_id, tree_name);
+
+        let contract_id_bytes = serialize(contract_id);
+        let ptr = contract_id.hash_state_id(tree_name);
+        let mut lock = self.0.lock().unwrap();
+
+        // See if there are existing state trees.
+        // If not, just start with an empty vector.
+        let mut state_pointers: Vec<[u8; 32]> =
+            if lock.contains_key(SLED_CONTRACTS_TREE, &contract_id_bytes)? {
+                let bytes = lock.get(SLED_CONTRACTS_TREE, &contract_id_bytes)?.unwrap();
+                deserialize(&bytes)?
+            } else {
+                vec![]
+            };
+
+        // If the db was never initialized, it should not be in here.
+        if state_pointers.contains(&ptr) {
+            return Err(Error::ContractAlreadyInitialized)
+        }
+
+        // Now we add it so it's marked as initialized and create its tree.
+        state_pointers.push(ptr);
+        lock.insert(SLED_CONTRACTS_TREE, &contract_id_bytes, &serialize(&state_pointers))?;
+        lock.open_tree(&ptr)?;
+
+        Ok(ptr)
+    }
+
+    /// Do a lookup of an existing contract state. In order to succeed, the
+    /// state must have been previously initialized with `init()`. If the
+    /// state has been found, a handle to it will be returned. Otherwise, we
+    /// return an error.
+    pub fn lookup(&self, contract_id: &ContractId, tree_name: &str) -> Result<[u8; 32]> {
+        debug!(target: "blockchain::contractstoreoverlay", "Looking up state tree for {}:{}", contract_id, tree_name);
+
+        let contract_id_bytes = serialize(contract_id);
+        let ptr = contract_id.hash_state_id(tree_name);
+        let mut lock = self.0.lock().unwrap();
+
+        // A guard to make sure we went through init()
+        if !lock.contains_key(SLED_CONTRACTS_TREE, &contract_id_bytes)? {
+            return Err(Error::ContractNotFound(contract_id.to_string()))
+        }
+
+        let state_pointers = lock.get(SLED_CONTRACTS_TREE, &contract_id_bytes)?.unwrap();
+        let state_pointers: Vec<[u8; 32]> = deserialize(&state_pointers)?;
+
+        // We assume the tree has been created already, so it should be listed
+        // in this array. If not, that's an error.
+        if !state_pointers.contains(&ptr) {
+            return Err(Error::ContractStateNotFound)
+        }
+
+        // We open the tree and return its handle
+        lock.open_tree(&ptr)?;
+        Ok(ptr)
+    }
+}

src/blockchain/mod.rs

@@ -16,10 +16,15 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
+use std::sync::{Arc, Mutex};
+
 use log::debug;
 
+use darkfi_serial::serialize;
+
 use crate::{
     consensus::{Block, BlockInfo, SlotCheckpoint},
+    tx::Transaction,
     util::time::Timestamp,
     Result,
 };
@@ -31,10 +36,12 @@ pub mod slot_checkpoint_store;
 pub use slot_checkpoint_store::SlotCheckpointStore;
 
 pub mod tx_store;
-pub use tx_store::{PendingTxStore, TxStore};
+pub use tx_store::{PendingTxOrderStore, PendingTxStore, TxStore};
 
 pub mod contract_store;
-pub use contract_store::{ContractStateStore, WasmStore};
+pub use contract_store::{
+    ContractStateStore, ContractStateStoreOverlay, WasmStore, WasmStoreOverlay,
+};
 
 /// Structure holding all sled trees that define the concept of Blockchain.
 #[derive(Clone)]
@@ -53,6 +60,8 @@ pub struct Blockchain {
     pub transactions: TxStore,
     /// Pending transactions sled tree
     pub pending_txs: PendingTxStore,
+    /// Pending transactions order sled tree
+    pub pending_txs_order: PendingTxOrderStore,
     /// Contract states
     pub contracts: ContractStateStore,
     /// Wasm bincodes
@@ -68,6 +77,7 @@ impl Blockchain {
         let slot_checkpoints = SlotCheckpointStore::new(db)?;
         let transactions = TxStore::new(db)?;
         let pending_txs = PendingTxStore::new(db)?;
+        let pending_txs_order = PendingTxOrderStore::new(db)?;
         let contracts = ContractStateStore::new(db)?;
         let wasm_bincode = WasmStore::new(db)?;
 
@@ -79,6 +89,7 @@ impl Blockchain {
             slot_checkpoints,
             transactions,
             pending_txs,
+            pending_txs_order,
             contracts,
             wasm_bincode,
         })
@@ -220,4 +231,80 @@ impl Blockchain {
         };
         Ok(!vec.is_empty())
     }
+
+    /// Insert a given slice of pending transactions into the blockchain database.
+    /// On success, the function returns the transaction hashes in the same order
+    /// as the input transactions.
+    pub fn add_pending_txs(&self, txs: &[Transaction]) -> Result<Vec<blake3::Hash>> {
+        // TODO: Make db writes here completely atomic
+        let txs_hashes = self.pending_txs.insert(&txs)?;
+        self.pending_txs_order.insert(&txs_hashes)?;
+
+        Ok(txs_hashes)
+    }
+
+    /// Retrieve all transactions from the pending tx store.
+    /// Be careful as this will try to load everything in memory.
+    pub fn get_pending_txs(&self) -> Result<Vec<Transaction>> {
+        let txs = self.pending_txs.get_all()?;
+        let indexes = self.pending_txs_order.get_all()?;
+        assert_eq!(txs.len(), indexes.len());
+
+        let mut ret = Vec::with_capacity(txs.len());
+        for index in indexes {
+            ret.push(txs.get(&index.1).unwrap().clone());
+        }
+
+        Ok(ret)
+    }
+
+    /// Remove a given slice of pending transactions from the blockchain database.
+    pub fn remove_pending_txs(&self, txs: &[Transaction]) -> Result<()> {
+        let mut txs_hashes = Vec::with_capacity(txs.len());
+        for tx in txs {
+            let tx_hash = blake3::hash(&serialize(tx));
+            txs_hashes.push(tx_hash);
+        }
+
+        let indexes = self.pending_txs_order.get_all()?;
+        let mut removed_indexes = vec![];
+        for index in indexes {
+            if txs_hashes.contains(&index.1) {
+                removed_indexes.push(index.0);
+            }
+        }
+
+        // TODO: Make db writes here completely atomic
+        self.pending_txs.remove(&txs_hashes)?;
+        self.pending_txs_order.remove(&removed_indexes)?;
+
+        Ok(())
+    }
+}
+
+/// Atomic pointer to sled db overlay.
+pub type SledDbOverlayPtr = Arc<Mutex<sled_overlay::SledDbOverlay>>;
+
+/// Atomic pointer to blockchain overlay.
+pub type BlockchainOverlayPtr = Arc<Mutex<BlockchainOverlay>>;
+
+/// Overlay structure over a [`Blockchain`] instance.
+pub struct BlockchainOverlay {
+    /// Main [`sled_overlay::SledDbOverlay`] to the sled db connection
+    pub overlay: SledDbOverlayPtr,
+    /// Contract states overlay
+    pub contracts: ContractStateStoreOverlay,
+    /// Wasm bincodes overlay
+    pub wasm_bincode: WasmStoreOverlay,
+}
+
+impl BlockchainOverlay {
+    /// Instantiate a new `BlockchainOverlay` over the given [`Blockchain`] instance.
+    pub fn new(blockchain: &Blockchain) -> Result<BlockchainOverlayPtr> {
+        let overlay = Arc::new(Mutex::new(sled_overlay::SledDbOverlay::new(&blockchain.sled_db)));
+        let contracts = ContractStateStoreOverlay::new(overlay.clone())?;
+        let wasm_bincode = WasmStoreOverlay::new(overlay.clone())?;
+
+        Ok(Arc::new(Mutex::new(Self { overlay, contracts, wasm_bincode })))
+    }
 }

src/blockchain/tx_store.rs

@@ -16,12 +16,15 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
+use std::collections::HashMap;
+
 use darkfi_serial::{deserialize, serialize};
 
 use crate::{tx::Transaction, Error, Result};
 
 const SLED_TX_TREE: &[u8] = b"_transactions";
 const SLED_PENDING_TX_TREE: &[u8] = b"_pending_transactions";
+const SLED_PENDING_TX_ORDER_TREE: &[u8] = b"_pending_transactions_order";
 
 /// The `TxStore` is a `sled` tree storing all the blockchain's
 /// transactions where the key is the transaction hash, and the value is
@@ -146,40 +149,98 @@ impl PendingTxStore {
         Ok(self.0.contains_key(tx_hash.as_bytes())?)
     }
 
-    /// Retrieve all transactions from the pending tx store in the form of a tuple
-    /// (`tx_hash`, `tx`).
+    /// Retrieve all transactions from the pending tx store in the form of
+    /// a HashMap with key the transaction hash and value the transaction
+    /// itself.
     /// Be careful as this will try to load everything in memory.
-    pub fn get_all(&self) -> Result<Vec<(blake3::Hash, Transaction)>> {
-        let mut txs = vec![];
+    pub fn get_all(&self) -> Result<HashMap<blake3::Hash, Transaction>> {
+        let mut txs = HashMap::new();
 
         for tx in self.0.iter() {
             let (key, value) = tx.unwrap();
             let hash_bytes: [u8; 32] = key.as_ref().try_into().unwrap();
             let tx = deserialize(&value)?;
-            txs.push((hash_bytes.into(), tx));
+            txs.insert(hash_bytes.into(), tx);
         }
 
         Ok(txs)
     }
 
-    /// Retrieve all transactions from the pending tx store.
+    /// Remove a slice of [`blake3::Hash`] from the pending tx store.
+    /// With sled, the operation is done as a batch.
+    pub fn remove(&self, txs_hashes: &[blake3::Hash]) -> Result<()> {
+        let mut batch = sled::Batch::default();
+
+        for tx_hash in txs_hashes {
+            batch.remove(tx_hash.as_bytes());
+        }
+
+        self.0.apply_batch(batch)?;
+        Ok(())
+    }
+}
+
+/// The `PendingTxOrderStore` is a `sled` tree storing the order of all
+/// the node pending transactions where the key is an incremental value,
+/// and the value is the serialized transaction.
+#[derive(Clone)]
+pub struct PendingTxOrderStore(sled::Tree);
+
+impl PendingTxOrderStore {
+    /// Opens a new or existing `PendingTxOrderStore` on the given sled database.
+    pub fn new(db: &sled::Db) -> Result<Self> {
+        let tree = db.open_tree(SLED_PENDING_TX_ORDER_TREE)?;
+        Ok(Self(tree))
+    }
+
+    /// Insert a slice of [`blake3::Hash`] into the pending tx order store.
+    /// With sled, the operation is done as a batch.
+    pub fn insert(&self, txs_hashes: &[blake3::Hash]) -> Result<()> {
+        let mut batch = sled::Batch::default();
+
+        let mut next_index = match self.0.last()? {
+            Some(n) => {
+                let prev_bytes: [u8; 8] = n.0.as_ref().try_into().unwrap();
+                let prev = u64::from_be_bytes(prev_bytes);
+                prev + 1
+            }
+            None => 0,
+        };
+
+        for txs_hash in txs_hashes {
+            batch.insert(&next_index.to_be_bytes(), txs_hash.as_bytes());
+            next_index += 1;
+        }
+
+        self.0.apply_batch(batch)?;
+        Ok(())
+    }
+
+    /// Retrieve all transactions from the pending tx order store in the form
+    /// of a tuple (`u64`, `blake3::Hash`).
     /// Be careful as this will try to load everything in memory.
-    pub fn get_all_txs(&self) -> Result<Vec<Transaction>> {
-        let txs = self.get_all()?;
-        Ok(txs.iter().map(|x| x.1.clone()).rev().collect())
+    pub fn get_all(&self) -> Result<Vec<(u64, blake3::Hash)>> {
+        let mut txs = vec![];
+
+        for tx in self.0.iter() {
+            let (key, value) = tx.unwrap();
+            let index_bytes: [u8; 8] = key.as_ref().try_into().unwrap();
+            let hash_bytes: [u8; 32] = value.as_ref().try_into().unwrap();
+            let index = u64::from_be_bytes(index_bytes);
+            let hash = blake3::Hash::from(hash_bytes);
+            txs.push((index, hash));
+        }
+
+        Ok(txs)
     }
 
-    /// Remove a slice of [`Transaction`] from the pending tx store.
+    /// Remove a slice of [`u64`] from the pending tx order store.
     /// With sled, the operation is done as a batch.
-    /// The transactions are hashed with BLAKE3 and this hash is used as
-    /// the key to remove.
-    pub fn remove(&self, transactions: &[Transaction]) -> Result<()> {
+    pub fn remove(&self, indexes: &[u64]) -> Result<()> {
         let mut batch = sled::Batch::default();
 
-        for tx in transactions {
-            let serialized = serialize(tx);
-            let tx_hash = blake3::hash(&serialized);
-            batch.remove(tx_hash.as_bytes());
+        for index in indexes {
+            batch.remove(&index.to_be_bytes());
         }
 
         self.0.apply_batch(batch)?;

src/consensus/proto/protocol_sync_consensus.rs

@@ -99,7 +99,7 @@ impl ProtocolSyncConsensus {
             for fork in &lock.consensus.forks {
                 forks.push(fork.clone().into());
             }
-            let pending_txs = match lock.blockchain.pending_txs.get_all_txs() {
+            let pending_txs = match lock.blockchain.get_pending_txs() {
                 Ok(v) => v,
                 Err(e) => {
                     debug!(

src/consensus/task/proposal.rs

@@ -209,7 +209,7 @@ async fn propose_period(consensus_p2p: P2pPtr, state: ValidatorStatePtr) -> bool
     let (won, fork_index, coin_index) =
         state.write().await.consensus.is_slot_leader(sigma1, sigma2);
     let result = if won {
-        state.write().await.propose(processing_slot, fork_index, coin_index, sigma1, sigma2)
+        state.write().await.propose(processing_slot, fork_index, coin_index, sigma1, sigma2).await
     } else {
         Ok(None)
     };

src/consensus/validator.rs

@@ -43,7 +43,7 @@ use super::{
 };
 
 use crate::{
-    blockchain::Blockchain,
+    blockchain::{Blockchain, BlockchainOverlay, BlockchainOverlayPtr},
     rpc::jsonrpc::JsonNotification,
     runtime::vm_runtime::Runtime,
     system::{Subscriber, SubscriberPtr},
@@ -167,12 +167,14 @@ impl ValidatorState {
         ];
 
         info!(target: "consensus::validator", "Deploying native wasm contracts");
+        let blockchain_overlay = BlockchainOverlay::new(&blockchain)?;
         for nc in native_contracts {
             info!(target: "consensus::validator", "Deploying {} with ContractID {}", nc.0, nc.1);
-            let mut runtime = Runtime::new(&nc.2[..], blockchain.clone(), nc.1)?;
+            let mut runtime = Runtime::new(&nc.2[..], blockchain_overlay.clone(), nc.1)?;
             runtime.deploy(&nc.3)?;
             info!(target: "consensus::validator", "Successfully deployed {}", nc.0);
         }
+        blockchain_overlay.lock().unwrap().overlay.lock().unwrap().apply()?;
 
         info!(target: "consensus::validator", "Finished deployment of native wasm contracts");
         // -----END NATIVE WASM CONTRACTS-----
@@ -180,7 +182,9 @@ impl ValidatorState {
         // Here we initialize various subscribers that can export live consensus/blockchain data.
         let mut subscribers = HashMap::new();
         let block_subscriber = Subscriber::new();
+        let err_txs_subscriber = Subscriber::new();
         subscribers.insert("blocks", block_subscriber);
+        subscribers.insert("err_txs", err_txs_subscriber);
 
         let state = Arc::new(RwLock::new(ValidatorState {
             lead_proving_key,
@@ -222,12 +226,20 @@ impl ValidatorState {
         }
 
         info!(target: "consensus::validator", "append_tx(): Starting state transition validation");
-        if let Err(e) = self.verify_transactions(&[tx.clone()], false).await {
-            error!(target: "consensus::validator", "append_tx(): Failed to verify transaction: {}", e);
-            return false
-        };
+        match self.verify_transactions(&[tx.clone()], false).await {
+            Ok(erroneous_txs) => {
+                if !erroneous_txs.is_empty() {
+                    error!(target: "consensus::validator", "append_tx(): Erroneous transaction detected");
+                    return false
+                }
+            }
+            Err(e) => {
+                error!(target: "consensus::validator", "append_tx(): Failed to verify transaction: {}", e);
+                return false
+            }
+        }
 
-        if let Err(e) = self.blockchain.pending_txs.insert(&[tx]) {
+        if let Err(e) = self.blockchain.add_pending_txs(&[tx]) {
             error!(target: "consensus::validator", "append_tx(): Failed to insert transaction to pending txs store: {}", e);
             return false
         }
@@ -239,6 +251,7 @@ impl ValidatorState {
     /// and appends successfull ones to the pending txs store.
     pub async fn append_pending_txs(&mut self, txs: &[Transaction]) {
         let mut filtered_txs = vec![];
+        // Filter already seen transactions
         for tx in txs {
             let tx_hash = blake3::hash(&serialize(tx));
             let tx_in_txstore = match self.blockchain.transactions.contains(&tx_hash) {
@@ -265,24 +278,59 @@ impl ValidatorState {
             filtered_txs.push(tx.clone());
         }
 
+        // Verify transactions and filter erroneous ones
         info!(target: "consensus::validator", "append_pending_txs(): Starting state transition validation");
-        // TODO: verify_transactions should return erroneous txs of the set to ignore them
-        if let Err(e) = self.verify_transactions(&filtered_txs[..], false).await {
-            error!(target: "consensus::validator", "append_pending_txs(): Failed to verify transaction: {}", e);
-            return
+        let erroneous_txs = match self.verify_transactions(&filtered_txs[..], false).await {
+            Ok(erroneous_txs) => erroneous_txs,
+            Err(e) => {
+                error!(target: "consensus::validator", "append_pending_txs(): Failed to verify transactions: {}", e);
+                return
+            }
         };
+        if !erroneous_txs.is_empty() {
+            filtered_txs.retain(|x| !erroneous_txs.contains(&x));
+        }
 
-        if let Err(e) = self.blockchain.pending_txs.insert(&filtered_txs) {
+        if let Err(e) = self.blockchain.add_pending_txs(&filtered_txs) {
             error!(target: "consensus::validator", "append_pending_txs(): Failed to insert transactions to pending txs store: {}", e);
             return
         }
         info!(target: "consensus::validator", "append_pending_txs(): Appended tx to pending txs store");
     }
 
+    /// The node removes erroneous transactions from the pending txs store.
+    async fn purge_pending_txs(&self) -> Result<()> {
+        info!(target: "consensus::validator", "purge_pending_txs(): Removing erroneous transactions from pending transactions store...");
+        let pending_txs = self.blockchain.get_pending_txs()?;
+        if pending_txs.is_empty() {
+            info!(target: "consensus::validator", "purge_pending_txs(): No pending transactions found");
+            return Ok(())
+        }
+        let erroneous_txs = self.verify_transactions(&pending_txs[..], false).await?;
+        if erroneous_txs.is_empty() {
+            info!(target: "consensus::validator", "purge_pending_txs(): No erroneous transactions found");
+            return Ok(())
+        }
+        info!(target: "consensus::validator", "purge_pending_txs(): Removing {} erroneous transactions...", erroneous_txs.len());
+        self.blockchain.remove_pending_txs(&erroneous_txs)?;
+
+        // TODO: Don't hardcode this:
+        let err_txs_subscriber = self.subscribers.get("err_txs").unwrap();
+        for err_tx in erroneous_txs {
+            let tx_hash = blake3::hash(&serialize(&err_tx)).to_hex().as_str().to_string();
+            let params = json!([bs58::encode(&serialize(&tx_hash)).into_string()]);
+            let notif = JsonNotification::new("blockchain.subscribe_err_txs", params);
+            info!(target: "consensus::validator", "purge_pending_txs(): Sending notification about erroneous transaction");
+            err_txs_subscriber.notify(notif).await;
+        }
+
+        Ok(())
+    }
+
     /// Generate a block proposal for the current slot, containing all
     /// pending transactions. Proposal extends the longest fork
     /// chain the node is holding.
-    pub fn propose(
+    pub async fn propose(
         &mut self,
         slot: u64,
         fork_index: i64,
@@ -297,7 +345,12 @@ impl ValidatorState {
         }
 
         // Generate proposal
-        let unproposed_txs = self.unproposed_txs(fork_index)?;
+        let mut unproposed_txs = self.unproposed_txs(fork_index)?;
+        // Verify transactions and filter erroneous ones
+        let erroneous_txs = self.verify_transactions(&unproposed_txs[..], false).await?;
+        if !erroneous_txs.is_empty() {
+            unproposed_txs.retain(|x| !erroneous_txs.contains(&x));
+        }
         let mut tree = BridgeTree::<MerkleNode, MERKLE_DEPTH>::new(100);
         // The following is pretty weird, so something better should be done.
         for tx in &unproposed_txs {
@@ -360,11 +413,11 @@ impl ValidatorState {
         let unproposed_txs = if index == -1 {
             // If index is -1 (canonical blockchain) a new fork will be generated,
             // therefore all unproposed transactions can be included in the proposal.
-            self.blockchain.pending_txs.get_all_txs()?
+            self.blockchain.get_pending_txs()?
         } else {
             // We iterate over the fork chain proposals to find already proposed
             // transactions and remove them from the local unproposed_txs vector.
-            let mut filtered_txs = self.blockchain.pending_txs.get_all_txs()?;
+            let mut filtered_txs = self.blockchain.get_pending_txs()?;
             let chain = &self.consensus.forks[index as usize];
             for state_checkpoint in &chain.sequence {
                 for tx in &state_checkpoint.proposal.block.txs {
@@ -577,10 +630,18 @@ impl ValidatorState {
         // Validate state transition against canonical state
         // TODO: This should be validated against fork state
         info!(target: "consensus::validator", "receive_proposal(): Starting state transition validation");
-        if let Err(e) = self.verify_transactions(&proposal.block.txs, false).await {
-            error!(target: "consensus::validator", "receive_proposal(): Transaction verifications failed: {}", e);
-            return Err(e)
-        };
+        match self.verify_transactions(&proposal.block.txs, false).await {
+            Ok(erroneous_txs) => {
+                if !erroneous_txs.is_empty() {
+                    error!(target: "consensus::validator", "Proposal contains erroneous transactions");
+                    return Err(Error::ErroneousTxsDetected)
+                }
+            }
+            Err(e) => {
+                error!(target: "consensus::validator", "receive_proposal(): Transaction verifications failed: {}", e);
+                return Err(e)
+            }
+        }
 
         // TODO: [PLACEHOLDER] Add rewards validation
 
@@ -703,13 +764,21 @@ impl ValidatorState {
             // TODO: FIXME: The state transitions have already been written, they have to be in memory
             //              until this point.
             info!(target: "consensus::validator", "Applying state transition for finalized block");
-            if let Err(e) = self.verify_transactions(&proposal.txs, true).await {
-                error!(target: "consensus::validator", "Finalized block transaction verifications failed: {}", e);
-                return Err(e)
+            match self.verify_transactions(&proposal.txs, true).await {
+                Ok(erroneous_txs) => {
+                    if !erroneous_txs.is_empty() {
+                        error!(target: "consensus::validator", "Finalized block contains erroneous transactions");
+                        return Err(Error::ErroneousTxsDetected)
+                    }
+                }
+                Err(e) => {
+                    error!(target: "consensus::validator", "Finalized block transaction verifications failed: {}", e);
+                    return Err(e)
+                }
             }
 
             // Remove proposal transactions from pending txs store
-            if let Err(e) = self.blockchain.pending_txs.remove(&proposal.txs) {
+            if let Err(e) = self.blockchain.remove_pending_txs(&proposal.txs) {
                 error!(target: "consensus::validator", "Removing finalized block transactions failed: {}", e);
                 return Err(e)
             }
@@ -754,6 +823,11 @@ impl ValidatorState {
         self.consensus.forks = vec![];
         self.consensus.slot_checkpoints = vec![];
 
+        // Purge pending erroneous txs since canonical state has been changed
+        if let Err(e) = self.purge_pending_txs().await {
+            error!(target: "consensus::validator", "consensus: Purging pending transactions failed: {}", e);
+        }
+
         Ok((finalized, finalized_slot_checkpoints))
     }
 
@@ -770,15 +844,23 @@ impl ValidatorState {
     // ==========================
 
     /// Validate and append to canonical state received blocks.
-    pub async fn receive_blocks(&mut self, blocks: &[BlockInfo]) -> Result<()> {
+    async fn receive_blocks(&mut self, blocks: &[BlockInfo]) -> Result<()> {
         // Verify state transitions for all blocks and their respective transactions.
         info!(target: "consensus::validator", "receive_blocks(): Starting state transition validations");
 
         for block in blocks {
-            if let Err(e) = self.verify_transactions(&block.txs, true).await {
-                error!(target: "consensus::validator", "receive_blocks(): Transaction verifications failed: {}", e);
-                return Err(e)
-            };
+            match self.verify_transactions(&block.txs, true).await {
+                Ok(erroneous_txs) => {
+                    if !erroneous_txs.is_empty() {
+                        error!(target: "consensus::validator", "receive_blocks(): Block contains erroneous transactions");
+                        return Err(Error::ErroneousTxsDetected)
+                    }
+                }
+                Err(e) => {
+                    error!(target: "consensus::validator", "receive_blocks(): Transaction verifications failed: {}", e);
+                    return Err(e)
+                }
+            }
         }
 
         info!(target: "consensus::validator", "receive_blocks(): All state transitions passed. Appending blocks to ledger.");
@@ -818,7 +900,12 @@ impl ValidatorState {
         blocks_subscriber.notify(notif).await;
 
         info!(target: "consensus::validator", "receive_finalized_block(): Removing block transactions from pending txs store");
-        self.blockchain.pending_txs.remove(&block.txs)?;
+        self.blockchain.remove_pending_txs(&block.txs)?;
+
+        // Purge pending erroneous txs since canonical state has been changed
+        if let Err(e) = self.purge_pending_txs().await {
+            error!(target: "consensus::validator", "receive_finalized_block(): Purging pending transactions failed: {}", e);
+        }
 
         Ok(true)
     }
@@ -867,162 +954,182 @@ impl ValidatorState {
         Ok(())
     }
 
-    /// Validate signatures, wasm execution, and zk proofs for given transactions.
-    /// If all of those succeed, try to execute a state update for the contract calls.
-    /// Currently the verifications are sequential, and the function will skip a
-    /// transaction if any of the verifications fail.
-    /// The function takes a boolean called `write` which tells it to actually write
-    /// the state transitions to the database.
-    // TODO: This should be paralellized as if even one tx in the batch fails to verify,
-    //       we can skip it. When things are parallel, make sure to write in a deterministic
-    //       order.
-    // TODO: This function should be refactored to be more readable and efficient.
-    //       1. Get metadata
-    //       2. Verify signatures
-    //       3. Verify execution
-    //       4. Verify ZK proofs
-    //       5. (optionally) write
-    pub async fn verify_transactions(&self, txs: &[Transaction], write: bool) -> Result<()> {
-        info!(target: "consensus::validator", "Verifying {} transaction(s)", txs.len());
+    /// Validate signatures, wasm execution, and zk proofs for given transaction in
+    /// provided runtimes. If all of those succeed, try to execute a state update
+    /// for the contract calls.
+    async fn verify_transaction(
+        &self,
+        blockchain_overlay: BlockchainOverlayPtr,
+        tx: &Transaction,
+    ) -> Result<()> {
+        let mut runtimes = HashMap::new();
+        let tx_hash = blake3::hash(&serialize(tx));
+        info!(target: "consensus::validator", "Verifying transaction {}", tx_hash);
+
+        // Table of public inputs used for ZK proof verification
+        let mut zkp_table = vec![];
+        // Table of public keys used for signature verification
+        let mut sig_table = vec![];
+        // State updates produced by contract execution
+        let mut updates = vec![];
+        // Map of zk proof verifying keys for the current transaction
+        let mut verifying_keys: HashMap<[u8; 32], HashMap<String, VerifyingKey>> = HashMap::new();
+
+        // Initialize the map
+        for call in tx.calls.iter() {
+            verifying_keys.insert(call.contract_id.to_bytes(), HashMap::new());
+        }
 
-        for tx in txs {
-            let tx_hash = blake3::hash(&serialize(tx));
-            info!(target: "consensus::validator", "Verifying transaction {}", tx_hash);
-
-            // Table of public inputs used for ZK proof verification
-            let mut zkp_table = vec![];
-            // Table of public keys used for signature verification
-            let mut sig_table = vec![];
-            // State updates produced by contract execution
-            let mut updates = vec![];
-            // Map of zk proof verifying keys for the current transaction
-            //let mut verifying_keys: HashMap<[u8; 32], Vec<(String, VerifyingKey)>> = HashMap::new();
-            let mut verifying_keys: HashMap<[u8; 32], HashMap<String, VerifyingKey>> =
-                HashMap::new();
-
-            // Initialize the map
-            for call in tx.calls.iter() {
-                verifying_keys.insert(call.contract_id.to_bytes(), HashMap::new());
-            }
+        // Iterate over all calls to get the metadata
+        for (idx, call) in tx.calls.iter().enumerate() {
+            info!(target: "consensus::validator", "Executing contract call {}", idx);
+
+            // Write the actual payload data
+            let mut payload = vec![];
+            payload.write_u32(idx as u32)?; // Call index
+            tx.calls.encode(&mut payload)?; // Actual call data
 
-            // Iterate over all calls to get the metadata
-            for (idx, call) in tx.calls.iter().enumerate() {
-                info!(target: "consensus::validator", "Executing contract call {}", idx);
+            // Instantiate the wasm runtime
+            let runtime_key = call.contract_id.to_string();
+            if !runtimes.contains_key(&runtime_key) {
                 let wasm = self.blockchain.wasm_bincode.get(call.contract_id)?;
+                let r = Runtime::new(&wasm, blockchain_overlay.clone(), call.contract_id)?;
+                runtimes.insert(runtime_key.clone(), r);
+            }
+            let runtime = runtimes.get_mut(&runtime_key).unwrap();
 
-                // Write the actual payload data
-                let mut payload = vec![];
-                payload.write_u32(idx as u32)?; // Call index
-                tx.calls.encode(&mut payload)?; // Actual call data
+            info!(target: "consensus::validator", "Executing \"metadata\" call");
+            let metadata = runtime.metadata(&payload)?;
 
-                // Instantiate the wasm runtime
-                let mut runtime = Runtime::new(&wasm, self.blockchain.clone(), call.contract_id)?;
+            // Decode the metadata retrieved from the execution
+            let mut decoder = Cursor::new(&metadata);
 
-                info!(target: "consensus::validator", "Executing \"metadata\" call");
-                let metadata = runtime.metadata(&payload)?;
+            // The tuple is (zkas_ns, public_inputs)
+            let zkp_pub: Vec<(String, Vec<pallas::Base>)> = Decodable::decode(&mut decoder)?;
 
-                // Decode the metadata retrieved from the execution
-                let mut decoder = Cursor::new(&metadata);
+            let sig_pub: Vec<PublicKey> = Decodable::decode(&mut decoder)?;
+            // TODO: Make sure we've read all the bytes above.
+            info!(target: "consensus::validator", "Successfully executed \"metadata\" call");
 
-                // The tuple is (zkas_ns, public_inputs)
-                let zkp_pub: Vec<(String, Vec<pallas::Base>)> = Decodable::decode(&mut decoder)?;
+            // Here we'll look up verifying keys and insert them into the per-contract map.
+            info!(target: "consensus::validator", "Performing VerifyingKey lookups from the sled db");
+            for (zkas_ns, _) in &zkp_pub {
+                let inner_vk_map = verifying_keys.get_mut(&call.contract_id.to_bytes()).unwrap();
 
-                let sig_pub: Vec<PublicKey> = Decodable::decode(&mut decoder)?;
-                // TODO: Make sure we've read all the bytes above.
-                info!(target: "consensus::validator", "Successfully executed \"metadata\" call");
+                if inner_vk_map.contains_key(zkas_ns.as_str()) {
+                    continue
+                }
 
-                // Here we'll look up verifying keys and insert them into the per-contract map.
-                info!(target: "consensus::validator", "Performing VerifyingKey lookups from the sled db");
-                for (zkas_ns, _) in &zkp_pub {
-                    let inner_vk_map =
-                        verifying_keys.get_mut(&call.contract_id.to_bytes()).unwrap();
+                let (_, vk) = self.blockchain.contracts.get_zkas(
+                    &self.blockchain.sled_db,
+                    &call.contract_id,
+                    zkas_ns,
+                )?;
 
-                    if inner_vk_map.contains_key(zkas_ns.as_str()) {
-                        continue
-                    }
+                inner_vk_map.insert(zkas_ns.to_string(), vk);
+            }
 
-                    let (_, vk) = self.blockchain.contracts.get_zkas(
-                        &self.blockchain.sled_db,
-                        &call.contract_id,
-                        zkas_ns,
-                    )?;
+            zkp_table.push(zkp_pub);
+            sig_table.push(sig_pub);
 
-                    inner_vk_map.insert(zkas_ns.to_string(), vk);
-                }
+            // After getting the metadata, we run the "exec" function with the same
+            // runtime and the same payload.
+            info!(target: "consensus::validator", "Executing \"exec\" call");
+            let state_update = runtime.exec(&payload)?;
 
-                zkp_table.push(zkp_pub);
-                sig_table.push(sig_pub);
+            info!(target: "consensus::validator", "Successfully executed \"exec\" call");
+            updates.push(state_update);
 
-                // After getting the metadata, we run the "exec" function with the same
-                // runtime and the same payload.
-                info!(target: "consensus::validator", "Executing \"exec\" call");
-                let state_update = runtime.exec(&payload)?;
+            // At this point we're done with the call and move on to the next one.
+        }
 
-                info!(target: "consensus::validator", "Successfully executed \"exec\" call");
-                updates.push(state_update);
+        // When we're done looping and executing over the tx's contract calls, we
+        // move on with verification. First we verify the signatures as that's
+        // cheaper, and then finally we verify the ZK proofs.
+        info!(target: "consensus::validator", "Verifying signatures for transaction {}", tx_hash);
+        if sig_table.len() != tx.signatures.len() {
+            error!(target: "consensus::validator", "Incorrect number of signatures in tx {}", tx_hash);
+            return Err(Error::InvalidSignature)
+        }
 
-                // At this point we're done with the call and move on to the next one.
+        match tx.verify_sigs(sig_table) {
+            Ok(()) => {
+                info!(target: "consensus::validator", "Signatures verification for tx {} successful", tx_hash)
+            }
+            Err(e) => {
+                error!(target: "consensus::validator", "Signature verification for tx {} failed: {}", tx_hash, e);
+                return Err(e)
             }
+        };
 
-            // When we're done looping and executing over the tx's contract calls, we
-            // move on with verification. First we verify the signatures as that's
-            // cheaper, and then finally we verify the ZK proofs.
-            info!(target: "consensus::validator", "Verifying signatures for transaction {}", tx_hash);
-            if sig_table.len() != tx.signatures.len() {
-                error!(target: "consensus::validator", "Incorrect number of signatures in tx {}", tx_hash);
-                return Err(Error::InvalidSignature)
+        info!(target: "consensus::validator", "Verifying ZK proofs for transaction {}", tx_hash);
+        match tx.verify_zkps(verifying_keys.clone(), zkp_table).await {
+            Ok(()) => {
+                info!(target: "consensus::validator", "ZK proof verification for tx {} successful", tx_hash)
+            }
+            Err(e) => {
+                error!(target: "consensus::validator", "ZK proof verification for tx {} failed: {}", tx_hash, e);
+                return Err(e)
             }
+        };
 
-            match tx.verify_sigs(sig_table) {
-                Ok(()) => {
-                    info!(target: "consensus::validator", "Signatures verification for tx {} successful", tx_hash)
-                }
-                Err(e) => {
-                    error!(target: "consensus::validator", "Signature verification for tx {} failed: {}", tx_hash, e);
-                    return Err(e)
-                }
-            };
+        // After the verifications stage passes we can apply the state updates.
+        assert!(tx.calls.len() == updates.len());
+
+        info!(target: "consensus::validator", "Performing state updates");
+        for (call, update) in tx.calls.iter().zip(updates.iter()) {
+            // Retrieve already initiated runtime and apply update
+            // TODO: Sum up the gas costs of previous calls during execution
+            //       and verification and these.
+            let runtime = runtimes.get_mut(&call.contract_id.to_string()).unwrap();
+            info!(target: "consensus::validator", "Executing \"apply\" call");
+            runtime.apply(update)?;
+            info!(target: "consensus::validator", "State update applied successfully")
+        }
 
-            info!(target: "consensus::validator", "Verifying ZK proofs for transaction {}", tx_hash);
-            match tx.verify_zkps(verifying_keys.clone(), zkp_table).await {
-                Ok(()) => {
-                    info!(target: "consensus::validator", "ZK proof verification for tx {} successful", tx_hash)
-                }
-                Err(e) => {
-                    error!(target: "consensus::validator", "ZK proof verification for tx {} failed: {}", tx_hash, e);
-                    return Err(e)
-                }
-            };
+        info!(target: "consensus::validator", "Transaction {} verified successfully", tx_hash);
 
-            // After the verifications stage passes, if we're told to write, we
-            // apply the state updates.
-            assert!(tx.calls.len() == updates.len());
-
-            if write {
-                info!(target: "consensus::validator", "Performing state updates");
-                for (call, update) in tx.calls.iter().zip(updates.iter()) {
-                    // For this we instantiate the runtimes again.
-                    // TODO: Optimize this
-                    // TODO: Sum up the gas costs of previous calls during execution
-                    //       and verification and these.
-                    let wasm = self.blockchain.wasm_bincode.get(call.contract_id)?;
-
-                    let mut runtime =
-                        Runtime::new(&wasm, self.blockchain.clone(), call.contract_id)?;
-
-                    info!(target: "consensus::validator", "Executing \"apply\" call");
-                    // TODO: FIXME: This should be done in an atomic tx/batch
-                    runtime.apply(update)?;
-                    info!(target: "consensus::validator", "State update applied successfully")
-                }
-            } else {
-                info!(target: "consensus::validator", "Skipping apply of state updates because write=false");
+        Ok(())
+    }
+
+    /// Validate a set of [`Transaction`] in sequence and apply them if all are valid.
+    /// Erroneous transactions are filtered out of the set and returned to caller.
+    /// The function takes a boolean called `write` which tells it to actually write
+    /// the state transitions to the database.
+    pub async fn verify_transactions(
+        &self,
+        txs: &[Transaction],
+        write: bool,
+    ) -> Result<Vec<Transaction>> {
+        info!(target: "consensus::validator", "Verifying {} transaction(s)", txs.len());
+
+        let mut erroneous_txs = vec![];
+        let blockchain_overlay = BlockchainOverlay::new(&self.blockchain)?;
+
+        for tx in txs {
+            if let Err(e) = self.verify_transaction(blockchain_overlay.clone(), tx).await {
+                warn!(target: "consensus::validator", "Transaction verification failed: {}", e);
+                erroneous_txs.push(tx.clone());
             }
+        }
 
-            info!(target: "consensus::validator", "Transaction {} verified successfully", tx_hash);
+        let lock = blockchain_overlay.lock().unwrap();
+        let overlay = lock.overlay.lock().unwrap();
+        if !erroneous_txs.is_empty() {
+            warn!(target: "consensus::validator", "Erroneous transactions found in set");
+            overlay.purge_new_trees()?;
+            return Ok(erroneous_txs)
         }
 
-        Ok(())
+        if !write {
+            info!(target: "consensus::validator", "Skipping apply of state updates because write=false");
+            overlay.purge_new_trees()?;
+            return Ok(erroneous_txs)
+        }
+
+        overlay.apply()?;
+
+        Ok(erroneous_txs)
     }
 
     /// Append to canonical state received finalized slot checkpoints from block sync task.

src/contract/money/Makefile

@@ -38,14 +38,19 @@ test-mint-pay-swap: all
 		--package darkfi-money-contract \
 		--test mint_pay_swap
 
+test-txs-verification: all
+	$(CARGO) test --release --features=no-entrypoint,client \
+		--package darkfi-money-contract \
+		--test txs_verification
+
 bench:
 	$(CARGO) test --release --features=no-entrypoint,client \
 		--package darkfi-money-contract \
 		--test verification_bench $(FILTER)
 
-test: test-integration test-mint-pay-swap
+test: test-integration test-mint-pay-swap test-txs-verification
 
 clean:
 	rm -f $(PROOFS_BIN) $(WASM_BIN)
 
-.PHONY: all test-integration test-mint-pay-swap bench test clean
+.PHONY: all test-integration test-mint-pay-swap test-txs-verification bench test clean

src/contract/money/tests/harness.rs

@@ -15,6 +15,7 @@
  * You should have received a copy of the GNU Affero General Public License
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
+
 use std::collections::HashMap;
 
 use darkfi::{
@@ -29,19 +30,21 @@ use darkfi::{
     zkas::ZkBinary,
     Result,
 };
-use darkfi_money_contract::client::OwnCoin;
 use darkfi_sdk::{
     crypto::{Keypair, MerkleTree, PublicKey, DARK_TOKEN_ID, MONEY_CONTRACT_ID},
     pasta::pallas,
     ContractCall,
 };
 use darkfi_serial::{deserialize, serialize, Encodable};
-use log::warn;
+use log::info;
 use rand::rngs::OsRng;
 
 use darkfi_money_contract::{
-    client::{mint_v1::MintCallBuilder, transfer_v1::TransferCallBuilder},
-    model::{MoneyMintParamsV1, MoneyTransferParamsV1},
+    client::{
+        freeze_v1::FreezeCallBuilder, mint_v1::MintCallBuilder, transfer_v1::TransferCallBuilder,
+        OwnCoin,
+    },
+    model::{MoneyFreezeParamsV1, MoneyMintParamsV1, MoneyTransferParamsV1},
     MoneyFunction, MONEY_CONTRACT_ZKAS_BURN_NS_V1, MONEY_CONTRACT_ZKAS_MINT_NS_V1,
     MONEY_CONTRACT_ZKAS_TOKEN_FRZ_NS_V1, MONEY_CONTRACT_ZKAS_TOKEN_MINT_NS_V1,
 };
@@ -50,17 +53,18 @@ pub fn init_logger() {
     let mut cfg = simplelog::ConfigBuilder::new();
     cfg.add_filter_ignore("sled".to_string());
     cfg.add_filter_ignore("blockchain::contractstore".to_string());
+
     // We check this error so we can execute same file tests in parallel,
     // otherwise second one fails to init logger here.
     if let Err(_) = simplelog::TermLogger::init(
-        //simplelog::LevelFilter::Info,
-        simplelog::LevelFilter::Debug,
+        simplelog::LevelFilter::Info,
+        //simplelog::LevelFilter::Debug,
         //simplelog::LevelFilter::Trace,
         cfg.build(),
         simplelog::TerminalMode::Mixed,
         simplelog::ColorChoice::Auto,
     ) {
-        warn!(target: "money_harness", "Logger already initialized");
+        info!(target: "money_harness", "Logger already initialized");
     }
 }
 
@@ -224,4 +228,29 @@ impl MoneyTestHarness {
 
         Ok((tx, debris.params))
     }
+
+    pub fn freeze_token(
+        &self,
+        mint_authority: Keypair,
+    ) -> Result<(Transaction, MoneyFreezeParamsV1)> {
+        let (token_freeze_pk, token_freeze_zkbin) =
+            self.proving_keys.get(&MONEY_CONTRACT_ZKAS_TOKEN_FRZ_NS_V1).unwrap();
+
+        let builder = FreezeCallBuilder {
+            mint_authority,
+            token_freeze_zkbin: token_freeze_zkbin.clone(),
+            token_freeze_pk: token_freeze_pk.clone(),
+        };
+        let debris = builder.build()?;
+
+        let mut data = vec![MoneyFunction::FreezeV1 as u8];
+        debris.params.encode(&mut data)?;
+        let calls = vec![ContractCall { contract_id: *MONEY_CONTRACT_ID, data }];
+        let proofs = vec![debris.proofs];
+        let mut tx = Transaction { calls, proofs, signatures: vec![] };
+        let sigs = tx.create_sigs(&mut OsRng, &[mint_authority.secret])?;
+        tx.signatures = vec![sigs];
+
+        Ok((tx, debris.params))
+    }
 }

src/contract/money/tests/integration.rs

@@ -29,20 +29,15 @@
 //!
 //! TODO: Malicious cases
 
-use darkfi::{tx::Transaction, Result};
+use darkfi::Result;
 use darkfi_sdk::{
-    crypto::{poseidon_hash, Keypair, MerkleNode, Nullifier, MONEY_CONTRACT_ID},
+    crypto::{poseidon_hash, Keypair, MerkleNode, Nullifier},
     incrementalmerkletree::Tree,
-    ContractCall,
 };
-use darkfi_serial::Encodable;
 use log::info;
 use rand::rngs::OsRng;
 
-use darkfi_money_contract::{
-    client::{freeze_v1::FreezeCallBuilder, MoneyNote, OwnCoin},
-    MoneyFunction, MONEY_CONTRACT_ZKAS_TOKEN_FRZ_NS_V1,
-};
+use darkfi_money_contract::client::{MoneyNote, OwnCoin};
 
 mod harness;
 use harness::{init_logger, MoneyTestHarness};
@@ -132,23 +127,7 @@ async fn money_integration() -> Result<()> {
 
     // Let's attempt to freeze the BOBTOKEN mint,
     // and after that we shouldn't be able to mint anymore.
-    let (token_freeze_pk, token_freeze_zkbin) =
-        th.proving_keys.get(&MONEY_CONTRACT_ZKAS_TOKEN_FRZ_NS_V1).unwrap();
-
-    let bob_frz_builder = FreezeCallBuilder {
-        mint_authority: bob_token_authority,
-        token_freeze_zkbin: token_freeze_zkbin.clone(),
-        token_freeze_pk: token_freeze_pk.clone(),
-    };
-
-    let debris = bob_frz_builder.build()?;
-    let mut data = vec![MoneyFunction::FreezeV1 as u8];
-    debris.params.encode(&mut data)?;
-    let calls = vec![ContractCall { contract_id: *MONEY_CONTRACT_ID, data }];
-    let proofs = vec![debris.proofs];
-    let mut bob_frz_tx = Transaction { calls, proofs, signatures: vec![] };
-    let sigs = bob_frz_tx.create_sigs(&mut OsRng, &[bob_token_authority.secret])?;
-    bob_frz_tx.signatures = vec![sigs];
+    let (bob_frz_tx, _) = th.freeze_token(bob_token_authority)?;
 
     info!("[Faucet] Executing BOBTOKEN freeze");
     th.faucet.state.read().await.verify_transactions(&[bob_frz_tx.clone()], true).await?;

src/contract/money/tests/txs_verification.rs

@@ -0,0 +1,257 @@
+/* This file is part of DarkFi (https://dark.fi)
+ *
+ * Copyright (C) 2020-2023 Dyne.org foundation
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+//! Test for transaction verification correctness between Alice and Bob.
+//!
+//! We first mint Alice some tokens, and then she send some to Bob
+//! a couple of times, including some double spending transactions.
+//!
+//! With this test, we want to confirm the transactions execution works
+//! between multiple parties, with detection of erroneous transactions.
+
+use darkfi::{tx::Transaction, Result};
+use darkfi_sdk::{
+    crypto::{
+        merkle_prelude::*, pallas, pasta_prelude::*, poseidon_hash, Coin, MerkleNode, Nullifier,
+        MONEY_CONTRACT_ID,
+    },
+    ContractCall,
+};
+use darkfi_serial::Encodable;
+use log::info;
+use rand::rngs::OsRng;
+
+use darkfi_money_contract::{
+    client::{transfer_v1::TransferCallBuilder, MoneyNote, OwnCoin},
+    MoneyFunction::TransferV1 as MoneyTransfer,
+    MONEY_CONTRACT_ZKAS_BURN_NS_V1, MONEY_CONTRACT_ZKAS_MINT_NS_V1,
+};
+
+mod harness;
+use harness::{init_logger, MoneyTestHarness};
+
+#[async_std::test]
+async fn txs_verification() -> Result<()> {
+    init_logger();
+
+    // Some numbers we want to assert
+    const ALICE_INITIAL: u64 = 100;
+
+    // Alice = 50 ALICE
+    // Bob = 50 ALICE
+    const ALICE_FIRST_SEND: u64 = ALICE_INITIAL - 50;
+
+    // Initialize harness
+    let mut th = MoneyTestHarness::new().await?;
+    let (mint_pk, mint_zkbin) = th.proving_keys.get(&MONEY_CONTRACT_ZKAS_MINT_NS_V1).unwrap();
+    let (burn_pk, burn_zkbin) = th.proving_keys.get(&MONEY_CONTRACT_ZKAS_BURN_NS_V1).unwrap();
+    let contract_id = *MONEY_CONTRACT_ID;
+
+    // We're just going to be using a zero spend-hook and user-data
+    let rcpt_spend_hook = pallas::Base::zero();
+    let rcpt_user_data = pallas::Base::zero();
+    let rcpt_user_data_blind = pallas::Base::random(&mut OsRng);
+    let change_spend_hook = pallas::Base::zero();
+    let change_user_data = pallas::Base::zero();
+    let change_user_data_blind = pallas::Base::random(&mut OsRng);
+
+    let mut alice_owncoins = vec![];
+    let mut bob_owncoins = vec![];
+
+    info!(target: "money", "[Alice] ================================");
+    info!(target: "money", "[Alice] Building token mint tx for Alice");
+    info!(target: "money", "[Alice] ================================");
+    let (alice_mint_tx, alice_params) =
+        th.mint_token(th.alice.keypair, ALICE_INITIAL, th.alice.keypair.public)?;
+
+    info!(target: "money", "[Faucet] =============================");
+    info!(target: "money", "[Faucet] Executing Alice token mint tx");
+    info!(target: "money", "[Faucet] =============================");
+    th.faucet.state.read().await.verify_transactions(&[alice_mint_tx.clone()], true).await?;
+    th.faucet.merkle_tree.append(&MerkleNode::from(alice_params.output.coin.inner()));
+
+    info!(target: "money", "[Alice] =============================");
+    info!(target: "money", "[Alice] Executing Alice token mint tx");
+    info!(target: "money", "[Alice] =============================");
+    th.alice.state.read().await.verify_transactions(&[alice_mint_tx.clone()], true).await?;
+    th.alice.merkle_tree.append(&MerkleNode::from(alice_params.output.coin.inner()));
+    // Alice has to witness this coin because it's hers.
+    let alice_leaf_pos = th.alice.merkle_tree.witness().unwrap();
+
+    info!(target: "money", "[Bob] =============================");
+    info!(target: "money", "[Bob] Executing Alice token mint tx");
+    info!(target: "money", "[Bob] =============================");
+    th.bob.state.read().await.verify_transactions(&[alice_mint_tx.clone()], true).await?;
+    th.bob.merkle_tree.append(&MerkleNode::from(alice_params.output.coin.inner()));
+
+    assert!(th.alice.merkle_tree.root(0).unwrap() == th.bob.merkle_tree.root(0).unwrap());
+    assert!(th.faucet.merkle_tree.root(0).unwrap() == th.bob.merkle_tree.root(0).unwrap());
+
+    // Alice builds an `OwnCoin` from her airdrop
+    let note: MoneyNote = alice_params.output.note.decrypt(&th.alice.keypair.secret)?;
+    let alice_token_id = note.token_id;
+    let alice_oc = OwnCoin {
+        coin: Coin::from(alice_params.output.coin),
+        note: note.clone(),
+        secret: th.alice.keypair.secret, // <-- What should this be?
+        nullifier: Nullifier::from(poseidon_hash([th.alice.keypair.secret.inner(), note.serial])),
+        leaf_position: alice_leaf_pos,
+    };
+    alice_owncoins.push(alice_oc);
+
+    // Now Alice can send a little bit of funds to Bob.
+    // We can duplicate this transaction to simulate double spending.
+    let duplicates = 3; // Change this number to >1 to double spend
+    let mut transactions = vec![];
+    let mut txs_params = vec![];
+    for i in 0..duplicates {
+        info!(target: "money", "[Alice] ======================================================");
+        info!(target: "money", "[Alice] Building Money::Transfer params for payment {i} to Bob");
+        info!(target: "money", "[Alice] ======================================================");
+        let alice2bob_call_debris = TransferCallBuilder {
+            keypair: th.alice.keypair,
+            recipient: th.bob.keypair.public,
+            value: ALICE_FIRST_SEND,
+            token_id: alice_token_id,
+            rcpt_spend_hook,
+            rcpt_user_data,
+            rcpt_user_data_blind,
+            change_spend_hook,
+            change_user_data,
+            change_user_data_blind,
+            coins: alice_owncoins.clone(),
+            tree: th.alice.merkle_tree.clone(),
+            mint_zkbin: mint_zkbin.clone(),
+            mint_pk: mint_pk.clone(),
+            burn_zkbin: burn_zkbin.clone(),
+            burn_pk: burn_pk.clone(),
+            clear_input: false,
+        }
+        .build()?;
+        let (alice2bob_params, alice2bob_proofs, alice2bob_secret_keys, alice2bob_spent_coins) = (
+            alice2bob_call_debris.params,
+            alice2bob_call_debris.proofs,
+            alice2bob_call_debris.signature_secrets,
+            alice2bob_call_debris.spent_coins,
+        );
+
+        assert!(alice2bob_params.inputs.len() == 1);
+        assert!(alice2bob_params.outputs.len() == 2);
+        assert!(alice2bob_spent_coins.len() == 1);
+
+        info!(target: "money", "[Alice] ==============================");
+        info!(target: "money", "[Alice] Building payment tx {i} to Bob");
+        info!(target: "money", "[Alice] ==============================");
+        let mut data = vec![MoneyTransfer as u8];
+        alice2bob_params.encode(&mut data)?;
+        let calls = vec![ContractCall { contract_id, data }];
+        let proofs = vec![alice2bob_proofs];
+        let mut alice2bob_tx = Transaction { calls, proofs, signatures: vec![] };
+        let sigs = alice2bob_tx.create_sigs(&mut OsRng, &alice2bob_secret_keys)?;
+        alice2bob_tx.signatures = vec![sigs];
+
+        // Now we simulate nodes verification, as transactions come one by one.
+        // Validation should pass, even when we are trying to double spent.
+        info!(target: "money", "[Faucet] ==================================");
+        info!(target: "money", "[Faucet] Verifying Alice2Bob payment tx {i}");
+        info!(target: "money", "[Faucet] ==================================");
+        th.faucet.state.read().await.verify_transactions(&[alice2bob_tx.clone()], false).await?;
+
+        info!(target: "money", "[Alice] ==================================");
+        info!(target: "money", "[Alice] Verifying Alice2Bob payment tx {i}");
+        info!(target: "money", "[Alice] ==================================");
+        th.alice.state.read().await.verify_transactions(&[alice2bob_tx.clone()], false).await?;
+
+        info!(target: "money", "[Bob] ==================================");
+        info!(target: "money", "[Bob] Verifying Alice2Bob payment tx {i}");
+        info!(target: "money", "[Bob] ==================================");
+        th.bob.state.read().await.verify_transactions(&[alice2bob_tx.clone()], false).await?;
+
+        transactions.push(alice2bob_tx);
+        txs_params.push(alice2bob_params);
+    }
+    alice_owncoins = vec![];
+    assert_eq!(transactions.len(), duplicates);
+    assert_eq!(txs_params.len(), duplicates);
+
+    // Now we can try to execute the transactions sequentialy.
+    // Each node will detect the duplicate txs and filter them out,
+    // then only apply the first txs from the set.
+    let valid_txs = vec![transactions[0].clone()];
+    info!(target: "money", "[Faucet] ==============================");
+    info!(target: "money", "[Faucet] Executing Alice2Bob payment tx");
+    info!(target: "money", "[Faucet] ==============================");
+    let erroneous_txs =
+        th.faucet.state.read().await.verify_transactions(&transactions, true).await?;
+    assert_eq!(erroneous_txs.len(), duplicates - 1);
+    th.faucet.state.read().await.verify_transactions(&valid_txs, true).await?;
+    th.faucet.merkle_tree.append(&MerkleNode::from(txs_params[0].outputs[0].coin.inner()));
+    th.faucet.merkle_tree.append(&MerkleNode::from(txs_params[0].outputs[1].coin.inner()));
+
+    info!(target: "money", "[Alice] ==============================");
+    info!(target: "money", "[Alice] Executing Alice2Bob payment tx");
+    info!(target: "money", "[Alice] ==============================");
+    let erroneous_txs =
+        th.alice.state.read().await.verify_transactions(&transactions, true).await?;
+    assert_eq!(erroneous_txs.len(), duplicates - 1);
+    th.alice.state.read().await.verify_transactions(&valid_txs, true).await?;
+    th.alice.merkle_tree.append(&MerkleNode::from(txs_params[0].outputs[0].coin.inner()));
+    let alice_leaf_pos = th.alice.merkle_tree.witness().unwrap();
+    th.alice.merkle_tree.append(&MerkleNode::from(txs_params[0].outputs[1].coin.inner()));
+
+    info!(target: "money", "[Bob] ==============================");
+    info!(target: "money", "[Bob] Executing Alice2Bob payment tx");
+    info!(target: "money", "[Bob] ==============================");
+    let erroneous_txs = th.bob.state.read().await.verify_transactions(&transactions, true).await?;
+    assert_eq!(erroneous_txs.len(), duplicates - 1);
+    th.bob.state.read().await.verify_transactions(&valid_txs, true).await?;
+    th.bob.merkle_tree.append(&MerkleNode::from(txs_params[0].outputs[0].coin.inner()));
+    th.bob.merkle_tree.append(&MerkleNode::from(txs_params[0].outputs[1].coin.inner()));
+    let bob_leaf_pos = th.bob.merkle_tree.witness().unwrap();
+
+    assert!(th.alice.merkle_tree.root(0).unwrap() == th.bob.merkle_tree.root(0).unwrap());
+    assert!(th.faucet.merkle_tree.root(0).unwrap() == th.bob.merkle_tree.root(0).unwrap());
+
+    // Alice should now have one OwnCoin with the change from the above transaction.
+    let note: MoneyNote = txs_params[0].outputs[0].note.decrypt(&th.alice.keypair.secret)?;
+    let alice_oc = OwnCoin {
+        coin: Coin::from(txs_params[0].outputs[0].coin),
+        note: note.clone(),
+        secret: th.alice.keypair.secret, // <-- What should this be?
+        nullifier: Nullifier::from(poseidon_hash([th.alice.keypair.secret.inner(), note.serial])),
+        leaf_position: alice_leaf_pos,
+    };
+    alice_owncoins.push(alice_oc);
+
+    // Bob should now have this new one.
+    let note: MoneyNote = txs_params[0].outputs[1].note.decrypt(&th.bob.keypair.secret)?;
+    let bob_oc = OwnCoin {
+        coin: Coin::from(txs_params[0].outputs[1].coin),
+        note: note.clone(),
+        secret: th.bob.keypair.secret, // <-- What should this be?
+        nullifier: Nullifier::from(poseidon_hash([th.bob.keypair.secret.inner(), note.serial])),
+        leaf_position: bob_leaf_pos,
+    };
+    bob_owncoins.push(bob_oc);
+
+    assert!(alice_owncoins.len() == 1);
+    assert!(bob_owncoins.len() == 1);
+
+    // Thanks for reading
+    Ok(())
+}

src/error.rs

@@ -272,6 +272,9 @@ pub enum Error {
     #[error("Proposer is not eligible to produce proposals")]
     ProposalProposerNotEligible,
 
+    #[error("Erroneous transactions detected")]
+    ErroneousTxsDetected,
+
     // ===============
     // Database errors
     // ===============

src/runtime/import/db.rs

@@ -33,44 +33,18 @@ use crate::{
     runtime::vm_runtime::{ContractSection, Env, SMART_CONTRACT_ZKAS_DB_NAME},
     zk::{empty_witnesses, VerifyingKey, ZkCircuit},
     zkas::ZkBinary,
-    Result,
 };
 
 /// Internal wasm runtime API for sled trees
 pub struct DbHandle {
     pub contract_id: ContractId,
-    tree: sled::Tree,
+    pub tree: [u8; 32],
 }
 
 impl DbHandle {
-    pub fn new(contract_id: ContractId, tree: sled::Tree) -> Self {
+    pub fn new(contract_id: ContractId, tree: [u8; 32]) -> Self {
         Self { contract_id, tree }
     }
-
-    pub fn tree(&self) -> sled::Tree {
-        self.tree.clone()
-    }
-
-    pub fn get(&self, key: &[u8]) -> Result<Option<Vec<u8>>> {
-        if let Some(v) = self.tree.get(key)? {
-            return Ok(Some(v.to_vec()))
-        };
-
-        Ok(None)
-    }
-
-    pub fn contains_key(&self, key: &[u8]) -> Result<bool> {
-        Ok(self.tree.contains_key(key)?)
-    }
-
-    pub fn apply_batch(&self, batch: sled::Batch) -> Result<()> {
-        Ok(self.tree.apply_batch(batch)?)
-    }
-
-    pub fn flush(&self) -> Result<()> {
-        let _ = self.tree.flush()?;
-        Ok(())
-    }
 }
 
 /// Only deploy() can call this. Creates a new database instance for this contract.
@@ -84,8 +58,7 @@ pub(crate) fn db_init(ctx: FunctionEnvMut<Env>, ptr: WasmPtr<u8>, len: u32) -> i
     }
 
     let memory_view = env.memory_view(&ctx);
-    let db = &env.blockchain.sled_db;
-    let contracts = &env.blockchain.contracts;
+    let contracts = &env.blockchain.lock().unwrap().contracts;
     let contract_id = &env.contract_id;
 
     let Ok(mem_slice) = ptr.slice(&memory_view, len) else {
@@ -134,7 +107,7 @@ pub(crate) fn db_init(ctx: FunctionEnvMut<Env>, ptr: WasmPtr<u8>, len: u32) -> i
         return CALLER_ACCESS_DENIED
     }
 
-    let tree_handle = match contracts.init(db, &cid, &db_name) {
+    let tree_handle = match contracts.init(&cid, &db_name) {
         Ok(v) => v,
         Err(e) => {
             error!(target: "runtime::db::db_init()", "Failed to init db: {}", e);
@@ -144,14 +117,8 @@ pub(crate) fn db_init(ctx: FunctionEnvMut<Env>, ptr: WasmPtr<u8>, len: u32) -> i
 
     // TODO: Make sure we don't duplicate the DbHandle in the vec.
     //       It should behave like an ordered set.
-    // In `lookup()` we also create a `sled::Batch`. This is done for
-    // some simplicity reasons, and also for possible future changes.
-    // However, we make sure that unauthorized writes are not available
-    // from other functions that interface with the databases.
     let mut db_handles = env.db_handles.borrow_mut();
-    let mut db_batches = env.db_batches.borrow_mut();
     db_handles.push(DbHandle::new(cid, tree_handle));
-    db_batches.push(sled::Batch::default());
     (db_handles.len() - 1) as i32
 }
 
@@ -174,8 +141,7 @@ pub(crate) fn db_lookup(ctx: FunctionEnvMut<Env>, ptr: WasmPtr<u8>, len: u32) ->
     }
 
     let memory_view = env.memory_view(&ctx);
-    let db = &env.blockchain.sled_db;
-    let contracts = &env.blockchain.contracts;
+    let contracts = &env.blockchain.lock().unwrap().contracts;
 
     let Ok(mem_slice) = ptr.slice(&memory_view, len) else {
         error!(target: "runtime::db::db_lookup()", "Failed to make slice from ptr");
@@ -218,24 +184,15 @@ pub(crate) fn db_lookup(ctx: FunctionEnvMut<Env>, ptr: WasmPtr<u8>, len: u32) ->
         return DB_LOOKUP_FAILED
     }*/
 
-    let tree_handle = match contracts.lookup(db, &cid, &db_name) {
+    let tree_handle = match contracts.lookup(&cid, &db_name) {
         Ok(v) => v,
-        Err(e) => {
-            error!(target: "runtime::db::db_lookup()", "Failed to lookup db: {}", e);
-            return DB_LOOKUP_FAILED
-        }
+        Err(_) => return DB_LOOKUP_FAILED,
     };
 
     // TODO: Make sure we don't duplicate the DbHandle in the vec.
     //       It should behave like an ordered set.
-    // In `lookup()` we also create a `sled::Batch`. This is done for
-    // some simplicity reasons, and also for possible future changes.
-    // However, we make sure that unauthorized writes are not available
-    // from other functions that interface with the databases.
     let mut db_handles = env.db_handles.borrow_mut();
-    let mut db_batches = env.db_batches.borrow_mut();
     db_handles.push(DbHandle::new(cid, tree_handle));
-    db_batches.push(sled::Batch::default());
     (db_handles.len() - 1) as i32
 }
 
@@ -299,23 +256,33 @@ pub(crate) fn db_set(ctx: FunctionEnvMut<Env>, ptr: WasmPtr<u8>, len: u32) -> i3
     }*/
 
     let db_handles = env.db_handles.borrow();
-    let mut db_batches = env.db_batches.borrow_mut();
 
-    if db_handles.len() <= db_handle || db_batches.len() <= db_handle {
+    if db_handles.len() <= db_handle {
         error!(target: "runtime::db::db_set()", "Requested DbHandle that is out of bounds");
         return DB_SET_FAILED
     }
 
     let handle_idx = db_handle;
     let db_handle = &db_handles[handle_idx];
-    let db_batch = &mut db_batches[handle_idx];
 
     if db_handle.contract_id != env.contract_id {
         error!(target: "runtime::db::db_set()", "Unauthorized to write to DbHandle");
         return CALLER_ACCESS_DENIED
     }
 
-    db_batch.insert(key, value);
+    if env
+        .blockchain
+        .lock()
+        .unwrap()
+        .overlay
+        .lock()
+        .unwrap()
+        .insert(&db_handle.tree, &key, &value)
+        .is_err()
+    {
+        error!(target: "runtime::db::db_set()", "Couldn't insert to db_handle tree");
+        return DB_SET_FAILED
+    }
 
     DB_SUCCESS
 }
@@ -372,23 +339,25 @@ pub(crate) fn db_del(ctx: FunctionEnvMut<Env>, ptr: WasmPtr<u8>, len: u32) -> i3
     }*/
 
     let db_handles = env.db_handles.borrow();
-    let mut db_batches = env.db_batches.borrow_mut();
 
-    if db_handles.len() <= db_handle || db_batches.len() <= db_handle {
+    if db_handles.len() <= db_handle {
         error!(target: "runtime::db::db_del()", "Requested DbHandle that is out of bounds");
         return DB_DEL_FAILED
     }
 
     let handle_idx = db_handle;
     let db_handle = &db_handles[handle_idx];
-    let db_batch = &mut db_batches[handle_idx];
 
     if db_handle.contract_id != env.contract_id {
         error!(target: "runtime::db::db_del()", "Unauthorized to write to DbHandle");
         return CALLER_ACCESS_DENIED
     }
 
-    db_batch.remove(key);
+    if env.blockchain.lock().unwrap().overlay.lock().unwrap().remove(&db_handle.tree, &key).is_err()
+    {
+        error!(target: "runtime::db::db_del()", "Couldn't remove key from db_handle tree");
+        return DB_DEL_FAILED
+    }
 
     DB_SUCCESS
 }
@@ -455,13 +424,14 @@ pub(crate) fn db_get(ctx: FunctionEnvMut<Env>, ptr: WasmPtr<u8>, len: u32) -> i6
     let handle_idx = db_handle;
     let db_handle = &db_handles[handle_idx];
 
-    let ret = match db_handle.get(&key) {
-        Ok(v) => v,
-        Err(e) => {
-            error!(target: "runtime::db::db_get()", "Internal error getting from tree: {}", e);
-            return DB_GET_FAILED.into()
-        }
-    };
+    let ret =
+        match env.blockchain.lock().unwrap().overlay.lock().unwrap().get(&db_handle.tree, &key) {
+            Ok(v) => v,
+            Err(e) => {
+                error!(target: "runtime::db::db_get()", "Internal error getting from tree: {}", e);
+                return DB_GET_FAILED.into()
+            }
+        };
 
     let Some(return_data) = ret else {
         debug!(target: "runtime::db::db_get()", "returned empty vec");
@@ -470,7 +440,7 @@ pub(crate) fn db_get(ctx: FunctionEnvMut<Env>, ptr: WasmPtr<u8>, len: u32) -> i6
 
     // Copy Vec<u8> to the VM
     let mut objects = env.objects.borrow_mut();
-    objects.push(return_data);
+    objects.push(return_data.to_vec());
     (objects.len() - 1) as i64
 }
 
@@ -537,7 +507,8 @@ pub(crate) fn db_contains_key(ctx: FunctionEnvMut<Env>, ptr: WasmPtr<u8>, len: u
     let handle_idx = db_handle;
     let db_handle = &db_handles[handle_idx];
 
-    match db_handle.contains_key(&key) {
+    match env.blockchain.lock().unwrap().overlay.lock().unwrap().contains_key(&db_handle.tree, &key)
+    {
         Ok(v) => i32::from(v), // <- 0=false, 1=true
         Err(e) => {
             error!(target: "runtime::db::db_contains_key()", "sled.tree.contains_key failed: {}", e);
@@ -588,9 +559,7 @@ pub(crate) fn zkas_db_set(ctx: FunctionEnvMut<Env>, ptr: WasmPtr<u8>, len: u32)
 
     // Because of `Runtime::Deploy`, we should be sure that the zkas db is index zero.
     let db_handles = env.db_handles.borrow();
-    let mut db_batches = env.db_batches.borrow_mut();
     let db_handle = &db_handles[0];
-    let db_batch = &mut db_batches[0];
     // Redundant check
     if &db_handle.contract_id != contract_id {
         error!(target: "runtime::db::zkas_db_set()", "Internal error, zkas db at index 0 incorrect");
@@ -600,7 +569,15 @@ pub(crate) fn zkas_db_set(ctx: FunctionEnvMut<Env>, ptr: WasmPtr<u8>, len: u32)
     // Check if there is existing bincode and compare it. Return DB_SUCCESS if
     // they're the same. The assumption should be that VerifyingKey was generated
     // already so we can skip things after this guard.
-    match db_handle.get(&serialize(&zkbin.namespace)) {
+    match env
+        .blockchain
+        .lock()
+        .unwrap()
+        .overlay
+        .lock()
+        .unwrap()
+        .get(&db_handle.tree, &serialize(&zkbin.namespace))
+    {
         Ok(v) => {
             if let Some(bytes) = v {
                 // We allow a panic here because this db should never be corrupted in this way.
@@ -632,7 +609,19 @@ pub(crate) fn zkas_db_set(ctx: FunctionEnvMut<Env>, ptr: WasmPtr<u8>, len: u32)
 
     let key = serialize(&zkbin.namespace);
     let value = serialize(&(zkas_bincode, vk_buf));
-    db_batch.insert(key, value);
+    if env
+        .blockchain
+        .lock()
+        .unwrap()
+        .overlay
+        .lock()
+        .unwrap()
+        .insert(&db_handle.tree, &key, &value)
+        .is_err()
+    {
+        error!(target: "runtime::db::zkas_db_set()", "Couldn't insert to db_handle tree");
+        return DB_SET_FAILED
+    }
 
     DB_SUCCESS
 }

src/runtime/import/merkle.rs

@@ -73,11 +73,9 @@ pub(crate) fn merkle_add(ctx: FunctionEnvMut<Env>, ptr: WasmPtr<u8>, len: u32) -
             let db_info = db_info as usize;
             let db_roots = db_roots as usize;
             let db_handles = env.db_handles.borrow();
-            let mut db_batches = env.db_batches.borrow_mut();
             let n_dbs = db_handles.len();
-            let n_bat = db_batches.len();
 
-            if n_dbs <= db_info || n_bat <= db_info || n_dbs <= db_roots || n_bat <= db_roots {
+            if n_dbs <= db_info || n_dbs <= db_roots {
                 error!(target: "runtime::merkle", "Requested DbHandle that is out of bounds");
                 return -2
             }
@@ -114,7 +112,15 @@ pub(crate) fn merkle_add(ctx: FunctionEnvMut<Env>, ptr: WasmPtr<u8>, len: u32) -
             // TODO: Ensure we've read the entire buffer above.
 
             // Read the current tree
-            let ret = match db_info.get(&key) {
+            let ret = match env
+                .blockchain
+                .lock()
+                .unwrap()
+                .overlay
+                .lock()
+                .unwrap()
+                .get(&db_info.tree, &key)
+            {
                 Ok(v) => v,
                 Err(e) => {
                     error!(target: "runtime::merkle", "Internal error getting from tree: {}", e);
@@ -176,12 +182,17 @@ pub(crate) fn merkle_add(ctx: FunctionEnvMut<Env>, ptr: WasmPtr<u8>, len: u32) -
                 error!(target: "runtime::merkle", "Couldn't reserialize modified tree");
                 return -2
             }
-            let db_info_batch = &mut db_batches[info_handle_idx];
-            db_info_batch.insert(key, tree_data);
+
+            // Apply changes to overlay
+            let lock = env.blockchain.lock().unwrap();
+            let mut overlay = lock.overlay.lock().unwrap();
+            if overlay.insert(&db_info.tree, &key, &tree_data).is_err() {
+                error!(target: "runtime::merkle", "Couldn't insert to db_info tree");
+                return -2
+            }
 
             // Here we add the Merkle root to our set of roots
             // TODO: We should probably make sure that this root isn't in the set
-            let db_roots_batch = &mut db_batches[roots_handle_idx];
             for root in new_roots.iter() {
                 // FIXME: Why were we writing the set size here?
                 //let root_index: Vec<u8> = serialize(&(set_size as u32));
@@ -190,8 +201,10 @@ pub(crate) fn merkle_add(ctx: FunctionEnvMut<Env>, ptr: WasmPtr<u8>, len: u32) -
                 let root_value: Vec<u8> = serialize(root);
                 // FIXME: This assert can be used to DoS nodes from contracts
                 assert_eq!(root_value.len(), 32);
-                //db_roots_batch.insert(root_index, root_value);
-                db_roots_batch.insert(root_value, &[]);
+                if overlay.insert(&db_roots.tree, &root_value, &[]).is_err() {
+                    error!(target: "runtime::merkle", "Couldn't insert to db_roots tree");
+                    return -2
+                }
             }
 
             0

src/runtime/vm_runtime.rs

@@ -24,7 +24,6 @@ use std::{
 use darkfi_sdk::{crypto::ContractId, entrypoint};
 use darkfi_serial::serialize;
 use log::{debug, error, info};
-use sled::{transaction::ConflictableTransactionError, Transactional};
 use wasmer::{
     imports, wasmparser::Operator, AsStoreRef, CompilerConfig, Function, FunctionEnv, Instance,
     Memory, MemoryView, Module, Pages, Store, Value, WASM_PAGE_SIZE,
@@ -36,7 +35,7 @@ use wasmer_middlewares::{
 };
 
 use super::{import, import::db::DbHandle, memory::MemoryManipulation};
-use crate::{blockchain::Blockchain, Error, Result};
+use crate::{blockchain::BlockchainOverlayPtr, Error, Result};
 
 /// Name of the wasm linear memory in our guest module
 const MEMORY: &str = "memory";
@@ -75,12 +74,10 @@ impl ContractSection {
 
 /// The wasm vm runtime instantiated for every smart contract that runs.
 pub struct Env {
-    /// Blockchain access
-    pub blockchain: Blockchain,
-    /// sled tree handles used with `db_*`
+    /// Blockchain overlay access
+    pub blockchain: BlockchainOverlayPtr,
+    /// Overlay tree handles used with `db_*`
     pub db_handles: RefCell<Vec<DbHandle>>,
-    /// sled tree batches, indexed the same as `db_handles`.
-    pub db_batches: RefCell<Vec<sled::Batch>>,
     /// The contract ID being executed
     pub contract_id: ContractId,
     /// The compiled wasm bincode being executed,
@@ -123,7 +120,11 @@ pub struct Runtime {
 
 impl Runtime {
     /// Create a new wasm runtime instance that contains the given wasm module.
-    pub fn new(wasm_bytes: &[u8], blockchain: Blockchain, contract_id: ContractId) -> Result<Self> {
+    pub fn new(
+        wasm_bytes: &[u8],
+        blockchain: BlockchainOverlayPtr,
+        contract_id: ContractId,
+    ) -> Result<Self> {
         info!(target: "runtime::vm_runtime", "Instantiating a new runtime");
         // TODO: Add necessary operators
         // This function will be called for each `Operator` encountered during
@@ -154,7 +155,6 @@ impl Runtime {
 
         // Initialize data
         let db_handles = RefCell::new(vec![]);
-        let db_batches = RefCell::new(vec![]);
         let logs = RefCell::new(vec![]);
 
         debug!(target: "runtime::vm_runtime", "Importing functions");
@@ -164,7 +164,6 @@ impl Runtime {
             Env {
                 blockchain,
                 db_handles,
-                db_batches,
                 contract_id,
                 contract_bincode: wasm_bytes.to_vec(),
                 contract_section: ContractSection::Null,
@@ -333,10 +332,10 @@ impl Runtime {
     /// The runtime will look for an `INITIALIZE` symbol in the wasm code, and execute
     /// it if found. Optionally, it is possible to pass in a payload for any kind of special
     /// instructions the developer wants to manage in the initialize function.
-    /// This process is supposed to set up the sled db trees for storing the smart contract
+    /// This process is supposed to set up the overlay trees for storing the smart contract
     /// state, and it can create, delete, modify, read, and write to databases it's allowed to.
-    /// The permissions for this are handled by the `ContractId` in the sled db API so we
-    /// assume that the contract is only able to do write operations on its own sled trees.
+    /// The permissions for this are handled by the `ContractId` in the overlay db API so we
+    /// assume that the contract is only able to do write operations on its own overlay trees.
     pub fn deploy(&mut self, payload: &[u8]) -> Result<()> {
         info!(target: "runtime::vm_runtime", "[wasm-runtime] Running deploy");
 
@@ -346,66 +345,35 @@ impl Runtime {
 
             // We always want to have the zkas db as index 0 in db handles and batches when
             // deploying.
-            let db = &env_mut.blockchain.sled_db;
-
-            let zkas_tree_handle = match env_mut.blockchain.contracts.lookup(
-                db,
-                &env_mut.contract_id,
-                SMART_CONTRACT_ZKAS_DB_NAME,
-            ) {
-                Ok(v) => v,
-                Err(_) => {
-                    // FIXME: All this is deploy code is "vulnerable" and able to init a
-                    // tree regardless of execution success. We can easily delete the db
-                    // if execution fails though, and we should charge gas for db_init.
-                    // and perhaps also for the zkas database in this specific case.
-                    env_mut.blockchain.contracts.init(
-                        db,
-                        &env_mut.contract_id,
-                        SMART_CONTRACT_ZKAS_DB_NAME,
-                    )?
-                }
-            };
+            let contracts = &env_mut.blockchain.lock().unwrap().contracts;
+
+            let zkas_tree_handle =
+                match contracts.lookup(&env_mut.contract_id, SMART_CONTRACT_ZKAS_DB_NAME) {
+                    Ok(v) => v,
+                    Err(_) => {
+                        // FIXME: All this is deploy code is "vulnerable" and able to init a
+                        // tree regardless of execution success. We can easily delete the db
+                        // if execution fails though, and we should charge gas for db_init.
+                        // and perhaps also for the zkas database in this specific case.
+                        contracts.init(&env_mut.contract_id, SMART_CONTRACT_ZKAS_DB_NAME)?
+                    }
+                };
 
             let mut db_handles = env_mut.db_handles.borrow_mut();
-            let mut db_batches = env_mut.db_batches.borrow_mut();
             db_handles.push(DbHandle::new(env_mut.contract_id, zkas_tree_handle));
-            db_batches.push(sled::Batch::default());
         }
 
         debug!(target: "runtime::vm_runtime", "[wasm-runtime] payload: {:?}", payload);
         let _ = self.call(ContractSection::Deploy, payload)?;
 
-        // If the above didn't fail, we write the batches.
-        self.write_batches()?;
-
         // Update the wasm bincode in the WasmStore
         let env_mut = self.ctx.as_mut(&mut self.store);
-        env_mut.blockchain.wasm_bincode.insert(env_mut.contract_id, &env_mut.contract_bincode)?;
-
-        Ok(())
-    }
-
-    /// Execute an atomic sled transaction to write all batches
-    fn write_batches(&mut self) -> Result<()> {
-        let mut dbs = vec![];
-        let mut batches = vec![];
-        let env_mut = self.ctx.as_mut(&mut self.store);
-        for (idx, db) in env_mut.db_handles.get_mut().iter().enumerate() {
-            let batch = env_mut.db_batches.borrow()[idx].clone();
-            dbs.push(db.tree());
-            batches.push(batch);
-        }
-
-        dbs.transaction(|dbs| {
-            for (idx, db) in dbs.iter().enumerate() {
-                db.apply_batch(&batches[idx])?;
-            }
-
-            Ok::<(), ConflictableTransactionError<sled::Error>>(())
-        })?;
-
-        env_mut.blockchain.sled_db.flush()?;
+        env_mut
+            .blockchain
+            .lock()
+            .unwrap()
+            .wasm_bincode
+            .insert(env_mut.contract_id, &env_mut.contract_bincode)?;
 
         Ok(())
     }
@@ -420,7 +388,7 @@ impl Runtime {
     }
 
     /// This function runs after successful execution of `exec` and tries to
-    /// apply the state change to the sled databases.
+    /// apply the state change to the overlay databases.
     /// The runtime will lok for an `UPDATE` symbol in the wasm code, and execute
     /// it if found. The function does not take an arbitrary payload, but just takes
     /// a state update from `env` and passes it into the wasm runtime.
@@ -428,9 +396,6 @@ impl Runtime {
         debug!(target: "runtime::vm_runtime", "apply: {:?}", update);
         let _ = self.call(ContractSection::Update, update)?;
 
-        // If the above didn't fail, we write the batches.
-        self.write_batches()?;
-
         Ok(())
     }
 

src/wallet/walletdb.rs

@@ -48,8 +48,10 @@ pub enum QueryType {
     OptionInteger = 0x02,
     /// OptionBlob gets decoded into `Option<Vec<u8>>`
     OptionBlob = 0x03,
+    /// Text gets decoded into `String`
+    Text = 0x04,
     /// Last type, increment this when you add new types.
-    Last = 0x04,
+    Last = 0x05,
 }
 
 impl From<u8> for QueryType {
@@ -59,6 +61,7 @@ impl From<u8> for QueryType {
             0x01 => Self::Blob,
             0x02 => Self::OptionInteger,
             0x03 => Self::OptionBlob,
+            0x04 => Self::Text,
             _ => unimplemented!(),
         }
     }