Implement transaction broadcasting through nodes.
MODIFY
WalletWasabi.Gui/Controls/WalletExplorer/HistoryTabViewModel.cs
WalletWasabi.Gui/Controls/WalletExplorer/HistoryTabViewModel.cs
@@ -125,20 +125,19 @@ namespace WalletWasabi.Gui.Controls.WalletExplorer
{
var found = txRecordList.FirstOrDefault(x => x.transactionId == coin.TransactionId);
- if (Global.WalletService is null) // disposed meanwhile
- {
- break;
- }
+ if (Global.WalletService is null) break; // Disposed meanwhile.
- SmartTransaction foundTransaction = Global.WalletService.TransactionCache.FirstOrDefault(x => x.GetHash() == coin.TransactionId);
+ SmartTransaction foundTransaction = Global.WalletService?.TransactionCache?.FirstOrDefault(x => x.GetHash() == coin.TransactionId);
if (foundTransaction is null)
{
+ if (Global.WalletService is null) break; // Disposed meanwhile.
continue;
}
DateTimeOffset dateTime;
if (foundTransaction.Height.Type == HeightType.Chain)
{
+ if (Global.WalletService is null) break; // Disposed meanwhile. Actually caught an NRE here.
if (Global.WalletService.ProcessedBlocks.Any(x => x.Value.height == foundTransaction.Height))
{
dateTime = Global.WalletService.ProcessedBlocks.First(x => x.Value.height == foundTransaction.Height).Value.dateTime;

MODIFY
WalletWasabi/Services/MemPoolBehavior.cs
WalletWasabi/Services/MemPoolBehavior.cs
@@ -35,9 +35,15 @@ namespace WalletWasabi.Services
{
try
{
+ if (message.Message.Payload is GetDataPayload getDataPayload)
+ {
+ await ProcessGetDataAsync(node, getDataPayload);
+ return;
+ }
+
if (message.Message.Payload is TxPayload txPayload)
{
- ProcessTxPayload(txPayload);
+ ProcessTx(txPayload);
return;
}
@@ -58,35 +64,92 @@ namespace WalletWasabi.Services
}
}
- private async Task ProcessInvAsync(Node node, InvPayload invPayload)
+ private async Task ProcessGetDataAsync(Node node, GetDataPayload payload)
{
- if (invPayload.Inventory.Count > MAX_INV_SIZE)
+ if (payload.Inventory.Count > MAX_INV_SIZE)
{
- Logger.LogDebug($"Received inventory too big. {nameof(MAX_INV_SIZE)}: {MAX_INV_SIZE}, Node: {node.RemoteSocketEndpoint}");
+ Logger.LogDebug<MemPoolBehavior>($"Received inventory too big. {nameof(MAX_INV_SIZE)}: {MAX_INV_SIZE}, Node: {node.RemoteSocketEndpoint}");
return;
}
- var payload = new GetDataPayload();
- foreach (var inv in invPayload.Inventory.Where(inv => inv.Type.HasFlag(InventoryType.MSG_TX)))
+ foreach (var inv in payload.Inventory.Where(inv => inv.Type.HasFlag(InventoryType.MSG_TX)))
+ {
+ if (MemPoolService.TryGetFromBroadcastStore(inv.Hash, out TransactionBroadcastEntry entry)) // If we have the transaction to be broadcasted then broadcast it now.
+ {
+ if (entry.NodeRemoteSocketEndpoint != node.RemoteSocketEndpoint.ToString())
+ {
+ continue; // Would be strange. It could be some kind of attack.
+ }
+
+ try
+ {
+ var txPayload = new TxPayload(entry.Transaction);
+ if (!node.IsConnected)
{
+ Logger.LogInfo<MemPoolBehavior>($"Couldn't serve transaction. Node ({node.RemoteSocketEndpoint}) is not connected anymore: {entry.TransactionId}.");
+ }
+ else
+ {
+ await node.SendMessageAsync(txPayload);
+ entry.MakeBroadcasted();
+ Logger.LogInfo<MemPoolBehavior>($"Successfully served transaction to node ({node.RemoteSocketEndpoint}): {entry.TransactionId}.");
+ }
+ }
+ catch (Exception ex)
+ {
+ Logger.LogInfo<MemPoolBehavior>(ex);
+ }
+ }
+ }
+ }
+
+ private async Task ProcessInvAsync(Node node, InvPayload payload)
+ {
+ if (payload.Inventory.Count > MAX_INV_SIZE)
+ {
+ Logger.LogDebug<MemPoolBehavior>($"Received inventory too big. {nameof(MAX_INV_SIZE)}: {MAX_INV_SIZE}, Node: {node.RemoteSocketEndpoint}");
+ return;
+ }
+
+ var getDataPayload = new GetDataPayload();
+ foreach (var inv in payload.Inventory.Where(inv => inv.Type.HasFlag(InventoryType.MSG_TX)))
+ {
+ if (MemPoolService.TryGetFromBroadcastStore(inv.Hash, out TransactionBroadcastEntry entry)) // If we have the transaction then adjust confirmation.
+ {
+ try
+ {
+ if (entry.NodeRemoteSocketEndpoint == node.RemoteSocketEndpoint.ToString())
+ {
+ continue; // Wtf, why are you trying to broadcast it back to us?
+ }
+
+ entry.ConfirmPropagationOnce();
+ }
+ catch (Exception ex)
+ {
+ Logger.LogInfo<MemPoolBehavior>(ex);
+ }
+ }
+
// if we already have it continue;
if (!MemPoolService.TransactionHashes.TryAdd(inv.Hash))
{
continue;
}
- payload.Inventory.Add(inv);
+ getDataPayload.Inventory.Add(inv);
}
- if (payload.Inventory.Any() && node.IsConnected) {
+ if (getDataPayload.Inventory.Any() && node.IsConnected)
+ {
// ask for the whole transaction
- await node.SendMessageAsync(payload);
+ await node.SendMessageAsync(getDataPayload);
}
}
- private void ProcessTxPayload(TxPayload transactionPayload)
+ private void ProcessTx(TxPayload payload)
{
- Transaction transaction = transactionPayload.Object;
+ Transaction transaction = payload.Object;
MemPoolService.OnTransactionReceived(new SmartTransaction(transaction, Height.MemPool));
}

MODIFY
WalletWasabi/Services/MemPoolService.cs
WalletWasabi/Services/MemPoolService.cs
@@ -16,6 +16,11 @@ namespace WalletWasabi.Services
{
public ConcurrentHashSet<uint256> TransactionHashes { get; }
+ // Transactions those we would reply to INV messages.
+ private List<TransactionBroadcastEntry> BroadcastStore { get; }
+
+ private object BroadcastStoreLock { get; }
+
public event EventHandler<SmartTransaction> TransactionReceived;
internal void OnTransactionReceived(SmartTransaction transaction) => TransactionReceived?.Invoke(this, transaction);
@@ -23,9 +28,72 @@ namespace WalletWasabi.Services
public MemPoolService()
{
TransactionHashes = new ConcurrentHashSet<uint256>();
+ BroadcastStore = new List<TransactionBroadcastEntry>();
+ BroadcastStoreLock = new object();
_cleanupInProcess = 0;
}
+ public bool TryAddToBroadcastStore(Transaction transaction, string nodeRemoteSocketEndpoint)
+ {
+ lock (BroadcastStoreLock)
+ {
+ if (BroadcastStore.Any(x => x.TransactionId == transaction.GetHash()))
+ {
+ return false;
+ }
+ else
+ {
+ var entry = new TransactionBroadcastEntry(transaction, nodeRemoteSocketEndpoint);
+ BroadcastStore.Add(entry);
+ return true;
+ }
+ }
+ }
+
+ public bool TryRemoveFromBroadcastStore(uint256 transactionHash, out TransactionBroadcastEntry entry)
+ {
+ lock (BroadcastStoreLock)
+ {
+ var found = BroadcastStore.FirstOrDefault(x => x.TransactionId == transactionHash);
+ entry = found;
+ if (found is null)
+ {
+ return false;
+ }
+ else
+ {
+ BroadcastStore.RemoveAll(x => x.TransactionId == transactionHash);
+ return true;
+ }
+ }
+ }
+
+ public bool TryGetFromBroadcastStore(uint256 transactionHash, out TransactionBroadcastEntry entry)
+ {
+ lock (BroadcastStoreLock)
+ {
+ var found = BroadcastStore.FirstOrDefault(x => x.TransactionId == transactionHash);
+ entry = found;
+
+ if (found is null)
+ {
+ return false;
+ }
+ else
+ {
+ return true;
+ }
+ }
+ }
+
+ public IEnumerable<TransactionBroadcastEntry> GetBroadcastStore()
+ {
+ lock (BroadcastStoreLock)
+ {
+ return BroadcastStore.ToList();
+ }
+ }
+
private int _cleanupInProcess;
/// <summary>

MODIFY
WalletWasabi/Services/WalletService.cs
WalletWasabi/Services/WalletService.cs
@@ -1243,13 +1243,85 @@ namespace WalletWasabi.Services
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 backend.
+
+ if (Network == Network.RegTest) throw new InvalidOperationException("Transaction broadcasting to nodes doesn't work in RegTest.");
+
+ while (true)
+ {
+ // 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.");
+
+ Node node = Nodes.ConnectedNodes.RandomElement();
+ if (node == default(Node))
+ {
+ await Task.Delay(100);
+ continue;
+ }
+
+ if (!node.IsConnected)
+ {
+ await Task.Delay(100);
+ continue;
+ }
+
+ Logger.LogInfo<WalletService>($"Trying to broadcast transaction with random node ({node.RemoteSocketAddress}):{transaction.GetHash()}");
+ MemPool.TryAddToBroadcastStore(transaction.Transaction, node.RemoteSocketEndpoint.ToString()); // So we'll reply to INV with this transaction.
+ var invPayload = new InvPayload(transaction.Transaction);
+ // Give 7 seconds to send the inv payload.
+ using (var cts = new CancellationTokenSource(TimeSpan.FromSeconds(7)))
+ {
+ await node.SendMessageAsync(invPayload).WithCancellation(cts.Token); // ToDo: It's dangerous way to cancel. Implement proper cancellation to NBitcoin!
+ }
+
+ // Give 7 seconds for serving.
+ var timeout = 0;
+ MemPool.TryGetFromBroadcastStore(transaction.GetHash(), out TransactionBroadcastEntry entry);
+ while (!entry.IsBroadcasted())
+ {
+ if (timeout > 7)
+ {
+ throw new TimeoutException("Didn't serve the transaction.");
+ }
+ await Task.Delay(1000);
+ timeout++;
+ }
+ node.DisconnectAsync("Thank you!");
+ Logger.LogInfo<MemPoolBehavior>($"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("Didn't serve the transaction.");
+ }
+ await Task.Delay(1000);
+ timeout++;
+ }
+ Logger.LogInfo<MemPoolBehavior>($"Transaction is successfully propagated: {transaction.GetHash()}.");
+
+ break;
+ }
+ }
+ catch (Exception ex)
+ {
+ Logger.LogInfo<WalletService>($"Random node couldn't broadcast transaction. Broadcasting with backend... Reason: {ex.Message}");
+ Logger.LogDebug<WalletService>(ex);
+
using (var client = new WasabiClient(Synchronizer.WasabiClient.TorClient.DestinationUriAction, Synchronizer.WasabiClient.TorClient.TorSocks5EndPoint))
{
try
{
await client.BroadcastAsync(transaction);
}
- catch (HttpRequestException ex) when (ex.Message.Contains("txn-mempool-conflict", StringComparison.InvariantCultureIgnoreCase))
+ catch (HttpRequestException ex2) when (ex2.Message.Contains("txn-mempool-conflict", StringComparison.InvariantCultureIgnoreCase))
{
if (transaction.Transaction.Inputs.Count == 1)
{
@@ -1260,14 +1332,23 @@ namespace WalletWasabi.Services
coin.SpentAccordingToBackend = true;
}
}
- throw;
+ throw new HttpRequestException("Coin has been already spent.");
+ }
+ catch (HttpRequestException ex2) when (ex2.Message.Contains("too-long-mempool-chain", StringComparison.InvariantCultureIgnoreCase))
+ {
+ throw new HttpRequestException("There are too many unconfirmed transactions. Please wait for confirmation.");
}
}
ProcessTransaction(new SmartTransaction(transaction.Transaction, Height.MemPool));
MemPool.TransactionHashes.TryAdd(transaction.GetHash());
- Logger.LogInfo<WalletService>($"Transaction is successfully broadcasted: {transaction.GetHash()}.");
+ Logger.LogInfo<WalletService>($"Transaction is successfully broadcasted to backend: {transaction.GetHash()}.");
+ }
+ finally
+ {
+ MemPool.TryRemoveFromBroadcastStore(transaction.GetHash(), out _); // Remove it just to be sure. Probably has been removed previously.
+ }
}
public IEnumerable<string> GetNonSpecialLabels()

Implement transaction broadcasting through nodes.