[PoC] Pre-fetching blocks to download

WalletWasabi.Daemon/Global.cs

@@ -1,4 +1,4 @@
 using Microsoft.Extensions.Caching.Memory;
 using NBitcoin;
 using Nito.AsyncEx;
 using System;
@@ -32,6 +32,7 @@
 using WalletWasabi.WabiSabi.Client.Banning;
 using WalletWasabi.WabiSabi.Client.RoundStateAwaiters;
 using WalletWasabi.Wallets;
+using WalletWasabi.Wallets.FilterProcessor;
 using WalletWasabi.WebClients.BlockstreamInfo;
 using WalletWasabi.WebClients.Wasabi;
 
@@ -111,14 +112,18 @@
 
 		// Block providers.
 		SpecificNodeBlockProvider = new SpecificNodeBlockProvider(Network, Config.ServiceConfiguration, HttpClientFactory.TorEndpoint);
-		var blockProvider = new SmartBlockProvider(
-			BitcoinStore.BlockRepository,
-			BitcoinCoreNode?.RpcClient is null ? null : new RpcBlockProvider(BitcoinCoreNode.RpcClient),
-			SpecificNodeBlockProvider,
-			new P2PBlockProvider(Network, HostedServices.Get<P2pNetwork>().Nodes, HttpClientFactory.IsTorEnabled),
-			Cache);
-
-		WalletManager = new WalletManager(config.Network, DataDir, new WalletDirectories(Config.Network, DataDir), BitcoinStore, wasabiSynchronizer, HostedServices.Get<HybridFeeProvider>(), blockProvider, config.ServiceConfiguration);
+		P2PNodesManager = new P2PNodesManager(Network, HostedServices.Get<P2pNetwork>().Nodes, HttpClientFactory.IsTorEnabled);
+
+		var trustedFullNodeBlockProviders = BitcoinCoreNode?.RpcClient is null
+			? new IBlockProvider[] { SpecificNodeBlockProvider }
+			: new IBlockProvider[] { new RpcBlockProvider(BitcoinCoreNode.RpcClient), SpecificNodeBlockProvider };
+
+		BlockDownloadService = new BlockDownloadService(
+			fileSystemBlockRepository: BitcoinStore.BlockRepository,
+			trustedFullNodeBlockProviders: trustedFullNodeBlockProviders,
+			new P2PBlockProvider(P2PNodesManager));
+
+		WalletManager = new WalletManager(config.Network, DataDir, new WalletDirectories(Config.Network, DataDir), BitcoinStore, wasabiSynchronizer, HostedServices.Get<HybridFeeProvider>(), BlockDownloadService, config.ServiceConfiguration);
 		TransactionBroadcaster = new TransactionBroadcaster(Network, BitcoinStore, HttpClientFactory, WalletManager);
 
 		CoinPrison = CoinPrison.CreateOrLoadFromFile(DataDir);
@@ -150,6 +155,8 @@
 	public TransactionBroadcaster TransactionBroadcaster { get; set; }
 	public CoinJoinProcessor? CoinJoinProcessor { get; set; }
 	private SpecificNodeBlockProvider SpecificNodeBlockProvider { get; }
+	private BlockDownloadService BlockDownloadService { get; }
+	private P2PNodesManager P2PNodesManager { get; }
 	private TorProcessManager? TorManager { get; set; }
 	public CoreNode? BitcoinCoreNode { get; private set; }
 	public TorStatusChecker TorStatusChecker { get; set; }
@@ -247,6 +254,8 @@
 
 				await StartRpcServerAsync(terminateService, cancel).ConfigureAwait(false);
 
+				await BlockDownloadService.StartAsync(cancel).ConfigureAwait(false);
+
 				WalletManager.Initialize();
 			}
 			finally
@@ -436,6 +445,12 @@
 					Logger.LogInfo($"{nameof(SpecificNodeBlockProvider)} is disposed.");
 				}
 
+				if (BlockDownloadService is { } blockDownloadService)
+				{
+					blockDownloadService.Dispose();
+					Logger.LogInfo($"{nameof(BlockDownloadService)} is disposed.");
+				}
+
 				if (CoinJoinProcessor is { } coinJoinProcessor)
 				{
 					coinJoinProcessor.Dispose();

WalletWasabi.Tests/IntegrationTests/BlockDownloadTests.cs

@@ -0,0 +1,110 @@
+using Moq;
+using NBitcoin;
+using NBitcoin.Protocol;
+using NBitcoin.Protocol.Behaviors;
+using System.Collections.Generic;
+using System.Diagnostics;
+using System.IO;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using WalletWasabi.Helpers;
+using WalletWasabi.Logging;
+using WalletWasabi.Tests.Helpers;
+using WalletWasabi.Wallets;
+using WalletWasabi.Wallets.BlockProvider;
+using WalletWasabi.Wallets.FilterProcessor;
+using Xunit;
+using static WalletWasabi.Wallets.FilterProcessor.BlockDownloadService;
+
+namespace WalletWasabi.Tests.IntegrationTests;
+
+/// <summary>
+/// Integration tests for <see cref="BlockDownloadService"/>.
+/// </summary>
+public class BlockDownloadTests
+{
+	public static readonly SortedDictionary<uint, uint256> HeightToBlockHash = new()
+	{
+		{ 610_000, new uint256("0000000000000000000a6f607f74db48dae0a94022c10354536394c17672b7f7") },
+		{ 610_001, new uint256("0000000000000000000b9cfe321443cb2ced7a4182b8594f077ee4e23d6c2ae2") },
+		{ 610_002, new uint256("0000000000000000000aef4754dfc99fed93f40d8b90d06037075bc59af3ab4d") },
+		{ 610_003, new uint256("00000000000000000012afa0710628d1c504f1d40e1fc1b71c8cc66982547d0b") },
+		{ 610_004, new uint256("000000000000000000081bb924d567c05c39dabc7feda74b02a1327dc376634d") },
+		{ 610_005, new uint256("0000000000000000000f68159a871ea320691afcb3e0b7e8e3e03f53c0c0d58d") },
+		{ 610_006, new uint256("0000000000000000000571a7383c5679bf6967b6eb43efa807167ea3d1761e84") },
+		{ 610_007, new uint256("0000000000000000000ab3657ecbfa7e4e555d58f4fe6a0279bee8ea29ae2df7") },
+		{ 610_008, new uint256("000000000000000000157e58b222e0097afee38d6920f0750e177c0d02a5d593") },
+		{ 610_009, new uint256("00000000000000000003354e9f48af4b5850fd884f38886d00b92bfd4a015d54") },
+	};
+
+	[Fact]
+	public async Task BlockDownloadingTestAsync()
+	{
+		using CancellationTokenSource testCts = new(TimeSpan.FromMinutes(10));
+
+		Mock<IFileSystemBlockRepository> mockFileSystemBlockRepository = new(MockBehavior.Strict);
+		_ = mockFileSystemBlockRepository.Setup(c => c.TryGetAsync(It.IsAny<uint256>(), It.IsAny<CancellationToken>()))
+			.ReturnsAsync((Block?)null);
+		_ = mockFileSystemBlockRepository.Setup(c => c.SaveAsync(It.IsAny<Block>(), It.IsAny<CancellationToken>()))
+			.Returns(Task.CompletedTask);
+
+		RuntimeParams.SetDataDir(Path.Combine(Common.DataDir, "RegTests", "Backend"));
+		await RuntimeParams.LoadAsync();
+
+		string roamingDir = Environment.GetFolderPath(Environment.SpecialFolder.ApplicationData);
+
+		string addressManagerFilePath = Path.Combine(roamingDir, "WalletWasabi", "Client", "BitcoinP2pNetwork", $"AddressManager{Network.Main}.dat");
+		AddressManager addressManager = AddressManager.LoadPeerFile(addressManagerFilePath);
+		AddressManagerBehavior addressManagerBehavior = new(addressManager)
+		{
+			Mode = AddressManagerBehaviorMode.Discover
+		};
+
+		using NodesGroup nodes = new(Network.Main);
+		nodes.NodeConnectionParameters.TemplateBehaviors.Add(addressManagerBehavior);
+		nodes.Connect();
+
+		while (nodes.ConnectedNodes.Count == 0)
+		{
+			await Task.Delay(1000, testCts.Token);
+		}
+
+		P2PBlockProvider p2PBlockProvider = new(Network.Main, nodes, isTorEnabled: false);
+		using BlockDownloadService blockDownloadService = new(mockFileSystemBlockRepository.Object, trustedFullNodeBlockProviders: [], p2PBlockProvider);
+
+		try
+		{
+			await blockDownloadService.StartAsync(testCts.Token);
+
+			Stopwatch stopwatch = Stopwatch.StartNew();
+
+			List<Task<IResult>> tasks = [];
+
+			foreach ((uint height, uint256 blockHash) in HeightToBlockHash)
+			{
+				Task<IResult> taskCompletionSource = blockDownloadService.TryGetBlockAsync(P2pSourceRequest.Automatic, blockHash, new Priority(SyncType.Complete, height), testCts.Token);
+				tasks.Add(taskCompletionSource);
+			}
+
+			await Task.WhenAll(tasks);
+
+			uint blockHeight = HeightToBlockHash.First().Key;
+
+			foreach (Task<IResult> task in tasks)
+			{
+				IResult result = await task;
+				Logger.LogInfo($"Block #{blockHeight} finished with result: {result}");
+
+				blockHeight++;
+			}
+
+			stopwatch.Stop();
+			Logger.LogInfo($"Test finished in {stopwatch.ElapsedMilliseconds} ms");
+		}
+		finally
+		{
+			await blockDownloadService.StopAsync(testCts.Token);
+		}
+	}
+}

WalletWasabi.Tests/IntegrationTests/P2pTests.cs

@@ -20,6 +20,8 @@
 using WalletWasabi.Stores;
 using WalletWasabi.Tests.Helpers;
 using WalletWasabi.Wallets;
+using WalletWasabi.Wallets.BlockProvider;
+using WalletWasabi.Wallets.FilterProcessor;
 using WalletWasabi.WebClients.Wasabi;
 using Xunit;
 
@@ -112,12 +114,10 @@
 		IFileSystemBlockRepository blockRepository = bitcoinStore.BlockRepository;
 		await using SpecificNodeBlockProvider specificNodeBlockProvider = new(network, serviceConfig, httpClientFactory.TorEndpoint);
 
-		IBlockProvider blockProvider = new SmartBlockProvider(
-			blockRepository,
-			rpcBlockProvider: null,
-			specificNodeBlockProvider,
-			new P2PBlockProvider(network, nodes, httpClientFactory.IsTorEnabled),
-			cache);
+		var blockDownloadService = new BlockDownloadService(
+			bitcoinStore.BlockRepository,
+			new IBlockProvider[] { specificNodeBlockProvider },
+			new P2PBlockProvider(network, nodes, httpClientFactory.IsTorEnabled));
 
 		using Wallet wallet = Wallet.CreateAndRegisterServices(
 			network,
@@ -127,7 +127,7 @@
 			dataDir,
 			new ServiceConfiguration(new IPEndPoint(IPAddress.Loopback, network.DefaultPort), Money.Coins(Constants.DefaultDustThreshold)),
 			feeProvider,
-			blockProvider);
+			blockDownloadService);
 		Assert.True(Directory.Exists(blocks.BlocksFolderPath));
 
 		try
@@ -144,18 +144,18 @@
 
 			nodes.Connect();
 
-			var downloadTasks = new List<Task<Block>>();
+			var downloadTasks = new List<Task<BlockDownloadService.IResult>>();
 			using var cts = new CancellationTokenSource(TimeSpan.FromMinutes(4));
 			foreach (var hash in blocksToDownload)
 			{
-				downloadTasks.Add(blockProvider.GetBlockAsync(hash, cts.Token));
+				downloadTasks.Add(blockDownloadService.TryGetBlockAsync(P2pSourceRequest.Automatic, hash, new Priority(SyncType.Complete), uint.MaxValue, cts.Token));
 			}
 
 			await nodeConnectionAwaiter.WaitAsync(TimeSpan.FromMinutes(3));
 
 			var i = 0;
 			var hashArray = blocksToDownload.ToArray();
-			foreach (var block in await Task.WhenAll(downloadTasks))
+			foreach (var result in await Task.WhenAll(downloadTasks))
 			{
 				Assert.True(File.Exists(Path.Combine(blocks.BlocksFolderPath, hashArray[i].ToString())));
 				i++;

WalletWasabi.Tests/RegressionTests/BuildTransactionReorgsTest.cs

@@ -21,6 +21,7 @@
 using WalletWasabi.Stores;
 using WalletWasabi.Tests.XunitConfiguration;
 using WalletWasabi.Wallets;
+using WalletWasabi.Wallets.FilterProcessor;
 using WalletWasabi.WebClients.Wasabi;
 using Xunit;
 
@@ -75,14 +76,12 @@
 		using MemoryCache cache = CreateMemoryCache();
 		await using SpecificNodeBlockProvider specificNodeBlockProvider = new(network, serviceConfiguration, httpClientFactory.TorEndpoint);
 
-		var blockProvider = new SmartBlockProvider(
+		var blockDownloadService = new BlockDownloadService(
 			bitcoinStore.BlockRepository,
-			rpcBlockProvider: null,
-			specificNodeBlockProvider,
-			new P2PBlockProvider(network, nodes, httpClientFactory.IsTorEnabled),
-			cache);
+			new IBlockProvider[] { specificNodeBlockProvider },
+			new P2PBlockProvider(network, nodes, httpClientFactory.IsTorEnabled));
 
-		WalletManager walletManager = new(network, workDir, new WalletDirectories(network, workDir), bitcoinStore, synchronizer, feeProvider, blockProvider, serviceConfiguration);
+		WalletManager walletManager = new(network, workDir, new WalletDirectories(network, workDir), bitcoinStore, synchronizer, feeProvider, blockDownloadService, serviceConfiguration);
 		walletManager.Initialize();
 
 		var baseTip = await rpc.GetBestBlockHashAsync();

WalletWasabi.Tests/RegressionTests/BuildTransactionValidationsTest.cs

@@ -21,6 +21,7 @@
 using WalletWasabi.Stores;
 using WalletWasabi.Tests.XunitConfiguration;
 using WalletWasabi.Wallets;
+using WalletWasabi.Wallets.FilterProcessor;
 using WalletWasabi.WebClients.Wasabi;
 using Xunit;
 
@@ -74,14 +75,12 @@
 		using MemoryCache cache = CreateMemoryCache();
 		await using SpecificNodeBlockProvider specificNodeBlockProvider = new(network, serviceConfiguration, httpClientFactory.TorEndpoint);
 
-		var blockProvider = new SmartBlockProvider(
+		var blockDownloadService = new BlockDownloadService(
 			bitcoinStore.BlockRepository,
-			rpcBlockProvider: null,
-			specificNodeBlockProvider,
-			p2PBlockProvider: new P2PBlockProvider(network, nodes, httpClientFactory.IsTorEnabled),
-			cache);
+			new IBlockProvider[] { specificNodeBlockProvider },
+			new P2PBlockProvider(network, nodes, httpClientFactory.IsTorEnabled));
 
-		using var wallet = Wallet.CreateAndRegisterServices(network, bitcoinStore, keyManager, synchronizer, workDir, serviceConfiguration, feeProvider, blockProvider);
+		using var wallet = Wallet.CreateAndRegisterServices(network, bitcoinStore, keyManager, synchronizer, workDir, serviceConfiguration, feeProvider, blockDownloadService);
 		wallet.NewFiltersProcessed += setup.Wallet_NewFiltersProcessed;
 
 		using Key key = new();

WalletWasabi.Tests/RegressionTests/CancelTests.cs

@@ -22,6 +22,7 @@
 using WalletWasabi.Helpers;
 using WalletWasabi.Logging;
 using WalletWasabi.Tests.Helpers;
+using WalletWasabi.Wallets.FilterProcessor;
 
 namespace WalletWasabi.Tests.RegressionTests;
 
@@ -73,14 +74,12 @@ public async Task CancelTestsAsync()
 		using MemoryCache cache = BitcoinFactory.CreateMemoryCache();
 		await using SpecificNodeBlockProvider specificNodeBlockProvider = new(network, serviceConfiguration, httpClientFactory.TorEndpoint);
 
-		var blockProvider = new SmartBlockProvider(
+		var blockDownloadService = new BlockDownloadService(
 			bitcoinStore.BlockRepository,
-			rpcBlockProvider: null,
-			specificNodeBlockProvider,
-			new P2PBlockProvider(network, nodes, httpClientFactory.IsTorEnabled),
-			cache);
+			new IBlockProvider[] { specificNodeBlockProvider },
+			new P2PBlockProvider(network, nodes, httpClientFactory.IsTorEnabled));
 
-		WalletManager walletManager = new(network, workDir, new WalletDirectories(network, workDir), bitcoinStore, synchronizer, feeProvider, blockProvider, serviceConfiguration);
+		WalletManager walletManager = new(network, workDir, new WalletDirectories(network, workDir), bitcoinStore, synchronizer, feeProvider, blockDownloadService, serviceConfiguration);
 		walletManager.Initialize();
 
 		// Get some money, make it confirm.

WalletWasabi.Tests/RegressionTests/MaxFeeTests.cs

@@ -20,6 +20,7 @@
 using WalletWasabi.Tests.Helpers;
 using WalletWasabi.Tests.XunitConfiguration;
 using WalletWasabi.Wallets;
+using WalletWasabi.Wallets.FilterProcessor;
 using WalletWasabi.WebClients.Wasabi;
 using Xunit;
 
@@ -73,14 +74,12 @@ public async Task CalculateMaxFeeTestAsync()
 		using MemoryCache cache = BitcoinFactory.CreateMemoryCache();
 		await using SpecificNodeBlockProvider specificNodeBlockProvider = new(network, serviceConfiguration, httpClientFactory.TorEndpoint);
 
-		var blockProvider = new SmartBlockProvider(
+		var blockDownloadService = new BlockDownloadService(
 			bitcoinStore.BlockRepository,
-			rpcBlockProvider: null,
-			specificNodeBlockProvider,
-			new P2PBlockProvider(network, nodes, httpClientFactory.IsTorEnabled),
-			cache);
+			new IBlockProvider[] { specificNodeBlockProvider },
+			new P2PBlockProvider(network, nodes, httpClientFactory.IsTorEnabled));
 
-		WalletManager walletManager = new(network, workDir, new WalletDirectories(network, workDir), bitcoinStore, synchronizer, feeProvider, blockProvider, serviceConfiguration);
+		WalletManager walletManager = new(network, workDir, new WalletDirectories(network, workDir), bitcoinStore, synchronizer, feeProvider, blockDownloadService, serviceConfiguration);
 		walletManager.Initialize();
 
 		// Get some money, make it confirm.

WalletWasabi.Tests/RegressionTests/ReceiveSpeedupTests.cs

@@ -24,6 +24,7 @@
 using WalletWasabi.Helpers;
 using WalletWasabi.Blockchain.Transactions;
 using WalletWasabi.Exceptions;
+using WalletWasabi.Wallets.FilterProcessor;
 
 namespace WalletWasabi.Tests.RegressionTests;
 
@@ -75,14 +76,12 @@ public async Task ReceiveSpeedupTestsAsync()
 		using MemoryCache cache = BitcoinFactory.CreateMemoryCache();
 		await using SpecificNodeBlockProvider specificNodeBlockProvider = new(network, serviceConfiguration, httpClientFactory.TorEndpoint);
 
-		var blockProvider = new SmartBlockProvider(
+		var blockDownloadService = new BlockDownloadService(
 			bitcoinStore.BlockRepository,
-			rpcBlockProvider: null,
-			specificNodeBlockProvider,
-			new P2PBlockProvider(network, nodes, httpClientFactory.IsTorEnabled),
-			cache);
+			new IBlockProvider[] { specificNodeBlockProvider },
+			new P2PBlockProvider(network, nodes, httpClientFactory.IsTorEnabled));
 
-		WalletManager walletManager = new(network, workDir, new WalletDirectories(network, workDir), bitcoinStore, synchronizer, feeProvider, blockProvider, serviceConfiguration);
+		WalletManager walletManager = new(network, workDir, new WalletDirectories(network, workDir), bitcoinStore, synchronizer, feeProvider, blockDownloadService, serviceConfiguration);
 		walletManager.Initialize();
 
 		try

WalletWasabi.Tests/RegressionTests/ReplaceByFeeTxTest.cs

@@ -22,6 +22,7 @@
 using WalletWasabi.WebClients.Wasabi;
 using Xunit;
 using WalletWasabi.Tests.Helpers;
+using WalletWasabi.Wallets.FilterProcessor;
 
 namespace WalletWasabi.Tests.RegressionTests;
 
@@ -67,18 +68,15 @@
 
 		// 5. Create wallet service.
 		var workDir = Common.GetWorkDir();
 
-		using MemoryCache cache = BitcoinFactory.CreateMemoryCache();
 		await using SpecificNodeBlockProvider specificNodeBlockProvider = new(network, serviceConfiguration, httpClientFactory.TorEndpoint);
 
-		var blockProvider = new SmartBlockProvider(
+		var blockDownloadService = new BlockDownloadService(
 			bitcoinStore.BlockRepository,
-			rpcBlockProvider: null,
-			specificNodeBlockProvider,
-			new P2PBlockProvider(network, nodes, httpClientFactory.IsTorEnabled),
-			cache);
+			new IBlockProvider[] { specificNodeBlockProvider },
+			new P2PBlockProvider(network, nodes, httpClientFactory.IsTorEnabled));
 
-		using var wallet = Wallet.CreateAndRegisterServices(network, bitcoinStore, keyManager, synchronizer, workDir, serviceConfiguration, feeProvider, blockProvider);
+		using var wallet = Wallet.CreateAndRegisterServices(network, bitcoinStore, keyManager, synchronizer, workDir, serviceConfiguration, feeProvider, blockDownloadService);
 		wallet.NewFiltersProcessed += setup.Wallet_NewFiltersProcessed;
 
 		Assert.Empty(wallet.Coins);