Add broadcastting with RPC
MODIFY
WalletWasabi.Gui/Controls/WalletExplorer/SendTabViewModel.cs
WalletWasabi.Gui/Controls/WalletExplorer/SendTabViewModel.cs
@@ -396,7 +396,7 @@ namespace WalletWasabi.Gui.Controls.WalletExplorer
}
MainWindowViewModel.Instance.StatusBar.TryAddStatus(StatusBarStatus.BroadcastingTransaction);
- await Task.Run(async () => await Global.WalletService.SendTransactionAsync(signedTransaction));
+ await Task.Run(async () => await Global.TransactionBroadcaster.SendTransactionAsync(signedTransaction));
TryResetInputsOnSuccess("Transaction is successfully sent!");

MODIFY
WalletWasabi.Gui/Controls/WalletExplorer/TransactionBroadcasterViewModel.cs
WalletWasabi.Gui/Controls/WalletExplorer/TransactionBroadcasterViewModel.cs
@@ -196,7 +196,7 @@ namespace WalletWasabi.Gui.Controls.WalletExplorer
}
MainWindowViewModel.Instance.StatusBar.TryAddStatus(StatusBarStatus.BroadcastingTransaction);
- await Task.Run(async () => await Global.WalletService.SendTransactionAsync(transaction));
+ await Task.Run(async () => await Global.TransactionBroadcaster.SendTransactionAsync(transaction));
SetSuccessMessage("Transaction is successfully sent!");
TransactionString = "";

MODIFY
WalletWasabi.Gui/Global.cs
WalletWasabi.Gui/Global.cs
@@ -34,6 +34,7 @@ using WalletWasabi.Mempool;
using WalletWasabi.Services;
using WalletWasabi.Stores;
using WalletWasabi.TorSocks5;
+using WalletWasabi.Transactions.TransactionBroadcasting;
namespace WalletWasabi.Gui
{
@@ -63,6 +64,7 @@ namespace WalletWasabi.Gui
public FeeProviders FeeProviders { get; private set; }
public CoinJoinClient ChaumianClient { get; private set; }
public WalletService WalletService { get; private set; }
+ public TransactionBroadcaster TransactionBroadcaster { get; set; }
public Node RegTestMempoolServingNode { get; private set; }
public UpdateChecker UpdateChecker { get; private set; }
public TorProcessManager TorManager { get; private set; }
@@ -340,6 +342,8 @@ namespace WalletWasabi.Gui
#endregion SynchronizerInitialization
+ TransactionBroadcaster = new TransactionBroadcaster(Network, BitcoinStore, Synchronizer, Nodes, BitcoinCoreNode?.RpcClient);
+
Initialized = true;
}
@@ -471,6 +475,8 @@ namespace WalletWasabi.Gui
token.ThrowIfCancellationRequested();
WalletService.TransactionProcessor.CoinReceived += CoinReceived;
+
+ TransactionBroadcaster.AddWalletService(WalletService);
}
_cancelWalletServiceInitialization = null; // Must make it null explicitly, because dispose won't make it null.
}
@@ -591,10 +597,7 @@ namespace WalletWasabi.Gui
WalletService.KeyManager?.ToFile(backupWalletFilePath);
Logger.LogInfo($"{nameof(KeyManager)} backup saved to `{backupWalletFilePath}`.");
}
- if (WalletService != null)
- {
- await WalletService.StopAsync();
- }
+ WalletService?.Dispose();
WalletService = null;
Logger.LogInfo($"{nameof(WalletService)} is stopped.");
}

MODIFY
WalletWasabi.Tests/IntegrationTests/P2pTests.cs
WalletWasabi.Tests/IntegrationTests/P2pTests.cs
@@ -143,10 +143,7 @@ namespace WalletWasabi.Tests.IntegrationTests
{
await walletService?.DeleteBlockAsync(hash);
}
- if (walletService != null)
- {
- await walletService.StopAsync();
- }
+ walletService?.Dispose();
if (Directory.Exists(blocksFolderPath))
{

MODIFY
WalletWasabi.Tests/IntegrationTests/RegTests.cs
WalletWasabi.Tests/IntegrationTests/RegTests.cs
@@ -37,6 +37,7 @@ using WalletWasabi.Services;
using WalletWasabi.Stores;
using WalletWasabi.Tests.XunitConfiguration;
using WalletWasabi.TorSocks5;
+using WalletWasabi.Transactions.TransactionBroadcasting;
using WalletWasabi.Transactions.TransactionBuilding;
using WalletWasabi.WebClients.Wasabi;
using Xunit;
@@ -621,10 +622,7 @@ namespace WalletWasabi.Tests.IntegrationTests
finally
{
wallet.NewFilterProcessed -= Wallet_NewFilterProcessed;
- if (wallet != null)
- {
- await wallet.StopAsync();
- }
+ wallet?.Dispose();
// Dispose wasabi synchronizer service.
await synchronizer?.StopAsync();
@@ -718,6 +716,8 @@ namespace WalletWasabi.Tests.IntegrationTests
{
await wallet.InitializeAsync(cts.Token); // Initialize wallet service.
}
+ var broadcaster = new TransactionBroadcaster(network, bitcoinStore, synchronizer, nodes, rpc);
+ broadcaster.AddWalletService(wallet);
var waitCount = 0;
while (wallet.Coins.Sum(x => x.Amount) == Money.Zero)
@@ -746,7 +746,7 @@ namespace WalletWasabi.Tests.IntegrationTests
Assert.Equal(Money.Coins(1m), res2.SpentCoins.Single().Amount);
Assert.False(res2.SpendsUnconfirmed);
- await wallet.SendTransactionAsync(res2.Transaction);
+ await broadcaster.SendTransactionAsync(res2.Transaction);
Assert.Contains(res2.InnerWalletOutputs.Single(), wallet.Coins);
@@ -794,7 +794,7 @@ namespace WalletWasabi.Tests.IntegrationTests
}
Assert.True(foundReceive);
- await wallet.SendTransactionAsync(res.Transaction);
+ await broadcaster.SendTransactionAsync(res.Transaction);
#endregion Basic
@@ -933,7 +933,7 @@ namespace WalletWasabi.Tests.IntegrationTests
Assert.InRange(res.Fee, Money.Zero, res.Fee);
Assert.InRange(res.Fee, res.Fee, res.Fee);
- await wallet.SendTransactionAsync(res.Transaction);
+ await broadcaster.SendTransactionAsync(res.Transaction);
#endregion HighFee
@@ -954,7 +954,7 @@ namespace WalletWasabi.Tests.IntegrationTests
Assert.Equal(receive, maxBuiltTxOutput.ScriptPubKey);
Assert.Equal(wallet.Coins.Where(x => !x.Unavailable).Sum(x => x.Amount) - res.Fee, maxBuiltTxOutput.Value);
- await wallet.SendTransactionAsync(res.Transaction);
+ await broadcaster.SendTransactionAsync(res.Transaction);
#endregion MaxAmount
@@ -1021,7 +1021,7 @@ namespace WalletWasabi.Tests.IntegrationTests
Assert.Contains("outgoing", change);
Assert.Contains("outgoing2", change);
- await wallet.SendTransactionAsync(res.Transaction);
+ await broadcaster.SendTransactionAsync(res.Transaction);
IEnumerable<string> unconfirmedCoinLabels = wallet.Coins.Where(x => x.Height == Height.Mempool).SelectMany(x => x.Label.Labels);
Assert.Contains("outgoing", unconfirmedCoinLabels);
@@ -1129,10 +1129,7 @@ namespace WalletWasabi.Tests.IntegrationTests
finally
{
wallet.NewFilterProcessed -= Wallet_NewFilterProcessed;
- if (wallet != null)
- {
- await wallet.StopAsync();
- }
+ wallet?.Dispose();
// Dispose wasabi synchronizer service.
await synchronizer?.StopAsync();
// Dispose connection service.
@@ -1298,10 +1295,7 @@ namespace WalletWasabi.Tests.IntegrationTests
}
finally
{
- if (wallet != null)
- {
- await wallet.StopAsync();
- }
+ wallet?.Dispose();
// Dispose wasabi synchronizer service.
await synchronizer?.StopAsync();
// Dispose connection service.
@@ -1374,15 +1368,17 @@ namespace WalletWasabi.Tests.IntegrationTests
await wallet.InitializeAsync(cts.Token); // Initialize wallet service.
}
Assert.Single(wallet.Coins);
+ var broadcaster = new TransactionBroadcaster(network, bitcoinStore, synchronizer, nodes, rpc);
+ broadcaster.AddWalletService(wallet);
// Send money before reorg.
var operations = new PaymentIntent(scp, Money.Coins(0.011m));
var btx1 = wallet.BuildTransaction(password, operations, FeeStrategy.TwentyMinutesConfirmationTargetStrategy);
- await wallet.SendTransactionAsync(btx1.Transaction);
+ await broadcaster.SendTransactionAsync(btx1.Transaction);
operations = new PaymentIntent(scp, Money.Coins(0.012m));
var btx2 = wallet.BuildTransaction(password, operations, FeeStrategy.TwentyMinutesConfirmationTargetStrategy, allowUnconfirmed: true);
- await wallet.SendTransactionAsync(btx2.Transaction);
+ await broadcaster.SendTransactionAsync(btx2.Transaction);
// Test synchronization after fork.
// Invalidate the blocks containing the funding transaction
@@ -1401,11 +1397,11 @@ namespace WalletWasabi.Tests.IntegrationTests
// are reintroduced when we generate a new block through the rpc call
operations = new PaymentIntent(scp, Money.Coins(0.013m));
var btx3 = wallet.BuildTransaction(password, operations, FeeStrategy.TwentyMinutesConfirmationTargetStrategy);
- await wallet.SendTransactionAsync(btx3.Transaction);
+ await broadcaster.SendTransactionAsync(btx3.Transaction);
operations = new PaymentIntent(scp, Money.Coins(0.014m));
var btx4 = wallet.BuildTransaction(password, operations, FeeStrategy.TwentyMinutesConfirmationTargetStrategy, allowUnconfirmed: true);
- await wallet.SendTransactionAsync(btx4.Transaction);
+ await broadcaster.SendTransactionAsync(btx4.Transaction);
// Test synchronization after fork with different transactions.
// Create a fork that invalidates the blocks containing the funding transaction
@@ -1461,10 +1457,7 @@ namespace WalletWasabi.Tests.IntegrationTests
}
finally
{
- if (wallet != null)
- {
- await wallet.StopAsync();
- }
+ wallet?.Dispose();
// Dispose wasabi synchronizer service.
await synchronizer?.StopAsync();
// Dispose connection service.
@@ -1542,6 +1535,9 @@ namespace WalletWasabi.Tests.IntegrationTests
Assert.Single(wallet.Coins);
+ var broadcaster = new TransactionBroadcaster(network, bitcoinStore, synchronizer, nodes, rpc);
+ broadcaster.AddWalletService(wallet);
+
var operations = new PaymentIntent(
new DestinationRequest(key.P2wpkhScript, Money.Coins(0.01m)),
new DestinationRequest(new Key().ScriptPubKey, Money.Coins(0.01m)),
@@ -1553,7 +1549,7 @@ namespace WalletWasabi.Tests.IntegrationTests
// Spend the unconfirmed coin (send it to ourself)
operations = new PaymentIntent(key.PubKey.WitHash.ScriptPubKey, Money.Coins(0.5m));
tx1Res = wallet.BuildTransaction(password, operations, FeeStrategy.TwentyMinutesConfirmationTargetStrategy, allowUnconfirmed: true);
- await wallet.SendTransactionAsync(tx1Res.Transaction);
+ await broadcaster.SendTransactionAsync(tx1Res.Transaction);
while (wallet.Coins.Count() != 2)
{
@@ -1576,7 +1572,7 @@ namespace WalletWasabi.Tests.IntegrationTests
// Spend the unconfirmed and unspent coin (send it to ourself)
operations = new PaymentIntent(key.PubKey.WitHash.ScriptPubKey, Money.Coins(0.5m), subtractFee: true);
var tx2Res = wallet.BuildTransaction(password, operations, FeeStrategy.TwentyMinutesConfirmationTargetStrategy, allowUnconfirmed: true);
- await wallet.SendTransactionAsync(tx2Res.Transaction);
+ await broadcaster.SendTransactionAsync(tx2Res.Transaction);
while (wallet.Coins.Count() != 2)
{
@@ -1621,7 +1617,7 @@ namespace WalletWasabi.Tests.IntegrationTests
var coinCount = GetAllCoins().Count();
var to = wallet.GetReceiveKey("foo");
var res = wallet.BuildTransaction(password, new PaymentIntent(to.P2wpkhScript, Money.Coins(0.2345m), label: "bar"), FeeStrategy.TwentyMinutesConfirmationTargetStrategy, allowUnconfirmed: true);
- await wallet.SendTransactionAsync(res.Transaction);
+ await broadcaster.SendTransactionAsync(res.Transaction);
Assert.Equal(coinCount + 2, GetAllCoins().Count());
Assert.Equal(2, GetAllCoins().Count(x => !x.Confirmed));
Interlocked.Exchange(ref _filtersProcessedByWalletCount, 0);
@@ -1632,10 +1628,7 @@ namespace WalletWasabi.Tests.IntegrationTests
}
finally
{
- if (wallet != null)
- {
- await wallet.StopAsync();
- }
+ wallet?.Dispose();
// Dispose wasabi synchronizer service.
await synchronizer?.StopAsync();
// Dispose connection service.
@@ -1741,10 +1734,7 @@ namespace WalletWasabi.Tests.IntegrationTests
}
finally
{
- if (wallet != null)
- {
- await wallet.StopAsync();
- }
+ wallet?.Dispose();
// Dispose wasabi synchronizer service.
await synchronizer?.StopAsync();
// Dispose connection service.
@@ -3399,10 +3389,7 @@ namespace WalletWasabi.Tests.IntegrationTests
finally
{
wallet.NewFilterProcessed -= Wallet_NewFilterProcessed;
- if (wallet != null)
- {
- await wallet.StopAsync();
- }
+ wallet?.Dispose();
// Dispose connection service.
nodes?.Dispose();
// Dispose mempool serving node.
@@ -3412,10 +3399,7 @@ namespace WalletWasabi.Tests.IntegrationTests
{
await chaumianClient.StopAsync();
}
- if (wallet2 != null)
- {
- await wallet2.StopAsync();
- }
+ wallet2?.Dispose();
// Dispose wasabi synchronizer service.
await synchronizer?.StopAsync();
// Dispose connection service.

MODIFY
WalletWasabi/Services/WalletService.cs
WalletWasabi/Services/WalletService.cs
@@ -29,7 +29,7 @@ using WalletWasabi.WebClients.Wasabi;
namespace WalletWasabi.Services
{
- public class WalletService
+ public class WalletService : IDisposable
{
public static event EventHandler<bool> DownloadingBlockChanged;
@@ -163,13 +163,11 @@ namespace WalletWasabi.Services
}
}
- private static object TransactionProcessingLock { get; } = new object();
-
private void Mempool_TransactionReceived(object sender, SmartTransaction tx)
{
try
{
- lock (TransactionProcessingLock)
+ lock (TransactionProcessor.Lock)
{
TransactionProcessor.Process(tx);
}
@@ -297,7 +295,7 @@ namespace WalletWasabi.Services
var mempoolHashes = await client.GetMempoolHashesAsync(compactness);
- lock (TransactionProcessingLock)
+ lock (TransactionProcessor.Lock)
{
foreach (var tx in BitcoinStore.TransactionStore.MempoolStore.GetTransactions())
{
@@ -369,7 +367,7 @@ namespace WalletWasabi.Services
private void ProcessBlock(Height height, Block block, IEnumerable<int> filterByTxIndexes = null, IEnumerable<SmartTransaction> skeletonBlock = null)
{
- lock (TransactionProcessingLock)
+ lock (TransactionProcessor.Lock)
{
if (filterByTxIndexes is null)
{
@@ -764,128 +762,6 @@ namespace WalletWasabi.Services
}
}
- private static long SendCount = 0;
-
- private async Task BroadcastTransactionToNetworkNodeAsync(SmartTransaction transaction, Node node)
- {
- Logger.LogInfo($"Trying to broadcast transaction with random node ({node.RemoteSocketAddress}):{transaction.GetHash()}.");
- if (!BitcoinStore.MempoolService.TryAddToBroadcastStore(transaction.Transaction, node.RemoteSocketEndpoint.ToString())) // So we'll reply to INV with this transaction.
- {
- Logger.LogWarning($"Transaction {transaction.GetHash()} was already present in the broadcast store.");
- }
- var invPayload = new InvPayload(transaction.Transaction);
- // Give 7 seconds to send the inv payload.
- await node.SendMessageAsync(invPayload).WithAwaitCancellationAsync(TimeSpan.FromSeconds(7)); // ToDo: It's dangerous way to cancel. Implement proper cancellation to NBitcoin!
-
- if (BitcoinStore.MempoolService.TryGetFromBroadcastStore(transaction.GetHash(), out TransactionBroadcastEntry entry))
- {
- // Give 7 seconds for serving.
- var timeout = 0;
- while (!entry.IsBroadcasted())
- {
- if (timeout > 7)
- {
- throw new TimeoutException("Did not serve the transaction.");
- }
- await Task.Delay(1_000);
- timeout++;
- }
- node.DisconnectAsync("Thank you!");
- Logger.LogInfo($"Disconnected node: {node.RemoteSocketAddress}. Successfully broadcasted transaction: {transaction.GetHash()}.");
-
- // Give 21 seconds for propagation.
- timeout = 0;
- while (entry.GetPropagationConfirmations() < 2)
- {
- if (timeout > 21)
- {
- throw new TimeoutException("Did not serve the transaction.");
- }
- await Task.Delay(1_000);
- timeout++;
- }
- Logger.LogInfo($"Transaction is successfully propagated: {transaction.GetHash()}.");
- }
- else
- {
- Logger.LogWarning($"Expected transaction {transaction.GetHash()} was not found in the broadcast store.");
- }
- }
-
- private async Task BroadcastTransactionToBackendAsync(SmartTransaction transaction)
- {
- using (var client = new WasabiClient(Synchronizer.WasabiClient.TorClient.DestinationUriAction, Synchronizer.WasabiClient.TorClient.TorSocks5EndPoint))
- {
- try
- {
- await client.BroadcastAsync(transaction);
- }
- catch (HttpRequestException ex2) when (ex2.Message.Contains("bad-txns-inputs-missingorspent", StringComparison.InvariantCultureIgnoreCase)
- || ex2.Message.Contains("missing-inputs", StringComparison.InvariantCultureIgnoreCase)
- || ex2.Message.Contains("txn-mempool-conflict", StringComparison.InvariantCultureIgnoreCase))
- {
- if (transaction.Transaction.Inputs.Count == 1) // If we tried to only spend one coin, then we can mark it as spent. If there were more coins, then we do not know.
- {
- OutPoint input = transaction.Transaction.Inputs.First().PrevOut;
- SmartCoin coin = Coins.GetByOutPoint(input);
- if (coin != default)
- {
- coin.SpentAccordingToBackend = true;
- }
- }
- }
- }
-
- lock (TransactionProcessingLock)
- {
- TransactionProcessor.Process(new SmartTransaction(transaction.Transaction, Height.Mempool));
- }
-
- Logger.LogInfo($"Transaction is successfully broadcasted to backend: {transaction.GetHash()}.");
- }
-
- public async Task SendTransactionAsync(SmartTransaction transaction)
- {
- try
- {
- Interlocked.Increment(ref SendCount);
- // Broadcast to a random node.
- // Wait until it arrives to at least two other nodes.
- // If something's wrong, fall back broadcasting with backend.
-
- if (Network == Network.RegTest)
- {
- throw new InvalidOperationException($"Transaction broadcasting to nodes does not work in {Network.RegTest}.");
- }
-
- Node node = Nodes.ConnectedNodes.RandomElement();
- while (node == default(Node) || !node.IsConnected || Nodes.ConnectedNodes.Count < 5)
- {
- // As long as we are connected to at least 4 nodes, we can always try again.
- // 3 should be enough, but make it 5 so 2 nodes could disconnect the meantime.
- if (Nodes.ConnectedNodes.Count < 5)
- {
- throw new InvalidOperationException("We are not connected to enough nodes.");
- }
- await Task.Delay(100);
- node = Nodes.ConnectedNodes.RandomElement();
- }
- await BroadcastTransactionToNetworkNodeAsync(transaction, node);
- }
- catch (Exception ex)
- {
- Logger.LogInfo($"Random node could not broadcast transaction. Broadcasting with backend... Reason: {ex.Message}.");
- Logger.LogDebug(ex);
-
- await BroadcastTransactionToBackendAsync(transaction);
- }
- finally
- {
- BitcoinStore.MempoolService.TryRemoveFromBroadcastStore(transaction.GetHash(), out _); // Remove it just to be sure. Probably has been removed previously.
- Interlocked.Decrement(ref SendCount);
- }
- }
-
public ISet<string> GetLabels() => TransactionProcessor.Coins.AsAllCoinsView()
.SelectMany(x => x.Label.Labels)
.Concat(KeyManager
@@ -942,13 +818,18 @@ namespace WalletWasabi.Services
Logger.LogInfo($"Current timeout value used on block download is: {timeout} seconds.");
}
- public async Task StopAsync()
+ #region IDisposable Support
+
+ private volatile bool _disposedValue = false; // To detect redundant calls
+
+ public bool IsDisposed => _disposedValue;
+
+ protected virtual void Dispose(bool disposing)
{
- while (Interlocked.Read(ref SendCount) != 0) // Make sure to wait for send to finish.
+ if (!_disposedValue)
+ {
+ if (disposing)
{
- await Task.Delay(50);
- }
-
BitcoinStore.IndexStore.NewFilter -= IndexDownloader_NewFilterAsync;
BitcoinStore.IndexStore.Reorged -= IndexDownloader_ReorgedAsync;
BitcoinStore.MempoolService.TransactionReceived -= Mempool_TransactionReceived;
@@ -957,5 +838,18 @@ namespace WalletWasabi.Services
DisconnectDisposeNullLocalBitcoinCoreNode();
}
+
+ _disposedValue = true;
+ }
+ }
+
+ // This code added to correctly implement the disposable pattern.
+ public void Dispose()
+ {
+ // Do not change this code. Put cleanup code in Dispose(bool disposing) above.
+ Dispose(true);
+ }
+
+ #endregion IDisposable Support
}
}

ADD
WalletWasabi/Transactions/TransactionBroadcasting/TransactionBroadcaster.cs
+using NBitcoin;
+using NBitcoin.Protocol;
+using NBitcoin.RPC;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Net.Http;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using WalletWasabi.Coins;
+using WalletWasabi.Helpers;
+using WalletWasabi.Logging;
+using WalletWasabi.Models;
+using WalletWasabi.Services;
+using WalletWasabi.Stores;
+using WalletWasabi.WebClients.Wasabi;
+
+namespace WalletWasabi.Transactions.TransactionBroadcasting
+{
+ public class TransactionBroadcaster
+ {
+ public BitcoinStore BitcoinStore { get; }
+ public WasabiSynchronizer Synchronizer { get; }
+ public Network Network { get; }
+ public NodesGroup Nodes { get; }
+ public List<WalletService> WalletServices { get; }
+ public IEnumerable<WalletService> AliveWalletServices => WalletServices.Where(x => x != null && !x.IsDisposed);
+ public object WalletServicesLock { get; }
+ public RPCClient RpcClient { get; private set; }
+
+ public TransactionBroadcaster(Network network, BitcoinStore bitcoinStore, WasabiSynchronizer synchronizer, NodesGroup nodes, RPCClient rpc)
+ {
+ Nodes = Guard.NotNull(nameof(nodes), nodes);
+ Network = Guard.NotNull(nameof(network), network);
+ BitcoinStore = Guard.NotNull(nameof(bitcoinStore), bitcoinStore);
+ Synchronizer = Guard.NotNull(nameof(synchronizer), synchronizer);
+ WalletServices = new List<WalletService>();
+ WalletServicesLock = new object();
+ RpcClient = rpc;
+ }
+
+ public void AddWalletService(WalletService walletService)
+ {
+ Guard.NotNull(nameof(walletService), walletService);
+ lock (WalletServicesLock)
+ {
+ WalletServices.Add(walletService);
+ }
+ }
+
+ private async Task BroadcastTransactionToNetworkNodeAsync(SmartTransaction transaction, Node node)
+ {
+ Logger.LogInfo($"Trying to broadcast transaction with random node ({node.RemoteSocketAddress}):{transaction.GetHash()}.");
+ if (!BitcoinStore.MempoolService.TryAddToBroadcastStore(transaction.Transaction, node.RemoteSocketEndpoint.ToString())) // So we'll reply to INV with this transaction.
+ {
+ Logger.LogWarning($"Transaction {transaction.GetHash()} was already present in the broadcast store.");
+ }
+ var invPayload = new InvPayload(transaction.Transaction);
+ // Give 7 seconds to send the inv payload.
+ await node.SendMessageAsync(invPayload).WithAwaitCancellationAsync(TimeSpan.FromSeconds(7)).ConfigureAwait(false); // ToDo: It's dangerous way to cancel. Implement proper cancellation to NBitcoin!
+
+ if (BitcoinStore.MempoolService.TryGetFromBroadcastStore(transaction.GetHash(), out TransactionBroadcastEntry entry))
+ {
+ // Give 7 seconds for serving.
+ var timeout = 0;
+ while (!entry.IsBroadcasted())
+ {
+ if (timeout > 7)
+ {
+ throw new TimeoutException("Did not serve the transaction.");
+ }
+ await Task.Delay(1_000).ConfigureAwait(false);
+ timeout++;
+ }
+ node.DisconnectAsync("Thank you!");
+ Logger.LogInfo($"Disconnected node: {node.RemoteSocketAddress}. Successfully broadcasted transaction: {transaction.GetHash()}.");
+
+ // Give 21 seconds for propagation.
+ timeout = 0;
+ while (entry.GetPropagationConfirmations() < 2)
+ {
+ if (timeout > 21)
+ {
+ throw new TimeoutException("Did not serve the transaction.");
+ }
+ await Task.Delay(1_000).ConfigureAwait(false);
+ timeout++;
+ }
+ Logger.LogInfo($"Transaction is successfully propagated: {transaction.GetHash()}.");
+ }
+ else
+ {
+ Logger.LogWarning($"Expected transaction {transaction.GetHash()} was not found in the broadcast store.");
+ }
+ }
+
+ private async Task BroadcastTransactionToBackendAsync(SmartTransaction transaction)
+ {
+ Logger.LogInfo("Broadcasting with backend...");
+ using (var client = new WasabiClient(Synchronizer.WasabiClient.TorClient.DestinationUriAction, Synchronizer.WasabiClient.TorClient.TorSocks5EndPoint))
+ {
+ try
+ {
+ await client.BroadcastAsync(transaction).ConfigureAwait(false);
+ }
+ catch (HttpRequestException ex2) when (ex2.Message.Contains("bad-txns-inputs-missingorspent", StringComparison.InvariantCultureIgnoreCase)
+ || ex2.Message.Contains("missing-inputs", StringComparison.InvariantCultureIgnoreCase)
+ || ex2.Message.Contains("txn-mempool-conflict", StringComparison.InvariantCultureIgnoreCase))
+ {
+ if (transaction.Transaction.Inputs.Count == 1) // If we tried to only spend one coin, then we can mark it as spent. If there were more coins, then we do not know.
+ {
+ OutPoint input = transaction.Transaction.Inputs.First().PrevOut;
+ lock (WalletServicesLock)
+ {
+ foreach (var walletService in AliveWalletServices)
+ {
+ SmartCoin coin = walletService.Coins.GetByOutPoint(input);
+ if (coin != default)
+ {
+ coin.SpentAccordingToBackend = true;
+ }
+ }
+ }
+ }
+ }
+ }
+
+ BelieveTransaction(transaction);
+
+ Logger.LogInfo($"Transaction is successfully broadcasted to backend: {transaction.GetHash()}.");
+ }
+
+ private void BelieveTransaction(SmartTransaction transaction)
+ {
+ lock (WalletServicesLock)
+ lock (TransactionProcessor.Lock)
+ {
+ foreach (var walletService in AliveWalletServices)
+ {
+ walletService.TransactionProcessor.Process(new SmartTransaction(transaction.Transaction, Height.Mempool));
+ }
+ }
+ }
+
+ public async Task SendTransactionAsync(SmartTransaction transaction)
+ {
+ try
+ {
+ // Broadcast to a random node.
+ // Wait until it arrives to at least two other nodes.
+ // If something's wrong, fall back broadcasting with rpc, then backend.
+
+ if (Network == Network.RegTest)
+ {
+ throw new InvalidOperationException($"Transaction broadcasting to nodes does not work in {Network.RegTest}.");
+ }
+
+ Node node = Nodes.ConnectedNodes.RandomElement();
+ while (node == default(Node) || !node.IsConnected || Nodes.ConnectedNodes.Count < 5)
+ {
+ // As long as we are connected to at least 4 nodes, we can always try again.
+ // 3 should be enough, but make it 5 so 2 nodes could disconnect the meantime.
+ if (Nodes.ConnectedNodes.Count < 5)
+ {
+ throw new InvalidOperationException("We are not connected to enough nodes.");
+ }
+ await Task.Delay(100).ConfigureAwait(false);
+ node = Nodes.ConnectedNodes.RandomElement();
+ }
+ await BroadcastTransactionToNetworkNodeAsync(transaction, node).ConfigureAwait(false);
+ }
+ catch (Exception ex)
+ {
+ Logger.LogInfo($"Random node could not broadcast transaction. Reason: {ex.Message}.");
+ Logger.LogDebug(ex);
+
+ if (RpcClient != null)
+ {
+ try
+ {
+ await BroadcastTransactionWithRpcAsync(transaction).ConfigureAwait(false);
+ }
+ catch (Exception ex2)
+ {
+ Logger.LogInfo($"RPC could not broadcast transaction. Reason: {ex2.Message}.");
+ Logger.LogDebug(ex2);
+
+ await BroadcastTransactionToBackendAsync(transaction).ConfigureAwait(false);
+ }
+ }
+ else
+ {
+ await BroadcastTransactionToBackendAsync(transaction).ConfigureAwait(false);
+ }
+ }
+ finally
+ {
+ BitcoinStore.MempoolService.TryRemoveFromBroadcastStore(transaction.GetHash(), out _); // Remove it just to be sure. Probably has been removed previously.
+ }
+ }
+
+ private async Task BroadcastTransactionWithRpcAsync(SmartTransaction transaction)
+ {
+ await RpcClient.SendRawTransactionAsync(transaction.Transaction).ConfigureAwait(false);
+ BelieveTransaction(transaction);
+ Logger.LogInfo($"Transaction is successfully broadcasted with RPC: {transaction.GetHash()}.");
+ }
+ }
+}

MODIFY
WalletWasabi/Transactions/TransactionProcessor.cs
WalletWasabi/Transactions/TransactionProcessor.cs
@@ -11,6 +11,7 @@ namespace WalletWasabi.Transactions
{
public class TransactionProcessor
{
+ public static object Lock { get; } = new object();
public AllTransactionStore TransactionStore { get; }
public KeyManager KeyManager { get; }

Add broadcastting with RPC