Fix rpc retry shutdown during SIGINT & implement tabled for TxInfo

Author: buffrr

node/src/sync.rs

@@ -171,15 +171,17 @@ impl Spaced {
                                 return Err(e);
                             }
                             warn!("Restore: {} - retrying in 1s", e);
-                            std_wait(|| shutdown_signal.try_recv().is_ok(), Duration::from_secs(1));
+                            let mut wait_recv = shutdown.subscribe();
+                            std_wait(|| wait_recv.try_recv().is_ok(), Duration::from_secs(1));
                         }
                         // Even if we couldn't restore just attempt to re-sync
                         let new_tip = self.chain.state.tip.read().expect("read").clone();
                         fetcher.restart(new_tip, &receiver);
                     }
                     BlockEvent::Error(e) => {
                         warn!("Fetcher: {} - retrying in 1s", e);
-                        std_wait(|| shutdown_signal.try_recv().is_ok(), Duration::from_secs(1));
+                        let mut wait_recv = shutdown.subscribe();
+                        std_wait(|| wait_recv.try_recv().is_ok(), Duration::from_secs(1));
                         // Even if we couldn't restore just attempt to re-sync
                         let new_tip = self.chain.state.tip.read().expect("read").clone();
                         fetcher.restart(new_tip, &receiver);

node/src/wallets.rs

@@ -10,6 +10,7 @@ use log::{info, warn};
 use protocol::{bitcoin::Txid, constants::ChainAnchor, hasher::{KeyHasher, SpaceKey}, script::SpaceScript, slabel::SLabel, FullSpaceOut, SpaceOut};
 use serde::{Deserialize, Serialize};
 use serde_json::json;
+use tabled::Tabled;
 use tokio::{
     select,
     sync::{broadcast, mpsc, mpsc::Receiver, oneshot},
@@ -45,16 +46,37 @@ pub struct ListSpacesResponse {
     pub owned: Vec<FullSpaceOut>,
 }
 
-#[derive(Debug, Clone, Serialize, Deserialize)]
+
+#[derive(Tabled, Debug, Clone, Serialize, Deserialize)]
+#[tabled(rename_all = "UPPERCASE")]
 pub struct TxInfo {
     pub txid: Txid,
     pub confirmed: bool,
     pub sent: Amount,
     pub received: Amount,
+    #[tabled(display_with = "display_fee")]
     pub fee: Option<Amount>,
+    #[tabled(rename = "DETAILS", display_with = "display_events")]
     pub events: Vec<TxEvent>,
 }
 
+fn display_fee(fee: &Option<Amount>) -> String {
+    match fee {
+        None => "--".to_string(),
+        Some(fee) => fee.to_string()
+    }
+}
+
+fn display_events(events: &Vec<TxEvent>) -> String {
+    events
+        .iter()
+        .map(|e|
+            format!("{} {}",
+                    e.kind,
+                    e.space.as_ref().map(|s| s.clone()).unwrap_or("".to_string())))
+        .collect::<Vec<String>>().join("\n")
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct WalletResponse {
     pub result: Vec<TxResponse>,
@@ -429,7 +451,7 @@ impl RpcWallet {
         mut state: LiveSnapshot,
         mut wallet: SpacesWallet,
         mut commands: Receiver<WalletCommand>,
-        mut shutdown: broadcast::Receiver<()>,
+        shutdown: broadcast::Sender<()>,
         num_workers: usize,
     ) -> anyhow::Result<()> {
         let (fetcher, receiver) = BlockFetcher::new(source.clone(), num_workers);
@@ -442,11 +464,12 @@ impl RpcWallet {
             }
         };
 
+        let mut shutdown_recv = shutdown.subscribe();
         fetcher.start(wallet_tip);
         let mut synced_at_least_once = false;
         let mut last_mempool_check = Instant::now();
         loop {
-            if shutdown.try_recv().is_ok() {
+            if shutdown_recv.try_recv().is_ok() {
                 info!("Shutting down wallet sync");
                 break;
             }
@@ -542,7 +565,8 @@ impl RpcWallet {
                     }
                     BlockEvent::Error(e) => {
                         warn!("Fetcher: {} - retrying in 1s", e);
-                        std_wait(|| shutdown.try_recv().is_ok(), Duration::from_secs(1));
+                        let mut wait_recv = shutdown.subscribe();
+                        std_wait(|| wait_recv.try_recv().is_ok(), Duration::from_secs(1));
                         fetcher.restart(wallet_tip, &receiver);
                     }
                 }
@@ -1091,7 +1115,7 @@ impl RpcWallet {
                         let wallet_name = loaded.export.label.clone();
                         let wallet_chain = store.clone();
                         let rpc = rpc.clone();
-                        let wallet_shutdown = shutdown.subscribe();
+                        let wallet_shutdown = shutdown.clone();
                         let (tx, rx) = oneshot::channel();
 
                         std::thread::spawn(move || {