wallet: broadcasting delay, repeat tx action

src/wallet/config.rs

@@ -46,6 +46,8 @@ pub struct WalletConfig {
     pub enable_tor_listener: Option<bool>,
     /// Wallet API port.
     pub api_port: Option<u16>,
+    /// Delay in blocks before another transaction broadcasting attempt.
+    pub tx_broadcast_timeout: Option<u64>,
 }
 
 /// Base wallets directory name.
@@ -68,6 +70,9 @@ impl WalletConfig {
     /// Default account name value.
     pub const DEFAULT_ACCOUNT_LABEL: &'static str = "default";
 
+    /// Default value of timeout for broadcasting transaction in blocks.
+    pub const BROADCASTING_TIMEOUT_DEFAULT: u64 = 10;
+
     /// Create new wallet config.
     pub fn create(name: String, conn_method: &ConnectionMethod) -> WalletConfig {
         // Setup configuration path.
@@ -88,6 +93,7 @@ impl WalletConfig {
             use_dandelion: Some(true),
             enable_tor_listener: Some(false),
             api_port: Some(rand::rng().random_range(10000..30000)),
+            tx_broadcast_timeout: Some(Self::BROADCASTING_TIMEOUT_DEFAULT),
         };
         Settings::write_to_file(&config, config_path);
         config

src/wallet/store.rs

@@ -16,32 +16,37 @@ use std::sync::{Arc, RwLock};
 use rkv::backend::{Lmdb, LmdbDatabase, LmdbEnvironment};
 use rkv::{IntegerStore, Manager, Rkv, StoreOptions, Value};
 
-/// Transaction confirmation height storage.
+/// Transaction height storage.
 pub struct TxHeightStore {
-    env_arc: Arc<RwLock<Rkv<LmdbEnvironment>>>,
-    store: IntegerStore<LmdbDatabase, u32>
+    env: Arc<RwLock<Rkv<LmdbEnvironment>>>,
+    /// Confirmed heights.
+    confirmed: IntegerStore<LmdbDatabase, u32>,
+    /// Broadcasting heights.
+    broadcasting: IntegerStore<LmdbDatabase, u32>
 }
 
 impl TxHeightStore {
-    /// Create new transaction height storage at provided directory.
+    /// Create new transaction height storage from provided directory.
     pub fn new(dir: String) -> Self {
         let mut manager = Manager::<LmdbEnvironment>::singleton().write().unwrap();
         let env_arc = manager.get_or_create(std::path::Path::new(&dir), Rkv::new::<Lmdb>).unwrap();
 
         let env_arc_store = env_arc.clone();
         let env = env_arc_store.read().unwrap();
-        let store = env.open_integer("tx_height", StoreOptions::create()).unwrap();
+        let confirmed = env.open_integer("tx_height", StoreOptions::create()).unwrap();
+        let broadcasting = env.open_integer("broadcast_tx_height", StoreOptions::create()).unwrap();
         Self {
-            env_arc,
-            store
+            env: env_arc,
+            confirmed,
+            broadcasting
         }
     }
 
     /// Read transaction height from database.
     pub fn read_tx_height(&self, id: u32) -> Option<u64> {
-        let env = self.env_arc.read().unwrap();
+        let env = self.env.read().unwrap();
         let reader = env.read().unwrap();
-        if let Ok(value) = self.store.get(&reader, id) {
+        if let Ok(value) = self.confirmed.get(&reader, id) {
             if let Some(height) = value {
                 return match height {
                     Value::U64(v) => Some(v),
@@ -55,9 +60,41 @@ impl TxHeightStore {
 
     /// Write transaction height to database.
     pub fn write_tx_height(&self, id: u32, height: u64) {
-        let env = self.env_arc.read().unwrap();
+        let env = self.env.read().unwrap();
         let mut writer = env.write().unwrap();
-        self.store.put(&mut writer, id, &Value::U64(height)).unwrap();
+        self.confirmed.put(&mut writer, id, &Value::U64(height)).unwrap();
+        writer.commit().unwrap();
+    }
+
+    /// Read broadcasting height from database.
+    pub fn read_broadcasting_height(&self, id: u32) -> Option<u64> {
+        let env = self.env.read().unwrap();
+        let reader = env.read().unwrap();
+        if let Ok(value) = self.broadcasting.get(&reader, id) {
+            if let Some(height) = value {
+                return match height {
+                    Value::U64(v) => Some(v),
+                    _ => None
+                };
+            }
+            return None;
+        }
+        None
+    }
+
+    /// Write broadcasting height to database.
+    pub fn write_broadcasting_height(&self, id: u32, height: u64) {
+        let env = self.env.read().unwrap();
+        let mut writer = env.write().unwrap();
+        self.broadcasting.put(&mut writer, id, &Value::U64(height)).unwrap();
+        writer.commit().unwrap();
+    }
+
+    /// Delete broadcasting height from database.
+    pub fn delete_broadcasting_height(&self, id: u32) {
+        let env = self.env.read().unwrap();
+        let mut writer = env.write().unwrap();
+        self.broadcasting.delete(&mut writer, id).unwrap_or_default();
         writer.commit().unwrap();
     }
 }

src/wallet/types.rs

@@ -12,14 +12,15 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::sync::Arc;
-
 use grin_keychain::ExtKeychain;
 use grin_util::Mutex;
 use grin_wallet_impls::{DefaultLCProvider, HTTPNodeClient};
 use grin_wallet_libwallet::{Error, Slate, SlateState, SlatepackAddress, TxLogEntry, TxLogEntryType, WalletInfo, WalletInst};
 use grin_wallet_util::OnionV3Address;
 use serde_derive::{Deserialize, Serialize};
+use std::sync::Arc;
+
+use crate::wallet::Wallet;
 
 /// Mnemonic phrase word.
 #[derive(Clone)]
@@ -212,10 +213,15 @@ pub struct WalletTransaction {
     pub data: TxLogEntry,
     /// State of transaction Slate.
     pub state: SlateState,
-    /// Calculated transaction amount between debited and credited amount.
+
+    /// Transaction amount without fees.
     pub amount: u64,
+    /// Possible receiver of transaction.
+    pub receiver: Option<SlatepackAddress>,
     /// Block height where tx was included.
     pub height: Option<u64>,
+    /// Block height where tx started broadcasting.
+    pub broadcasting_height: Option<u64>,
 
     /// Action on transaction.
     pub action: Option<WalletTransactionAction>,
@@ -224,6 +230,75 @@ pub struct WalletTransaction {
 }
 
 impl WalletTransaction {
+    /// Create new wallet transaction.
+    pub fn new(tx: TxLogEntry,
+               wallet: &Wallet,
+               height: Option<u64>,
+               broadcasting_height: Option<u64>,
+               action: Option<WalletTransactionAction>,
+               action_error: Option<Error>) -> Self {
+        let amount = if tx.amount_debited > tx.amount_credited {
+            tx.amount_debited - tx.amount_credited
+        } else {
+            tx.amount_credited - tx.amount_debited
+        };
+        let receiver: Option<SlatepackAddress> = {
+            if let Some(proof) = &tx.payment_proof {
+                let onion_addr = OnionV3Address::from_bytes(proof.receiver_address.to_bytes());
+                if let Ok(addr) = SlatepackAddress::try_from(onion_addr) {
+                    Some(addr);
+                }
+            }
+            None
+        };
+        let mut t = Self {
+            data: tx,
+            state: SlateState::Unknown,
+            amount,
+            receiver,
+            height,
+            broadcasting_height,
+            action,
+            action_error,
+        };
+        // Update Slate state for unconfirmed.
+        if !t.data.confirmed {
+            t.update_slate_state(wallet);
+        }
+        t
+    }
+
+    /// Update transaction [`Slate`] state for provided wallet.
+    pub fn update_slate_state(&mut self, wallet: &Wallet) {
+        let tx = &self.data;
+        let mut slate = Slate::blank(1, false);
+        slate.id = tx.tx_slate_id.unwrap();
+        slate.state = match tx.tx_type {
+            TxLogEntryType::TxReceived => SlateState::Invoice3,
+            _ => SlateState::Standard3
+        };
+        // Transaction was finalized.
+        if wallet.slatepack_exists(&slate) {
+            self.state = slate.state;
+        } else {
+            slate.id = tx.tx_slate_id.unwrap();
+            slate.state = match tx.tx_type {
+                TxLogEntryType::TxReceived => SlateState::Standard2,
+                _ => SlateState::Invoice2
+            };
+            // Transaction signed to be finalized.
+            if wallet.slatepack_exists(&slate) {
+                self.state = slate.state;
+            } else {
+                // Transaction just was created.
+                self.state = match tx.tx_type {
+                    TxLogEntryType::TxReceived => SlateState::Invoice1,
+                    _ => SlateState::Standard1
+                };
+            }
+        }
+    }
+
     /// Check if transactions can be finalized after receiving response.
     pub fn can_finalize(&self) -> bool {
         !self.cancelling() && !self.data.confirmed &&
@@ -266,12 +341,18 @@ impl WalletTransaction {
 
     /// Check if transaction can be cancelled.
     pub fn can_cancel(&self) -> bool {
-        !self.cancelling() && !self.data.confirmed &&
+        !self.cancelling() && !self.data.confirmed && !self.broadcasting() &&
             (!self.sending_tor() || self.action_error.is_some()) &&
             self.data.tx_type != TxLogEntryType::TxReceivedCancelled &&
             self.data.tx_type != TxLogEntryType::TxSentCancelled
     }
 
+    /// Check if transaction can be sent over Tor again.
+    pub fn can_resend_tor(&self) -> bool {
+       !self.sending_tor() && (self.state == SlateState::Standard1 ||
+            self.state == SlateState::Invoice1) && self.receiver.is_some()
+    }
+
     /// Check if transaction is finalizing.
     pub fn finalizing(&self) -> bool {
         if let Some(a) = self.action.as_ref() {
@@ -280,15 +361,22 @@ impl WalletTransaction {
         false
     }
 
-    /// Get receiver address if payment proof was created.
-    pub fn receiver(&self) -> Option<SlatepackAddress> {
-        if let Some(proof) = &self.data.payment_proof {
-            let onion_addr = OnionV3Address::from_bytes(proof.receiver_address.to_bytes());
-            if let Ok(addr) = SlatepackAddress::try_from(onion_addr) {
-                return Some(addr);
+    /// Check if transaction is broadcasting after finalization.
+    pub fn broadcasting(&self) -> bool {
+        !self.data.confirmed && self.finalized()
+    }
+
+    /// Check if broadcasting of transaction was timed out.
+    pub fn broadcasting_timed_out(&self, wallet: &Wallet) -> bool {
+        if let Some(data) = wallet.get_data() {
+            if self.broadcasting() {
+                let last_height = data.info.last_confirmed_height;
+                let broadcasting_height = self.broadcasting_height.unwrap_or(0);
+                let delay = wallet.broadcasting_delay();
+                return last_height - broadcasting_height > delay;
             }
         }
-        None
+        false
     }
 }
 
@@ -301,7 +389,7 @@ pub enum WalletTask {
     /// * amount
     /// * receiver
     Send(u64, Option<SlatepackAddress>),
-    /// Resend request over Tor.
+    /// Send request over Tor.
     /// * local tx id
     /// * receiver
     SendTor(u32, SlatepackAddress),

src/wallet/wallet.rs

@@ -420,6 +420,19 @@ impl Wallet {
         w_config.save();
     }
 
+    /// Get transaction broadcasting delay in blocks.
+    pub fn broadcasting_delay(&self) -> u64 {
+        let r_config = self.config.read();
+        r_config.tx_broadcast_timeout.unwrap_or(WalletConfig::BROADCASTING_TIMEOUT_DEFAULT)
+    }
+
+    /// Update transaction broadcasting delay in blocks.
+    pub fn update_broadcasting_delay(&self, delay: u64) {
+        let mut w_config = self.config.write();
+        w_config.tx_broadcast_timeout = Some(delay);
+        w_config.save();
+    }
+
     /// Update external connection identifier.
     pub fn update_connection(&self, conn: &ConnectionMethod) {
         let mut w_config = self.config.write();
@@ -991,30 +1004,29 @@ impl Wallet {
         res
     }
 
-    /// Get possible transaction confirmation height from db or node.
-    fn tx_height(&self, tx: &TxLogEntry, store: &TxHeightStore) -> Result<Option<u64>, Error> {
+    /// Get possible transaction confirmation height, .
+    fn tx_height(&self, tx: &WalletTransaction) -> Result<Option<u64>, Error> {
         let mut tx_height = None;
-        if tx.kernel_lookup_min_height.is_some() && tx.kernel_excess.is_some() && tx.confirmed {
-            if let Some(height) = store.read_tx_height(tx.id) {
-                tx_height = Some(height);
-            } else {
-                let r_inst = self.instance.as_ref().read();
-                let instance = r_inst.clone().unwrap();
-                let mut w_lock = instance.lock();
-                let w = w_lock.lc_provider()?.wallet_inst()?;
-                if let Ok(res) = w.w2n_client().get_kernel(
-                    tx.kernel_excess.as_ref().unwrap(),
-                    tx.kernel_lookup_min_height,
-                    None
-                ) {
-                    if let Some((_, h, _)) = res {
-                        tx_height = Some(h);
-                        store.write_tx_height(tx.id, h);
-                    } else {
-                        tx_height = Some(0);
-                    }
-                }
+        if tx.data.confirmed && tx.data.kernel_excess.is_some() {
+            let r_inst = self.instance.as_ref().read();
+            let instance = r_inst.clone().unwrap();
+            let mut w_lock = instance.lock();
+            let w = w_lock.lc_provider()?.wallet_inst()?;
+            if let Ok(res) = w.w2n_client().get_kernel(
+                tx.data.kernel_excess.as_ref().unwrap(),
+                tx.data.kernel_lookup_min_height,
+                None
+            ) {
+                tx_height = Some(match res {
+                    None => 0,
+                    Some((_, h, _)) => h
+                });
             }
+        } else if tx.broadcasting() {
+            tx_height = match self.get_data() {
+                None => None,
+                Some(data) => Some(data.info.last_confirmed_height)
+            };
         }
         Ok(tx_height)
     }
@@ -1117,10 +1129,8 @@ impl Wallet {
 
 /// Delay in seconds to sync [`WalletData`] (60 seconds as average block time).
 const SYNC_DELAY: Duration = Duration::from_millis(60 * 1000);
-
 /// Delay in seconds for sync thread to wait before start of new attempt.
 const ATTEMPT_DELAY: Duration = Duration::from_millis(3 * 1000);
-
 /// Number of attempts to sync [`WalletData`] before setting an error.
 const SYNC_ATTEMPTS: u8 = 10;
 
@@ -1328,10 +1338,9 @@ async fn handle_task(w: &Wallet, t: WalletTask) {
             w.invoice_creating.store(false, Ordering::Relaxed);
         },
         WalletTask::Finalize(s, id) => {
-            let slate = if let Some(s) = s {
-                s
-            } else {
-                &w.get_tx(*id).unwrap()
+            let slate = match s {
+                None => &w.get_tx(*id).unwrap(),
+                Some(s) => s
             };
             match w.finalize(slate) {
                 Ok(s) => {
@@ -1350,10 +1359,9 @@ async fn handle_task(w: &Wallet, t: WalletTask) {
             }
         }
         WalletTask::Post(s, id) => {
-            let slate = if let Some(s) = s {
-                s
-            } else {
-                &w.get_tx(*id).unwrap()
+            let slate = match s {
+                None => &w.get_tx(*id).unwrap(),
+                Some(s) => s
             };
             match w.post(slate) {
                 Ok(_) => {
@@ -1501,81 +1509,62 @@ fn update_txs(wallet: &Wallet, instance: WalletInstance, info: WalletInfo)
         }
     }).collect::<Vec<TxLogEntry>>();
 
-    // Initialize tx confirmation height storage.
     let tx_height_store = TxHeightStore::new(wallet.get_config().get_extra_db_path());
-
     let data = wallet.get_data().unwrap();
     let data_txs = data.txs.unwrap_or(vec![]);
     let mut new_txs: Vec<WalletTransaction> = vec![];
     for tx in &account_txs {
-        // Setup transaction amount.
-        let amount = if tx.amount_debited > tx.amount_credited {
-            tx.amount_debited - tx.amount_credited
-        } else {
-            tx.amount_credited - tx.amount_debited
-        };
-
-        // Setup confirmation height, action and state.
         let mut height: Option<u64> = None;
+        let mut broadcasting_height: Option<u64> = None;
         let mut action: Option<WalletTransactionAction> = None;
         let mut action_error: Option<Error> = None;
-        let mut state: Option<SlateState> = None;
         for t in &data_txs {
             if t.data.id == tx.id {
-                height = t.height;
                 action = t.action.clone();
                 action_error = t.action_error.clone();
-                state = Some(t.state.clone());
+                height = t.height;
+                broadcasting_height = t.broadcasting_height;
                 break;
             }
         }
-        if tx.kernel_lookup_min_height.is_some() &&
-            tx.kernel_excess.is_some() && tx.confirmed {
-            if height.is_none() {
-                height = wallet.tx_height(tx, &tx_height_store).unwrap_or(None);
-            }
-        }
-
-        // Setup transaction state for unconfirmed tx or for initial update.
+        let mut new = WalletTransaction::new(tx.clone(),
+                                             wallet,
+                                             height,
+                                             broadcasting_height,
+                                             action,
+                                             action_error);
+        // Update Slate state for unconfirmed.
         let unconfirmed = !tx.confirmed && (tx.tx_type == TxLogEntryType::TxSent ||
             tx.tx_type == TxLogEntryType::TxReceived);
-        if unconfirmed || state.is_none() {
-            let mut slate = Slate::blank(1, false);
-            slate.id = tx.tx_slate_id.unwrap();
-            slate.state = match tx.tx_type {
-                TxLogEntryType::TxReceived => SlateState::Invoice3,
-                _ => SlateState::Standard3
-            };
-            // Transaction was finalized.
-            if wallet.slatepack_exists(&slate) {
-                state = Some(slate.state);
-            } else {
-                slate.id = tx.tx_slate_id.unwrap();
-                slate.state = match tx.tx_type {
-                    TxLogEntryType::TxReceived => SlateState::Standard2,
-                    _ => SlateState::Invoice2
-                };
-                // Transaction signed to be finalized.
-                if wallet.slatepack_exists(&slate) {
-                    state = Some(slate.state);
-                } else {
-                    // Transaction just was created.
-                    state = Some(match tx.tx_type {
-                        TxLogEntryType::TxReceived => SlateState::Invoice1,
-                        _ => SlateState::Standard1
-                    });
-                }
-            }
+        if unconfirmed {
+            new.update_slate_state(wallet);
         }
-        // Add transaction to the list.
-        new_txs.push(WalletTransaction {
-            data: tx.clone(),
-            state: state.unwrap(),
-            amount,
-            height,
-            action,
-            action_error,
-        });
+
+        // Setup initial tx heights.
+        if height.is_none() && tx.confirmed {
+            height = if let Some(height) = tx_height_store.read_tx_height(tx.id) {
+                Some(height)
+            } else {
+                tx_height_store.delete_broadcasting_height(tx.id);
+                let h = wallet.tx_height(&new)?;
+                if let Some(h) = h {
+                    tx_height_store.write_tx_height(tx.id, h);
+                }
+                h
+            };
+            new.height = height;
+        } else if broadcasting_height.is_none() && new.broadcasting() {
+            broadcasting_height = if let Some(h) = tx_height_store.read_broadcasting_height(tx.id) {
+                Some(h)
+            } else {
+                let h = data.info.last_confirmed_height;
+                tx_height_store.write_broadcasting_height(tx.id, h);
+                Some(h)
+            };
+            new.broadcasting_height = Some(broadcasting_height.unwrap_or(0));
+        }
+
+        new_txs.push(new);
     }
     // Update wallet txs.
     let mut w_data = wallet.data.write();