Implement transaction negotiation message resending in the wallet

Currently the logic for sending transactions and negotiating them between parties is not very robust. Each stage in the negotiation is sent only once. When sending we first try to send the message directly and if that fails, or discovery needs to occur, the message is sent indirectly via Store and Forward (SAF). We hoped to get SAF 100% reliable but due to reasons laid out in #2137 it is not tenable. 

This PR adds in the logic to periodically resends the initial Transaction Sender Message and the Transaction Reply in order to illicit the next step in the negotiation protocol from the counterparty. The PR also adds the logic to respond to repeated messages but with a cool down period in which repeats will be ignored to protect against DoS attacks. 

The PR also adds in a new comms level message to inform the counterparty if a transaction protocol has been cancelled. Initially this is sent if the Sender cancels an in progress transaction so that the receiver knows to stop expecting it. This message is also sent if a Transaction Reply is received for a cancelled message to let the Received know about the cancellation.

In order to stop a wallet resending these message indefinitely a transaction will be cancelled if it hasn’t resolved after a long timeout period (Default is being set to 3 days).

A large portion of the PR are tests so its not as big as it looks.

applications/tari_console_wallet/src/ui/app.rs

@@ -42,7 +42,7 @@ use tari_wallet::{
     contacts_service::storage::sqlite_db::ContactsServiceSqliteDatabase,
     output_manager_service::storage::sqlite_db::OutputManagerSqliteDatabase,
     storage::sqlite_db::WalletSqliteDatabase,
-    transaction_service::storage::{database::CompletedTransaction, sqlite_db::TransactionServiceSqliteDatabase},
+    transaction_service::storage::{models::CompletedTransaction, sqlite_db::TransactionServiceSqliteDatabase},
     Wallet,
 };
 use tokio::sync::RwLock;

applications/tari_console_wallet/src/ui/components/transactions_tab.rs

@@ -5,7 +5,7 @@ use crate::ui::{
     SelectedTransactionList,
     MAX_WIDTH,
 };
-use tari_wallet::transaction_service::storage::database::{TransactionDirection, TransactionStatus};
+use tari_wallet::transaction_service::storage::models::{TransactionDirection, TransactionStatus};
 use tui::{
     backend::Backend,
     layout::{Constraint, Direction, Layout, Rect},

applications/tari_console_wallet/src/ui/state/app_state.rs

@@ -6,7 +6,7 @@ use qrcode::{render::unicode, QrCode};
 use tari_common::Network;
 use tari_comms::NodeIdentity;
 use tari_crypto::tari_utilities::hex::Hex;
-use tari_wallet::{transaction_service::storage::database::CompletedTransaction, util::emoji::EmojiId};
+use tari_wallet::{transaction_service::storage::models::CompletedTransaction, util::emoji::EmojiId};
 
 pub struct AppState {
     pub pending_txs: StatefulList<CompletedTransaction>,

base_layer/core/src/proto/generated/tari.transaction_protocol.rs

@@ -8,6 +8,12 @@ pub struct TransactionFinalizedMessage {
     pub transaction: ::std::option::Option<super::types::Transaction>,
 }
 #[derive(Clone, PartialEq, ::prost::Message)]
+pub struct TransactionCancelledMessage {
+    /// The transaction id for the cancelled transaction
+    #[prost(uint64, tag = "1")]
+    pub tx_id: u64,
+}
+#[derive(Clone, PartialEq, ::prost::Message)]
 pub struct TransactionMetadata {
     /// The absolute fee for the transaction
     #[prost(uint64, tag = "1")]

base_layer/core/src/transactions/transaction_protocol/proto/transaction_cancelled.proto

@@ -0,0 +1,9 @@
+syntax = "proto3";
+
+package tari.transaction_protocol;
+
+message TransactionCancelledMessage {
+    // The transaction id for the cancelled transaction
+    uint64 tx_id = 1;
+}
+

base_layer/p2p/src/proto/message_type.proto

@@ -22,6 +22,7 @@ enum TariMessageType {
     TariMessageTypeMempoolRequest= 71;
     TariMessageTypeMempoolResponse = 72;
     TariMessageTypeTransactionFinalized = 73;
+    TariMessageTypeTransactionCancelled = 74;
     // -- DAN Messages --
 
     // -- Extended --

base_layer/p2p/src/proto/tari.p2p.message_type.rs

@@ -15,8 +15,9 @@ pub enum TariMessageType {
     BaseNodeResponse = 70,
     MempoolRequest = 71,
     MempoolResponse = 72,
-    /// -- DAN Messages --
     TransactionFinalized = 73,
+    /// -- DAN Messages --
+    TransactionCancelled = 74,
     // -- Extended --
     Text = 225,
     TextAck = 226,

base_layer/wallet/migrations/2020-08-17-141407_add_resend_count_and_timestamp_add_status_to_pending_txs/down.sql

@@ -0,0 +1,58 @@
+PRAGMA foreign_keys=off;
+ALTER TABLE completed_transactions RENAME TO completed_transactions_old;
+CREATE TABLE completed_transactions (
+    tx_id INTEGER PRIMARY KEY NOT NULL,
+    source_public_key BLOB NOT NULL,
+    destination_public_key BLOB NOT NULL,
+    amount INTEGER NOT NULL,
+    fee INTEGER NOT NULL,
+    transaction_protocol TEXT NOT NULL,
+    status INTEGER NOT NULL,
+    message TEXT NOT NULL,
+    timestamp DATETIME NOT NULL,
+    cancelled INTEGER NOT NULL DEFAULT 0,
+    direction INTEGER NULL DEFAULT NULL,
+    coinbase_block_height INTEGER NULL DEFAULT NULL
+);
+INSERT INTO completed_transactions (tx_id, source_public_key, destination_public_key, amount, fee, transaction_protocol, status, message, timestamp, cancelled, direction, coinbase_block_height)
+SELECT tx_id, source_public_key, destination_public_key, amount, fee, transaction_protocol, status, message, timestamp, cancelled, direction, coinbase_block_height
+FROM completed_transactions_old;
+
+DROP TABLE completed_transactions_old;
+
+ALTER TABLE inbound_transactions RENAME TO inbound_transactions_old;
+CREATE TABLE inbound_transactions (
+    tx_id INTEGER PRIMARY KEY NOT NULL,
+    source_public_key BLOB NOT NULL,
+    amount INTEGER NOT NULL,
+    receiver_protocol TEXT NOT NULL,
+    message TEXT NOT NULL,
+    timestamp DATETIME NOT NULL,
+    cancelled INTEGER NOT NULL DEFAULT 0,
+    direct_send_success INTEGER NOT NULL DEFAULT 0
+);
+INSERT INTO inbound_transactions (tx_id, source_public_key, amount, receiver_protocol, message, timestamp, cancelled, direct_send_success)
+SELECT tx_id, source_public_key, amount, receiver_protocol, message, timestamp, cancelled, direct_send_success
+FROM inbound_transactions_old;
+
+DROP TABLE inbound_transactions_old;
+
+ALTER TABLE outbound_transactions RENAME TO outbound_transactions_old;
+CREATE TABLE outbound_transactions (
+    tx_id INTEGER PRIMARY KEY NOT NULL,
+    destination_public_key BLOB NOT NULL,
+    amount INTEGER NOT NULL,
+    fee INTEGER NOT NULL,
+    sender_protocol TEXT NOT NULL,
+    message TEXT NOT NULL,
+    timestamp DATETIME NOT NULL,
+    cancelled INTEGER NOT NULL DEFAULT 0,
+    direct_send_success INTEGER NOT NULL DEFAULT 0
+);
+INSERT INTO outbound_transactions (tx_id, destination_public_key, amount, fee, sender_protocol, message, timestamp, cancelled, direct_send_success)
+SELECT tx_id, destination_public_key, amount, fee, sender_protocol, message, timestamp, cancelled, direct_send_success
+FROM outbound_transactions_old;
+
+DROP TABLE outbound_transactions_old;
+
+PRAGMA foreign_keys=on;

base_layer/wallet/migrations/2020-08-17-141407_add_resend_count_and_timestamp_add_status_to_pending_txs/up.sql

@@ -0,0 +1,14 @@
+ALTER TABLE completed_transactions
+    ADD COLUMN send_count INTEGER NOT NULL DEFAULT 0;
+ALTER TABLE completed_transactions
+    ADD COLUMN last_send_timestamp DATETIME NULL DEFAULT NULL;
+
+ALTER TABLE inbound_transactions
+    ADD COLUMN send_count INTEGER NOT NULL DEFAULT 0;
+ALTER TABLE inbound_transactions
+    ADD COLUMN last_send_timestamp DATETIME NULL DEFAULT NULL;
+
+ALTER TABLE outbound_transactions
+    ADD COLUMN send_count INTEGER NOT NULL DEFAULT 0;
+ALTER TABLE outbound_transactions
+    ADD COLUMN last_send_timestamp DATETIME NULL DEFAULT NULL;

base_layer/wallet/src/lib.rs

@@ -1,4 +1,4 @@
-#![recursion_limit = "1024"]
+#![recursion_limit = "2048"]
 #![feature(drain_filter)]
 #![feature(type_alias_impl_trait)]
 

base_layer/wallet/src/schema.rs

@@ -12,6 +12,8 @@ table! {
         cancelled -> Integer,
         direction -> Nullable<Integer>,
         coinbase_block_height -> Nullable<BigInt>,
+        send_count -> Integer,
+        last_send_timestamp -> Nullable<Timestamp>,
     }
 }
 
@@ -32,6 +34,8 @@ table! {
         timestamp -> Timestamp,
         cancelled -> Integer,
         direct_send_success -> Integer,
+        send_count -> Integer,
+        last_send_timestamp -> Nullable<Timestamp>,
     }
 }
 
@@ -56,6 +60,8 @@ table! {
         timestamp -> Timestamp,
         cancelled -> Integer,
         direct_send_success -> Integer,
+        send_count -> Integer,
+        last_send_timestamp -> Nullable<Timestamp>,
     }
 }
 

base_layer/wallet/src/testnet_utils.rs

@@ -33,8 +33,9 @@ use crate::{
     transaction_service::{
         handle::TransactionEvent,
         storage::{
-            database::{CompletedTransaction, TransactionBackend, TransactionDirection, TransactionStatus},
+            database::TransactionBackend,
             memory_db::TransactionMemoryDatabase,
+            models::{CompletedTransaction, TransactionDirection, TransactionStatus},
         },
     },
     wallet::WalletConfig,

base_layer/wallet/src/transaction_service/config.rs

@@ -31,8 +31,10 @@ pub struct TransactionServiceConfig {
     pub chain_monitoring_timeout: Duration,
     pub direct_send_timeout: Duration,
     pub broadcast_send_timeout: Duration,
-    pub low_power_polling_timeout: Duration, /* This is the timeout period that will be used when the wallet is in
-                                              * low_power mode */
+    pub low_power_polling_timeout: Duration,
+    pub transaction_resend_period: Duration,
+    pub resend_response_cooldown: Duration,
+    pub pending_transaction_cancellation_timeout: Duration,
 }
 
 impl Default for TransactionServiceConfig {
@@ -43,6 +45,9 @@ impl Default for TransactionServiceConfig {
             direct_send_timeout: Duration::from_secs(20),
             broadcast_send_timeout: Duration::from_secs(30),
             low_power_polling_timeout: Duration::from_secs(300),
+            transaction_resend_period: Duration::from_secs(3600),
+            resend_response_cooldown: Duration::from_secs(300),
+            pending_transaction_cancellation_timeout: Duration::from_secs(259200), // 3 Days
         }
     }
 }

base_layer/wallet/src/transaction_service/error.rs

@@ -115,12 +115,16 @@ pub enum TransactionServiceError {
     NodeIdError(#[from] NodeIdError),
     #[error("Broadcast recv error: `{0}`")]
     BroadcastRecvError(#[from] RecvError),
+    #[error("Broadcast send error: `{0}`")]
+    BroadcastSendError(String),
     #[error("Oneshot cancelled error: `{0}`")]
     OneshotCancelled(#[from] Canceled),
     #[error("Liveness error: `{0}`")]
     LivenessError(#[from] LivenessError),
     #[error("Coinbase build error: `{0}`")]
     CoinbaseBuildError(#[from] CoinbaseBuildError),
+    #[error("Pending Transaction Timed out")]
+    Timeout,
 }
 
 #[derive(Debug, Error)]

base_layer/wallet/src/transaction_service/handle.rs

@@ -24,7 +24,7 @@ use crate::{
     output_manager_service::TxId,
     transaction_service::{
         error::TransactionServiceError,
-        storage::database::{CompletedTransaction, InboundTransaction, OutboundTransaction},
+        storage::models::{CompletedTransaction, InboundTransaction, OutboundTransaction},
     },
 };
 use aes_gcm::Aes256Gcm;

base_layer/wallet/src/transaction_service/mod.rs

@@ -26,6 +26,7 @@ pub mod handle;
 pub mod protocols;
 pub mod service;
 pub mod storage;
+pub mod tasks;
 
 use crate::{
     output_manager_service::handle::OutputManagerHandle,
@@ -163,6 +164,19 @@ where T: TransactionBackend
             .map(map_decode::<BaseNodeProto::BaseNodeServiceResponse>)
             .filter_map(ok_or_skip_result)
     }
+
+    fn transaction_cancelled_stream(&self) -> impl Stream<Item = DomainMessage<proto::TransactionCancelledMessage>> {
+        trace!(
+            target: LOG_TARGET,
+            "Subscription '{}' for topic '{:?}' created.",
+            SUBSCRIPTION_LABEL,
+            TariMessageType::TransactionCancelled
+        );
+        self.subscription_factory
+            .get_subscription(TariMessageType::TransactionCancelled, SUBSCRIPTION_LABEL)
+            .map(map_decode::<proto::TransactionCancelledMessage>)
+            .filter_map(ok_or_skip_result)
+    }
 }
 
 impl<T> ServiceInitializer for TransactionServiceInitializer<T>
@@ -183,6 +197,7 @@ where T: TransactionBackend + Clone + 'static
         let transaction_finalized_stream = self.transaction_finalized_stream();
         let mempool_response_stream = self.mempool_response_stream();
         let base_node_response_stream = self.base_node_response_stream();
+        let transaction_cancelled_stream = self.transaction_cancelled_stream();
 
         let (publisher, _) = broadcast::channel(200);
 
@@ -219,6 +234,7 @@ where T: TransactionBackend + Clone + 'static
                 transaction_finalized_stream,
                 mempool_response_stream,
                 base_node_response_stream,
+                transaction_cancelled_stream,
                 output_manager_service,
                 outbound_message_service,
                 publisher,

base_layer/wallet/src/transaction_service/protocols/transaction_broadcast_protocol.rs

@@ -24,7 +24,7 @@ use crate::transaction_service::{
     error::{TransactionServiceError, TransactionServiceProtocolError},
     handle::TransactionEvent,
     service::TransactionServiceResources,
-    storage::database::{TransactionBackend, TransactionStatus},
+    storage::{database::TransactionBackend, models::TransactionStatus},
 };
 use futures::{channel::mpsc::Receiver, FutureExt, StreamExt};
 use log::*;

base_layer/wallet/src/transaction_service/protocols/transaction_chain_monitoring_protocol.rs

@@ -26,7 +26,7 @@ use crate::{
         error::{TransactionServiceError, TransactionServiceProtocolError},
         handle::TransactionEvent,
         service::TransactionServiceResources,
-        storage::database::{TransactionBackend, TransactionStatus},
+        storage::{database::TransactionBackend, models::TransactionStatus},
     },
 };
 use futures::{channel::mpsc::Receiver, FutureExt, StreamExt};

base_layer/wallet/src/transaction_service/protocols/transaction_coinbase_monitoring_protocol.rs

@@ -26,7 +26,7 @@ use crate::{
         error::{TransactionServiceError, TransactionServiceProtocolError},
         handle::TransactionEvent,
         service::TransactionServiceResources,
-        storage::database::{TransactionBackend, TransactionStatus},
+        storage::{database::TransactionBackend, models::TransactionStatus},
     },
 };
 use futures::{channel::mpsc::Receiver, FutureExt, StreamExt};