drk: restored scan ux by introducing print or insert functionality to output buffer

bin/drk/src/cli_util.rs

@@ -21,6 +21,7 @@ use std::{
 };
 
 use rodio::{Decoder, OutputStream, Sink};
+use smol::channel::Sender;
 use structopt_toml::clap::{App, Arg, Shell, SubCommand};
 
 use darkfi::{
@@ -549,9 +550,45 @@ pub fn generate_completions(shell: &str) -> Result<String> {
     Ok(String::from_utf8(buf)?)
 }
 
-/// Auxiliary function to generate provided shell completions.
+/// Auxiliary function to print provided string buffer.
 pub fn print_output(buf: &[String]) {
     for line in buf {
         println!("{line}");
     }
 }
+
+/// Auxiliary function to print or insert provided messages to given
+/// buffer reference. If a channel sender is provided, the messages
+/// are send to that instead.
+pub async fn append_or_print(
+    buf: &mut Vec<String>,
+    sender: Option<&Sender<Vec<String>>>,
+    print: &bool,
+    messages: Vec<String>,
+) {
+    // Send the messages to the channel, if provided
+    if let Some(sender) = sender {
+        if let Err(e) = sender.send(messages).await {
+            let err_msg = format!("[append_or_print] Sending messages to channel failed: {e}");
+            if *print {
+                println!("{err_msg}");
+            } else {
+                buf.push(err_msg);
+            }
+        }
+        return
+    }
+
+    // Print the messages
+    if *print {
+        for msg in messages {
+            println!("{msg}");
+        }
+        return
+    }
+
+    // Insert the messages in the buffer
+    for msg in messages {
+        buf.push(msg);
+    }
+}

bin/drk/src/interactive.rs

@@ -48,8 +48,8 @@ use darkfi_serial::{deserialize_async, serialize_async};
 
 use crate::{
     cli_util::{
-        generate_completions, kaching, parse_token_pair, parse_tx_from_input, parse_value_pair,
-        print_output,
+        append_or_print, generate_completions, kaching, parse_token_pair, parse_tx_from_input,
+        parse_value_pair, print_output,
     },
     money::BALANCE_BASE10_DECIMALS,
     rpc::subscribe_blocks,
@@ -59,8 +59,7 @@ use crate::{
 
 // TODO:
 //  1. Add rest commands handling, along with their completions, hints and help message.
-//  2. Subscribe/scan ux is a bit flaky, fix it.
-//  3. Create a transactions cache in the wallet db, so you can use it to handle them.
+//  2. Create a transactions cache in the wallet db, so you can use it to handle them.
 
 /// Auxiliary function to print the help message.
 fn help(output: &mut Vec<String>) {
@@ -295,7 +294,7 @@ pub async fn interactive(drk: &DrkPtr, endpoint: &Url, history_path: &str, ex: &
 
         // Process each command
         let mut output = vec![];
-        'commands_loop: for command in commands {
+        'commands_loop: for (command_index, command) in commands.iter().enumerate() {
             let mut input = output;
             output = vec![];
 
@@ -322,7 +321,7 @@ pub async fn interactive(drk: &DrkPtr, endpoint: &Url, history_path: &str, ex: &
                 }
                 (parts[0], Some(file), append)
             } else {
-                (command, None, false)
+                (*command, None, false)
             };
 
             // Check if we have to use a file as input
@@ -410,7 +409,16 @@ pub async fn interactive(drk: &DrkPtr, endpoint: &Url, history_path: &str, ex: &
                 }
                 "snooze" => snooze_active = true,
                 "unsnooze" => snooze_active = false,
-                "scan" => handle_scan(drk, &subscription_active, &parts, &mut output).await,
+                "scan" => {
+                    handle_scan(
+                        drk,
+                        &subscription_active,
+                        &parts,
+                        &mut output,
+                        &(command_index + 1 == commands.len() && file.is_none()),
+                    )
+                    .await
+                }
                 _ => output.push(format!("Unreconized command: {}", parts[0])),
             }
 
@@ -1302,15 +1310,17 @@ async fn handle_scan(
     subscription_active: &bool,
     parts: &[&str],
     output: &mut Vec<String>,
+    print: &bool,
 ) {
     if *subscription_active {
-        output.push(String::from("Subscription is already active!"));
+        append_or_print(output, None, print, vec![String::from("Subscription is already active!")])
+            .await;
         return
     }
 
     // Check correct command structure
     if parts.len() != 1 && parts.len() != 3 {
-        output.push(String::from("Malformed `scan` command"));
+        append_or_print(output, None, print, vec![String::from("Malformed `scan` command")]).await;
         return
     }
 
@@ -1318,28 +1328,40 @@ async fn handle_scan(
     let lock = drk.read().await;
     if parts.len() == 3 {
         if parts[1] != "--reset" {
-            output.push(String::from("Malformed `scan` command"));
-            output.push(String::from("Usage: scan --reset <height>"));
+            append_or_print(
+                output,
+                None,
+                print,
+                vec![
+                    String::from("Malformed `scan` command"),
+                    String::from("Usage: scan --reset <height>"),
+                ],
+            )
+            .await;
             return
         }
 
         let height = match u32::from_str(parts[2]) {
             Ok(h) => h,
             Err(e) => {
-                output.push(format!("Invalid reset height: {e:?}"));
+                append_or_print(output, None, print, vec![format!("Invalid reset height: {e:?}")])
+                    .await;
                 return
             }
         };
 
-        if let Err(e) = lock.reset_to_height(height, output) {
-            output.push(format!("Failed during wallet reset: {e:?}"));
+        let mut buf = vec![];
+        if let Err(e) = lock.reset_to_height(height, &mut buf) {
+            buf.push(format!("Failed during wallet reset: {e:?}"));
+            append_or_print(output, None, print, buf).await;
             return
         }
+        append_or_print(output, None, print, buf).await;
     }
 
-    if let Err(e) = lock.scan_blocks(output).await {
-        output.push(format!("Failed during scanning: {e:?}"));
+    if let Err(e) = lock.scan_blocks(output, None, print).await {
+        append_or_print(output, None, print, vec![format!("Failed during scanning: {e:?}")]).await;
         return
     }
-    output.push(String::from("Finished scanning blockchain"));
+    append_or_print(output, None, print, vec![String::from("Finished scanning blockchain")]).await;
 }

bin/drk/src/main.rs

@@ -2000,19 +2000,20 @@ async fn realmain(args: Args, ex: ExecutorPtr) -> Result<()> {
             )
             .await;
 
-            let mut output = vec![];
             if let Some(height) = reset {
-                if let Err(e) = drk.reset_to_height(height, &mut output) {
+                let mut buf = vec![];
+                if let Err(e) = drk.reset_to_height(height, &mut buf) {
+                    print_output(&buf);
                     eprintln!("Failed during wallet reset: {e:?}");
                     exit(2);
                 }
+                print_output(&buf);
             }
 
-            if let Err(e) = drk.scan_blocks(&mut output).await {
+            if let Err(e) = drk.scan_blocks(&mut vec![], None, &true).await {
                 eprintln!("Failed during scanning: {e:?}");
                 exit(2);
             }
-            print_output(&output);
             println!("Finished scanning blockchain");
 
             drk.stop_rpc_client().await

bin/drk/src/rpc.rs

@@ -51,6 +51,7 @@ use darkfi_serial::{deserialize_async, serialize_async};
 
 use crate::{
     cache::{CacheOverlay, CacheSmt, CacheSmtStorage, SLED_MONEY_SMT_TREE},
+    cli_util::append_or_print,
     dao::{SLED_MERKLE_TREES_DAO_DAOS, SLED_MERKLE_TREES_DAO_PROPOSALS},
     error::{WalletDbError, WalletDbResult},
     money::SLED_MERKLE_TREES_MONEY,
@@ -284,7 +285,12 @@ impl Drk {
     /// Scans the blockchain for wallet relevant transactions,
     /// starting from the last scanned block. If a reorg has happened,
     /// we revert to its previous height and then scan from there.
-    pub async fn scan_blocks(&self, output: &mut Vec<String>) -> WalletDbResult<()> {
+    pub async fn scan_blocks(
+        &self,
+        output: &mut Vec<String>,
+        sender: Option<&Sender<Vec<String>>>,
+        print: &bool,
+    ) -> WalletDbResult<()> {
         // Grab last scanned block height
         let (mut height, hash) = self.get_last_scanned_block()?;
 
@@ -294,7 +300,13 @@ impl Drk {
             // Check if block was found
             Err(Error::JsonRpcError((-32121, _))) => None,
             Err(e) => {
-                output.push(format!("[scan_blocks] RPC client request failed: {e:?}"));
+                append_or_print(
+                    output,
+                    sender,
+                    print,
+                    vec![format!("[scan_blocks] RPC client request failed: {e:?}")],
+                )
+                .await;
                 return Err(WalletDbError::GenericError)
             }
         };
@@ -302,7 +314,8 @@ impl Drk {
         // Check if a reorg has happened
         if block.is_none() || hash != block.unwrap().hash().to_string() {
             // Find the exact block height the reorg happened
-            output.push(String::from("A reorg has happened, finding last known common block..."));
+            let mut buf =
+                vec![String::from("A reorg has happened, finding last known common block...")];
             height = height.saturating_sub(1);
             while height != 0 {
                 // Grab our scanned block hash for that height
@@ -314,7 +327,8 @@ impl Drk {
                     // Check if block was found
                     Err(Error::JsonRpcError((-32121, _))) => None,
                     Err(e) => {
-                        output.push(format!("[scan_blocks] RPC client request failed: {e:?}"));
+                        buf.push(format!("[scan_blocks] RPC client request failed: {e:?}"));
+                        append_or_print(output, sender, print, buf).await;
                         return Err(WalletDbError::GenericError)
                     }
                 };
@@ -326,8 +340,9 @@ impl Drk {
                 }
 
                 // Reset to its height
-                output.push(format!("Last common block found: {height} - {scanned_block_hash}"));
-                self.reset_to_height(height, output)?;
+                buf.push(format!("Last common block found: {height} - {scanned_block_hash}"));
+                self.reset_to_height(height, &mut buf)?;
+                append_or_print(output, sender, print, buf).await;
                 break
             }
         }
@@ -335,7 +350,9 @@ impl Drk {
         // If last scanned block is genesis(0) we reset,
         // otherwise continue with the next block height.
         if height == 0 {
-            self.reset(output)?;
+            let mut buf = vec![];
+            self.reset(&mut buf)?;
+            append_or_print(output, sender, print, buf).await;
         } else {
             height += 1;
         }
@@ -344,24 +361,32 @@ impl Drk {
         let mut scan_cache = match self.scan_cache().await {
             Ok(c) => c,
             Err(e) => {
-                output.push(format!("[scan_blocks] Generating scan cache failed: {e:?}"));
+                append_or_print(
+                    output,
+                    sender,
+                    print,
+                    vec![format!("[scan_blocks] Generating scan cache failed: {e:?}")],
+                )
+                .await;
                 return Err(WalletDbError::GenericError)
             }
         };
 
         loop {
             // Grab last confirmed block
-            output.push(format!("Requested to scan from block number: {height}"));
+            let mut buf = vec![format!("Requested to scan from block number: {height}")];
             let (last_height, last_hash) = match self.get_last_confirmed_block().await {
                 Ok(last) => last,
                 Err(e) => {
-                    output.push(format!("[scan_blocks] RPC client request failed: {e:?}"));
+                    buf.push(format!("[scan_blocks] RPC client request failed: {e:?}"));
+                    append_or_print(output, sender, print, buf).await;
                     return Err(WalletDbError::GenericError)
                 }
             };
-            output.push(format!(
+            buf.push(format!(
                 "Last confirmed block reported by darkfid: {last_height} - {last_hash}"
             ));
+            append_or_print(output, sender, print, buf).await;
 
             // Already scanned last confirmed block
             if height > last_height {
@@ -369,22 +394,26 @@ impl Drk {
             }
 
             while height <= last_height {
-                output.push(format!("Requesting block {height}..."));
+                let mut buf = vec![format!("Requesting block {height}...")];
+                buf.push(format!("Requesting block {height}..."));
                 let block = match self.get_block_by_height(height).await {
                     Ok(b) => b,
                     Err(e) => {
-                        output.push(format!("[scan_blocks] RPC client request failed: {e:?}"));
+                        buf.push(format!("[scan_blocks] RPC client request failed: {e:?}"));
+                        append_or_print(output, sender, print, buf).await;
                         return Err(WalletDbError::GenericError)
                     }
                 };
-                output.push(format!("Block {height} received! Scanning block..."));
+                buf.push(format!("Block {height} received! Scanning block..."));
                 if let Err(e) = self.scan_block(&mut scan_cache, &block).await {
-                    output.push(format!("[scan_blocks] Scan block failed: {e:?}"));
+                    buf.push(format!("[scan_blocks] Scan block failed: {e:?}"));
+                    append_or_print(output, sender, print, buf).await;
                     return Err(WalletDbError::GenericError)
                 };
                 for msg in scan_cache.flush_messages() {
-                    output.push(msg);
+                    buf.push(msg);
                 }
+                append_or_print(output, sender, print, buf).await;
                 height += 1;
             }
         }
@@ -573,29 +602,23 @@ pub async fn subscribe_blocks(
 ) -> Result<()> {
     // First we do a clean scan
     let lock = drk.read().await;
-    let mut output = vec![];
-    if let Err(e) = lock.scan_blocks(&mut output).await {
+    if let Err(e) = lock.scan_blocks(&mut vec![], Some(&shell_sender), &false).await {
         let err_msg = format!("Failed during scanning: {e:?}");
-        output.push(err_msg.clone());
-        shell_sender.send(output).await?;
+        shell_sender.send(vec![err_msg.clone()]).await?;
         return Err(Error::Custom(err_msg))
     }
-    output.push(String::from("Finished scanning blockchain"));
-    shell_sender.send(output).await?;
+    shell_sender.send(vec![String::from("Finished scanning blockchain")]).await?;
 
     // Grab last confirmed block height
     let (last_confirmed_height, _) = lock.get_last_confirmed_block().await?;
 
     // Handle genesis(0) block
     if last_confirmed_height == 0 {
-        output = vec![];
-        if let Err(e) = lock.scan_blocks(&mut output).await {
+        if let Err(e) = lock.scan_blocks(&mut vec![], Some(&shell_sender), &false).await {
             let err_msg = format!("[subscribe_blocks] Scanning from genesis block failed: {e:?}");
-            output.push(err_msg.clone());
-            shell_sender.send(output).await?;
+            shell_sender.send(vec![err_msg.clone()]).await?;
             return Err(Error::Custom(err_msg))
         }
-        shell_sender.send(output).await?;
     }
 
     // Grab last confirmed block again