Implement local mempool receiving.
MODIFY
WalletWasabi.Backend/Global.cs
WalletWasabi.Backend/Global.cs
@@ -8,6 +8,7 @@ using System.Threading;
using System.Threading.Tasks;
using WalletWasabi.Blockchain.BlockFilters;
using WalletWasabi.Blockchain.Blocks;
+using WalletWasabi.Blockchain.Mempool;
using WalletWasabi.CoinJoin.Coordinator;
using WalletWasabi.CoinJoin.Coordinator.Rounds;
using WalletWasabi.Helpers;
@@ -137,7 +138,7 @@ namespace WalletWasabi.Backend
IsRelay = true
};
- nodeConnectionParameters.TemplateBehaviors.Add(new TrustedNodeNotifyingBehavior());
+ nodeConnectionParameters.TemplateBehaviors.Add(new TrustedNodeNotifyingBehavior(new MempoolService()));
var node = await Node.ConnectAsync(network, endPoint, nodeConnectionParameters);
// We have to find it, because it's cloned by the node and not perfectly cloned (event handlers cannot be cloned.)
BlockNotifier = new BlockNotifier(TimeSpan.FromSeconds(7), RpcClient, node.Behaviors.Find<TrustedNodeNotifyingBehavior>());

MODIFY
WalletWasabi.Gui/Global.cs
WalletWasabi.Gui/Global.cs
@@ -159,24 +159,6 @@ namespace WalletWasabi.Gui
BitcoinStore = new BitcoinStore();
var bstoreInitTask = BitcoinStore.InitializeAsync(Path.Combine(DataDir, "BitcoinStore"), Network);
var addressManagerFolderPath = Path.Combine(DataDir, "AddressManager");
- Task<CoreNode> bitcoinCoreNodeInitTask;
- if (Config.StartLocalBitcoinCoreOnStartup)
- {
- bitcoinCoreNodeInitTask = CoreNode
- .CreateAsync(new CoreNodeParams(
- Network,
- Config.LocalBitcoinCoreDataDir,
- tryRestart: false,
- tryDeleteDataDir: false,
- EndPointStrategy.Custom(Config.GetBitcoinP2pEndPoint()),
- EndPointStrategy.Default(Network, EndPointType.Rpc),
- txIndex: null,
- prune: null));
- }
- else
- {
- bitcoinCoreNodeInitTask = Task.FromResult<CoreNode>(null);
- }
AddressManagerFilePath = Path.Combine(addressManagerFolderPath, $"AddressManager{Network}.dat");
var addrManTask = InitializeAddressManagerBehaviorAsync();
@@ -247,9 +229,21 @@ namespace WalletWasabi.Gui
var feeProviderList = new List<IFeeProvider>();
try
{
- BitcoinCoreNode = await bitcoinCoreNodeInitTask.ConfigureAwait(false);
if (Config.StartLocalBitcoinCoreOnStartup)
{
+ BitcoinCoreNode = await CoreNode
+ .CreateAsync(new CoreNodeParams(
+ Network,
+ BitcoinStore.MempoolService,
+ Config.LocalBitcoinCoreDataDir,
+ tryRestart: false,
+ tryDeleteDataDir: false,
+ EndPointStrategy.Custom(Config.GetBitcoinP2pEndPoint()),
+ EndPointStrategy.Default(Network, EndPointType.Rpc),
+ txIndex: null,
+ prune: null))
+ .ConfigureAwait(false);
+
RpcMonitor.RpcClient = BitcoinCoreNode.RpcClient;
RpcMonitor.Start();
@@ -698,10 +692,15 @@ namespace WalletWasabi.Gui
Logger.LogInfo("Stopped monitoring RPC.");
}
- if (Config.StopLocalBitcoinCoreOnShutdown && BitcoinCoreNode != null)
+ if (BitcoinCoreNode != null)
+ {
+ await BitcoinCoreNode.DisposeAsync().ConfigureAwait(false);
+
+ if (Config.StopLocalBitcoinCoreOnShutdown)
{
await BitcoinCoreNode.TryStopAsync().ConfigureAwait(false);
}
+ }
if (TorManager != null)
{

MODIFY
WalletWasabi.Tests/Helpers/TestNodeBuilder.cs
WalletWasabi.Tests/Helpers/TestNodeBuilder.cs
@@ -7,15 +7,20 @@ using System.Text;
using System.Threading.Tasks;
using WalletWasabi.BitcoinCore;
using WalletWasabi.BitcoinCore.Endpointing;
+using WalletWasabi.Blockchain.Mempool;
using WalletWasabi.Helpers;
+using WalletWasabi.Stores;
namespace WalletWasabi.Tests.Helpers
{
public static class TestNodeBuilder
{
- public static async Task<CoreNode> CreateAsync([CallerFilePath]string callerFilePath = null, [CallerMemberName]string callerMemberName = null, string additionalFolder = null)
- => await CoreNode.CreateAsync(new CoreNodeParams(
- Network.RegTest,
+ public static async Task<CoreNode> CreateAsync([CallerFilePath]string callerFilePath = null, [CallerMemberName]string callerMemberName = null, string additionalFolder = null, MempoolService mempoolService = null)
+ {
+ var network = Network.RegTest;
+ return await CoreNode.CreateAsync(new CoreNodeParams(
+ network,
+ mempoolService ?? new MempoolService(),
Path.Combine(Global.Instance.DataDir, EnvironmentHelpers.ExtractFileName(callerFilePath), callerMemberName, additionalFolder ?? ""),
tryRestart: true,
tryDeleteDataDir: true,
@@ -25,3 +30,4 @@ namespace WalletWasabi.Tests.Helpers
prune: 0));
}
}
+}

MODIFY
WalletWasabi.Tests/UnitTests/BitcoinCore/P2pBasedTests.cs
WalletWasabi.Tests/UnitTests/BitcoinCore/P2pBasedTests.cs
@@ -25,14 +25,14 @@ namespace WalletWasabi.Tests.UnitTests.BitcoinCore
using var node = await coreNode.CreateNewP2pNodeAsync();
try
{
+ var network = coreNode.Network;
var rpc = coreNode.RpcClient;
- await rpc.GenerateAsync(101);
- var network = rpc.Network;
var bitcoinStore = new BitcoinStore();
-
var dir = Path.Combine(Global.Instance.DataDir, EnvironmentHelpers.GetMethodName());
await bitcoinStore.InitializeAsync(dir, network);
+ await rpc.GenerateAsync(101);
+
node.Behaviors.Add(bitcoinStore.CreateMempoolBehavior());
node.VersionHandshake();
@@ -81,17 +81,12 @@ namespace WalletWasabi.Tests.UnitTests.BitcoinCore
var dir = Path.Combine(Global.Instance.DataDir, EnvironmentHelpers.GetMethodName());
var addr = new Key().PubKey.GetSegwitAddress(network);
- var notifier = coreNode.TrustedNodeNotifyingBehavior;
+ var notifier = coreNode.MempoolService;
var txNum = 10;
- var txInvEventAwaiter = new EventsAwaiter<uint256>(
- h => notifier.TransactionInv += h,
- h => notifier.TransactionInv -= h,
- txNum);
-
var txEventAwaiter = new EventsAwaiter<SmartTransaction>(
- h => notifier.Transaction += h,
- h => notifier.Transaction -= h,
+ h => notifier.TransactionReceived += h,
+ h => notifier.TransactionReceived -= h,
txNum);
var txTasks = new List<Task<uint256>>();
@@ -102,16 +97,10 @@ namespace WalletWasabi.Tests.UnitTests.BitcoinCore
}
var batchTask = batch.SendBatchAsync();
- var aht = txInvEventAwaiter.WaitAsync(TimeSpan.FromSeconds(21));
var arrivedTxs = await txEventAwaiter.WaitAsync(TimeSpan.FromSeconds(21));
- var arrivedHashes = await aht;
await batchTask;
var hashes = await Task.WhenAll(txTasks);
- foreach (var hash in arrivedHashes)
- {
- Assert.Contains(hash, hashes);
- }
foreach (var hash in arrivedTxs.Select(x => x.GetHash()))
{
Assert.Contains(hash, hashes);

MODIFY
WalletWasabi/BitcoinCore/CoreNode.cs
WalletWasabi/BitcoinCore/CoreNode.cs
@@ -17,9 +17,11 @@ using WalletWasabi.BitcoinCore.Configuration;
using WalletWasabi.BitcoinCore.Endpointing;
using WalletWasabi.BitcoinCore.Processes;
using WalletWasabi.Blockchain.Blocks;
+using WalletWasabi.Blockchain.Mempool;
using WalletWasabi.Helpers;
using WalletWasabi.Logging;
using WalletWasabi.Services;
+using WalletWasabi.Stores;
namespace WalletWasabi.BitcoinCore
{
@@ -31,6 +33,7 @@ namespace WalletWasabi.BitcoinCore
private BitcoindRpcProcessBridge Bridge { get; set; }
public string DataDir { get; private set; }
public Network Network { get; private set; }
+ public MempoolService MempoolService { get; private set; }
public CoreConfig Config { get; private set; }
public TrustedNodeNotifyingBehavior TrustedNodeNotifyingBehavior => P2pNode.Behaviors.Find<TrustedNodeNotifyingBehavior>();
@@ -45,6 +48,7 @@ namespace WalletWasabi.BitcoinCore
var coreNode = new CoreNode();
coreNode.DataDir = coreNodeParams.DataDir;
coreNode.Network = coreNodeParams.Network;
+ coreNode.MempoolService = coreNodeParams.MempoolService;
var configPath = Path.Combine(coreNode.DataDir, "bitcoin.conf");
coreNode.Config = new CoreConfig();
@@ -194,9 +198,11 @@ namespace WalletWasabi.BitcoinCore
IsRelay = true
};
- nodeConnectionParameters.TemplateBehaviors.Add(new TrustedNodeNotifyingBehavior());
+ nodeConnectionParameters.TemplateBehaviors.Add(new TrustedNodeNotifyingBehavior(coreNode.MempoolService));
coreNode.P2pNode = await Node.ConnectAsync(coreNode.Network, coreNode.P2pEndPoint, nodeConnectionParameters).ConfigureAwait(false);
coreNode.P2pNode.VersionHandshake();
+ coreNode.P2pNode.StateChanged += coreNode.P2pNode_StateChanged;
+ coreNode.P2pNodeStateChangedSubscribed = true;
coreNode.BlockNotifier = new BlockNotifier(TimeSpan.FromSeconds(7), coreNode.RpcClient, coreNode.TrustedNodeNotifyingBehavior);
coreNode.BlockNotifier.Start();
@@ -205,6 +211,22 @@ namespace WalletWasabi.BitcoinCore
}
}
+ private bool P2pNodeStateChangedSubscribed { get; set; }
+
+ private void P2pNode_StateChanged(Node node, NodeState oldState)
+ {
+ if (node.IsConnected)
+ {
+ Logger.LogInfo("Local node got connected. Turned on trusted mempool mode.");
+ MempoolService.TrustedNodeMode = true;
+ }
+ else
+ {
+ Logger.LogInfo("Local node isn't connected. Turned off trusted mempool mode.");
+ MempoolService.TrustedNodeMode = false;
+ }
+ }
+
public static async Task<Version> GetVersionAsync(CancellationToken cancel)
{
var arguments = "-version";
@@ -235,14 +257,58 @@ namespace WalletWasabi.BitcoinCore
return await Task.WhenAll(tasks).ConfigureAwait(false);
}
- /// <param name="onlyOwned">Only stop if this node owns the process.</param>
- public async Task<bool> TryStopAsync(bool onlyOwned = true)
+ private volatile bool _disposedValue = false; // To detect redundant calls
+
+ public async Task DisposeAsync()
+ {
+ if (!_disposedValue)
{
if (BlockNotifier != null)
{
await BlockNotifier.StopAsync().ConfigureAwait(false);
}
- DisconnectDisposeNullP2pNode();
+
+ if (P2pNodeStateChangedSubscribed)
+ {
+ P2pNode.StateChanged -= P2pNode_StateChanged;
+ }
+
+ if (P2pNode != null)
+ {
+ try
+ {
+ P2pNode?.Disconnect();
+ }
+ catch (Exception ex)
+ {
+ Logger.LogDebug(ex);
+ }
+ finally
+ {
+ try
+ {
+ P2pNode?.Dispose();
+ }
+ catch (Exception ex)
+ {
+ Logger.LogDebug(ex);
+ }
+ finally
+ {
+ P2pNode = null;
+ Logger.LogInfo("P2p Bitcoin node is disconnected.");
+ }
+ }
+ }
+ _disposedValue = true;
+ }
+ }
+
+ /// <param name="onlyOwned">Only stop if this node owns the process.</param>
+ public async Task<bool> TryStopAsync(bool onlyOwned = true)
+ {
+ await DisposeAsync().ConfigureAwait(false);
+
Exception exThrown = null;
BitcoindRpcProcessBridge bridge = null;
@@ -280,36 +346,5 @@ namespace WalletWasabi.BitcoinCore
}
return false;
}
-
- public void DisconnectDisposeNullP2pNode()
- {
- if (P2pNode != null)
- {
- try
- {
- P2pNode?.Disconnect();
- }
- catch (Exception ex)
- {
- Logger.LogDebug(ex);
- }
- finally
- {
- try
- {
- P2pNode?.Dispose();
- }
- catch (Exception ex)
- {
- Logger.LogDebug(ex);
- }
- finally
- {
- P2pNode = null;
- Logger.LogInfo("P2p Bitcoin node is disconnected.");
- }
- }
- }
- }
}
}

MODIFY
WalletWasabi/BitcoinCore/CoreNodeParams.cs
WalletWasabi/BitcoinCore/CoreNodeParams.cs
@@ -3,7 +3,9 @@ using System;
using System.Collections.Generic;
using System.Text;
using WalletWasabi.BitcoinCore.Endpointing;
+using WalletWasabi.Blockchain.Mempool;
using WalletWasabi.Helpers;
+using WalletWasabi.Stores;
namespace WalletWasabi.BitcoinCore
{
@@ -11,6 +13,7 @@ namespace WalletWasabi.BitcoinCore
{
public CoreNodeParams(
Network network,
+ MempoolService mempoolService,
string dataDir,
bool tryRestart,
bool tryDeleteDataDir,
@@ -20,6 +23,7 @@ namespace WalletWasabi.BitcoinCore
int? prune)
{
Network = Guard.NotNull(nameof(network), network);
+ MempoolService = Guard.NotNull(nameof(mempoolService), mempoolService);
DataDir = Guard.NotNullOrEmptyOrWhitespace(nameof(dataDir), dataDir);
TryRestart = tryRestart;
TryDeleteDataDir = tryDeleteDataDir;
@@ -31,6 +35,7 @@ namespace WalletWasabi.BitcoinCore
public string DataDir { get; }
public Network Network { get; }
+ public MempoolService MempoolService { get; }
public bool TryRestart { get; }
public bool TryDeleteDataDir { get; }
public int? TxIndex { get; }

MODIFY
WalletWasabi/Blockchain/Mempool/MempoolBehavior.cs
WalletWasabi/Blockchain/Mempool/MempoolBehavior.cs
@@ -40,7 +40,7 @@ namespace WalletWasabi.Blockchain.Mempool
{
await ProcessGetDataAsync(node, getDataPayload).ConfigureAwait(false);
}
- else if (message.Message.Payload is TxPayload txPayload)
+ else if (message.Message.Payload is TxPayload txPayload && !MempoolService.TrustedNodeMode)
{
ProcessTx(txPayload);
}
@@ -127,8 +127,8 @@ namespace WalletWasabi.Blockchain.Mempool
}
}
- // if we already processed it continue;
- if (MempoolService.IsProcessed(inv.Hash))
+ // if we already processed it or we're in trusted node mode, then don't ask for it;
+ if (MempoolService.TrustedNodeMode || MempoolService.IsProcessed(inv.Hash))
{
continue;
}

MODIFY
WalletWasabi/Blockchain/Mempool/MempoolService.cs
WalletWasabi/Blockchain/Mempool/MempoolService.cs
@@ -22,6 +22,7 @@ namespace WalletWasabi.Blockchain.Mempool
private List<TransactionBroadcastEntry> BroadcastStore { get; }
private object BroadcastStoreLock { get; }
+ public bool TrustedNodeMode { get; set; }
public event EventHandler<SmartTransaction> TransactionReceived;

MODIFY
WalletWasabi/Blockchain/Transactions/TransactionBroadcastEntry.cs
WalletWasabi/Blockchain/Transactions/TransactionBroadcastEntry.cs
@@ -51,6 +51,14 @@ namespace WalletWasabi.Blockchain.Transactions
}
}
+ public void ConfirmPropagationForGood()
+ {
+ lock (Lock)
+ {
+ PropagationConfirmations = 21;
+ }
+ }
+
public int GetPropagationConfirmations()
{
lock (Lock)

MODIFY
WalletWasabi/Services/TrustedNodeNotifyingBehavior.cs
WalletWasabi/Services/TrustedNodeNotifyingBehavior.cs
@@ -5,7 +5,10 @@ using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
+using System.Threading.Tasks;
+using WalletWasabi.Blockchain.Mempool;
using WalletWasabi.Blockchain.Transactions;
+using WalletWasabi.Helpers;
using WalletWasabi.Logging;
using WalletWasabi.Models;
@@ -13,14 +16,13 @@ namespace WalletWasabi.Services
{
public class TrustedNodeNotifyingBehavior : NodeBehavior
{
- public event EventHandler<uint256> TransactionInv;
-
public event EventHandler<uint256> BlockInv;
- public event EventHandler<SmartTransaction> Transaction;
+ public MempoolService MempoolService { get; }
- public TrustedNodeNotifyingBehavior()
+ public TrustedNodeNotifyingBehavior(MempoolService mempoolService)
{
+ MempoolService = Guard.NotNull(nameof(mempoolService), mempoolService);
}
protected override void AttachCore()
@@ -39,16 +41,55 @@ namespace WalletWasabi.Services
{
if (message.Message.Payload is TxPayload txPayload)
{
- Transaction?.Invoke(this, new SmartTransaction(txPayload.Object, Height.Mempool));
+ ProcessTx(txPayload);
}
else if (message.Message.Payload is InvPayload invPayload)
+ {
+ await ProcessInvAsync(node, invPayload).ConfigureAwait(false);
+ }
+ }
+ catch (OperationCanceledException ex)
+ {
+ Logger.LogDebug(ex);
+ }
+ catch (Exception ex)
+ {
+ Logger.LogInfo($"Ignoring {ex.GetType()}: {ex.Message}");
+ Logger.LogDebug(ex);
+ }
+ }
+
+ private async Task ProcessInvAsync(Node node, InvPayload invPayload)
{
var getDataPayload = new GetDataPayload();
foreach (var inv in invPayload.Inventory)
{
if (inv.Type.HasFlag(InventoryType.MSG_TX))
{
- TransactionInv?.Invoke(this, inv.Hash);
+ 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.MakeBroadcasted();
+ entry.ConfirmPropagationForGood();
+ }
+ catch (Exception ex)
+ {
+ Logger.LogInfo(ex);
+ }
+ }
+
+ // if we already processed it continue;
+ if (MempoolService.IsProcessed(inv.Hash))
+ {
+ continue;
+ }
+
getDataPayload.Inventory.Add(inv);
}
@@ -64,22 +105,18 @@ namespace WalletWasabi.Services
await node.SendMessageAsync(getDataPayload).ConfigureAwait(false);
}
}
- }
- catch (OperationCanceledException ex)
- {
- Logger.LogDebug(ex);
- }
- catch (Exception ex)
+
+ private void ProcessTx(TxPayload payload)
{
- Logger.LogInfo($"Ignoring {ex.GetType()}: {ex.Message}");
- Logger.LogDebug(ex);
- }
+ Transaction transaction = payload.Object;
+ transaction.PrecomputeHash(false, true);
+ MempoolService.Process(transaction);
}
public override object Clone()
{
// Note that, this is not clone! So this must be used after we are connected to a node, because it'll have as many clones as nodes.
- return new TrustedNodeNotifyingBehavior();
+ return new TrustedNodeNotifyingBehavior(MempoolService);
}
}
}

MODIFY
WalletWasabi/Stores/BitcoinStore.cs
WalletWasabi/Stores/BitcoinStore.cs
@@ -19,7 +19,7 @@ namespace WalletWasabi.Stores
{
public bool IsInitialized { get; private set; }
private string WorkFolderPath { get; set; }
- private Network Network { get; set; }
+ public Network Network { get; private set; }
public IndexStore IndexStore { get; private set; }
public AllTransactionStore TransactionStore { get; private set; }

Implement local mempool receiving.