drk: Block scanning

Cargo.lock

@@ -1681,7 +1681,10 @@ dependencies = [
  "prettytable-rs",
  "rand",
  "serde_json",
+ "signal-hook",
+ "signal-hook-async-std",
  "simplelog",
+ "smol",
  "sqlx",
  "url",
 ]
@@ -4039,9 +4042,9 @@ checksum = "f67ad224767faa3c7d8b6d91985b78e70a1324408abcb1cfcc2be4c06bc06043"
 
 [[package]]
 name = "smol"
-version = "1.2.5"
+version = "1.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "85cf3b5351f3e783c1d79ab5fc604eeed8b8ae9abd36b166e8b87a089efd85e4"
+checksum = "13f2b548cd8447f8de0fdf1c592929f70f4fc7039a05e47404b0d096ec6987a1"
 dependencies = [
  "async-channel",
  "async-executor",
@@ -4052,7 +4055,6 @@ dependencies = [
  "async-process",
  "blocking",
  "futures-lite",
- "once_cell",
 ]
 
 [[package]]

Makefile

@@ -37,6 +37,7 @@ zkas: $(BINDEPS)
 
 contracts: zkas
 	$(MAKE) -C src/contract/money
+	$(MAKE) -C src/contract/dao
 
 $(PROOFS_BIN): $(PROOFS)
 	./zkas $(basename $@) -o $@

bin/darkfid/src/main.rs

@@ -208,6 +208,9 @@ impl RequestHandler for Darkfid {
             // Blockchain methods
             // ==================
             Some("blockchain.get_slot") => return self.blockchain_get_slot(req.id, params).await,
+            Some("blockchain.last_known_slot") => {
+                return self.blockchain_last_known_slot(req.id, params).await
+            }
             Some("blockchain.merkle_roots") => {
                 return self.blockchain_merkle_roots(req.id, params).await
             }

bin/darkfid/src/rpc_blockchain.rs

@@ -20,7 +20,7 @@ use darkfi_sdk::{
     crypto::{ContractId, MerkleNode},
     db::ZKAS_DB_NAME,
 };
-use darkfi_serial::deserialize;
+use darkfi_serial::{deserialize, serialize};
 use log::{debug, error};
 use serde_json::{json, Value};
 
@@ -62,9 +62,25 @@ impl Darkfid {
             return server_error(RpcError::UnknownSlot, id, None)
         }
 
-        // TODO: Return block as JSON
-        debug!("{:#?}", blocks[0]);
-        JsonResponse::new(json!(true), id).into()
+        JsonResponse::new(json!(serialize(&blocks[0])), id).into()
+    }
+
+    // RPCAPI:
+    // Queries the blockchain database to find the last known slot
+    //
+    // --> {"jsonrpc": "2.0", "method": "blockchain.last_known_slot", "params": [], "id": 1}
+    // <-- {"jsonrpc": "2.0", "result": 1234, "id": 1}
+    pub async fn blockchain_last_known_slot(&self, id: Value, params: &[Value]) -> JsonResult {
+        if !params.is_empty() {
+            return JsonError::new(InvalidParams, None, id).into()
+        }
+
+        let blockchain = { self.validator_state.read().await.blockchain.clone() };
+        let Ok(last_slot) = blockchain.last() else {
+                return JsonError::new(InternalError, None, id).into()
+        };
+
+        JsonResponse::new(json!(last_slot.0), id).into()
     }
 
     // RPCAPI:

bin/drk/Cargo.toml

@@ -20,6 +20,9 @@ darkfi-money-contract = {path = "../../src/contract/money", features = ["no-entr
 prettytable-rs = "0.9.0"
 rand = "0.8.5"
 serde_json = "1.0.89"
+smol = "1.3.0"
 simplelog = "0.12.0"
+signal-hook-async-std = "0.2.2"
+signal-hook = "0.3.14"
 sqlx = {version = "0.6.2", features = ["runtime-async-std-native-tls", "sqlite"]}
 url = "2.3.1"

bin/drk/src/main.rs

@@ -152,6 +152,12 @@ enum Subcmd {
     /// 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,
+
+    /// Scan the blockchain and parse relevant transactions
+    Scan {
+        /// Slot number to start scanning from (optional)
+        slot: Option<u64>,
+    },
 }
 
 pub struct Drk {
@@ -389,5 +395,18 @@ async fn main() -> Result<()> {
 
             Ok(())
         }
+
+        Subcmd::Scan { slot } => {
+            let rpc_client = RpcClient::new(args.endpoint)
+                .await
+                .with_context(|| "Could not connect to darkfid RPC endpoint")?;
+
+            let drk = Drk { rpc_client };
+
+            drk.scan_blocks(slot).await.with_context(|| "Failed during scanning")?;
+            eprintln!("Finished scanning blockchain");
+
+            Ok(())
+        }
     }
 }

bin/drk/src/rpc_blockchain.rs

@@ -17,7 +17,7 @@
  */
 
 use anyhow::{anyhow, Result};
-use async_std::task;
+use async_std::{stream::StreamExt, task};
 use darkfi::{
     consensus::BlockInfo,
     rpc::{
@@ -34,7 +34,8 @@ use darkfi_money_contract::{
         MONEY_COINS_COL_IS_SPENT, MONEY_COINS_COL_LEAF_POSITION, MONEY_COINS_COL_MEMO,
         MONEY_COINS_COL_NULLIFIER, MONEY_COINS_COL_SECRET, MONEY_COINS_COL_SERIAL,
         MONEY_COINS_COL_TOKEN_BLIND, MONEY_COINS_COL_TOKEN_ID, MONEY_COINS_COL_VALUE,
-        MONEY_COINS_COL_VALUE_BLIND, MONEY_COINS_TABLE,
+        MONEY_COINS_COL_VALUE_BLIND, MONEY_COINS_TABLE, MONEY_INFO_COL_LAST_SCANNED_SLOT,
+        MONEY_INFO_TABLE,
     },
     state::{MoneyTransferParams, Output},
     MoneyFunction,
@@ -46,6 +47,8 @@ use darkfi_sdk::{
 };
 use darkfi_serial::{deserialize, serialize};
 use serde_json::json;
+use signal_hook::consts::{SIGINT, SIGQUIT, SIGTERM};
+use signal_hook_async_std::Signals;
 use url::Url;
 
 use super::Drk;
@@ -263,4 +266,83 @@ impl Drk {
         let txid = serde_json::from_value(rep)?;
         Ok(txid)
     }
+
+    /// Queries darkfid for a block with given slot
+    async fn get_block_by_slot(&self, slot: u64) -> Result<Option<BlockInfo>> {
+        let req = JsonRequest::new("blockchain.get_slot", json!([slot]));
+
+        // This API is weird, we need some way of telling it's an empty slot and
+        // not an error
+        match self.rpc_client.request(req).await {
+            Ok(v) => {
+                let block_bytes: Vec<u8> = serde_json::from_value(v)?;
+                let block = deserialize(&block_bytes)?;
+                Ok(Some(block))
+            }
+
+            Err(_) => Ok(None),
+        }
+    }
+
+    /// Scans the blockchain optionally starting from the given slot for relevant
+    /// money transfer transactions. Alternatively it looks for a checkpoint in the
+    /// wallet to start scanning from.
+    pub async fn scan_blocks(&self, slot: Option<u64>) -> Result<()> {
+        let mut sl = if let Some(sl) = slot { sl } else { self.wallet_last_scanned_slot().await? };
+
+        let req = JsonRequest::new("blockchain.last_known_slot", json!([]));
+        let rep = self.rpc_client.request(req).await?;
+
+        let last: u64 = serde_json::from_value(rep)?;
+
+        eprintln!("Requested to scan from slot number: {}", sl);
+        eprintln!("Last known slot number reported by darkfid: {}", last);
+
+        // We set this up to handle an interrupt
+        let mut signals = Signals::new(&[SIGTERM, SIGINT, SIGQUIT])?;
+        let handle = signals.handle();
+        let (term_tx, _term_rx) = smol::channel::bounded::<()>(1);
+
+        let term_tx_ = term_tx.clone();
+        let signals_task = task::spawn(async move {
+            while let Some(signal) = signals.next().await {
+                match signal {
+                    SIGTERM | SIGINT | SIGQUIT => term_tx_.close(),
+                    _ => unreachable!(),
+                };
+            }
+        });
+
+        while !term_tx.is_closed() {
+            if sl == last {
+                term_tx.close();
+                break
+            }
+
+            sl += 1;
+
+            eprint!("Requesting slot {}... ", sl);
+            if let Some(block) = self.get_block_by_slot(sl).await? {
+                eprintln!("Found");
+                self.scan_block(&block).await?;
+            } else {
+                eprintln!("Not found");
+            }
+
+            // Write down the slot number into back to the wallet
+            // TODO: Why doesn't it work?
+            let query = format!(
+                "INSERT INTO {} ({}) VALUES (?1);",
+                MONEY_INFO_TABLE, MONEY_INFO_COL_LAST_SCANNED_SLOT
+            );
+            let params = json!([query, QueryType::Integer as u8, sl]);
+            let req = JsonRequest::new("wallet.exec_sql", params);
+            let _ = self.rpc_client.request(req).await?;
+        }
+
+        handle.close();
+        signals_task.await;
+
+        Ok(())
+    }
 }

bin/drk/src/rpc_wallet.rs

@@ -25,9 +25,9 @@ use darkfi_money_contract::client::{
     MONEY_COINS_COL_IS_SPENT, MONEY_COINS_COL_LEAF_POSITION, MONEY_COINS_COL_MEMO,
     MONEY_COINS_COL_NULLIFIER, MONEY_COINS_COL_SECRET, MONEY_COINS_COL_SERIAL,
     MONEY_COINS_COL_TOKEN_BLIND, MONEY_COINS_COL_TOKEN_ID, MONEY_COINS_COL_VALUE,
-    MONEY_COINS_COL_VALUE_BLIND, MONEY_COINS_TABLE, MONEY_KEYS_COL_IS_DEFAULT,
-    MONEY_KEYS_COL_PUBLIC, MONEY_KEYS_COL_SECRET, MONEY_KEYS_TABLE, MONEY_TREE_COL_TREE,
-    MONEY_TREE_TABLE,
+    MONEY_COINS_COL_VALUE_BLIND, MONEY_COINS_TABLE, MONEY_INFO_COL_LAST_SCANNED_SLOT,
+    MONEY_INFO_TABLE, MONEY_KEYS_COL_IS_DEFAULT, MONEY_KEYS_COL_PUBLIC, MONEY_KEYS_COL_SECRET,
+    MONEY_KEYS_TABLE, MONEY_TREE_COL_TREE, MONEY_TREE_TABLE,
 };
 use darkfi_sdk::{
     crypto::{
@@ -84,6 +84,16 @@ impl Drk {
             println!("Successfully initialized Merkle tree");
         }
 
+        if let Err(_) = self.wallet_last_scanned_slot().await {
+            let query = format!(
+                "INSERT INTO {} ({}) VALUES (?1);",
+                MONEY_INFO_TABLE, MONEY_INFO_COL_LAST_SCANNED_SLOT
+            );
+            let params = json!([query, QueryType::Integer as u8, 0]);
+            let req = JsonRequest::new("wallet.exec_sql", params);
+            let _ = self.rpc_client.request(req).await?;
+        }
+
         Ok(())
     }
 
@@ -362,6 +372,18 @@ impl Drk {
         Ok(tree)
     }
 
+    /// Get the last scanned slot from the wallet
+    pub async fn wallet_last_scanned_slot(&self) -> Result<u64> {
+        let query =
+            format!("SELECT {} FROM {};", MONEY_INFO_COL_LAST_SCANNED_SLOT, MONEY_INFO_TABLE);
+
+        let params = json!([query, QueryType::Integer as u8, MONEY_INFO_COL_LAST_SCANNED_SLOT]);
+        let req = JsonRequest::new("wallet.query_row_single", params);
+        let rep = self.rpc_client.request(req).await?;
+
+        Ok(serde_json::from_value(rep[0].clone())?)
+    }
+
     /// Mark a coin in the wallet as spent
     pub async fn mark_spent_coin(&self, coin: &Coin) -> Result<()> {
         let query = format!(