Change the filter index builder to use bitcoin knots getblockverbose rpc
MODIFY
WalletWasabi.Backend/Global.cs
WalletWasabi.Backend/Global.cs
@@ -84,9 +84,8 @@ namespace WalletWasabi.Backend
// Initialize index building
var indexBuilderServiceDir = Path.Combine(DataDir, "IndexBuilderService");
var indexFilePath = Path.Combine(indexBuilderServiceDir, $"Index{RpcClient.Network}.dat");
- var utxoSetFilePath = Path.Combine(indexBuilderServiceDir, $"UtxoSet{RpcClient.Network}.dat");
var blockNotifier = HostedServices.FirstOrDefault<BlockNotifier>();
- IndexBuilderService = new IndexBuilderService(RpcClient, blockNotifier, indexFilePath, utxoSetFilePath);
+ IndexBuilderService = new IndexBuilderService(RpcClient, blockNotifier, indexFilePath);
Coordinator = new Coordinator(RpcClient.Network, blockNotifier, Path.Combine(DataDir, "CcjCoordinator"), RpcClient, roundConfig);
IndexBuilderService.Synchronize();
Logger.LogInfo($"{nameof(IndexBuilderService)} is successfully initialized and started synchronization.");

MODIFY
WalletWasabi.Tests/RegressionTests/BackendTests.cs
WalletWasabi.Tests/RegressionTests/BackendTests.cs
@@ -135,9 +135,8 @@ namespace WalletWasabi.Tests.RegressionTests
var indexBuilderServiceDir = Common.GetWorkDir();
var indexFilePath = Path.Combine(indexBuilderServiceDir, $"Index{rpc.Network}.dat");
- var utxoSetFilePath = Path.Combine(indexBuilderServiceDir, $"UtxoSet{rpc.Network}.dat");
- var indexBuilderService = new IndexBuilderService(rpc, global.HostedServices.FirstOrDefault<BlockNotifier>(), indexFilePath, utxoSetFilePath);
+ var indexBuilderService = new IndexBuilderService(rpc, global.HostedServices.FirstOrDefault<BlockNotifier>(), indexFilePath);
try
{
indexBuilderService.Synchronize();

MODIFY
WalletWasabi.Tests/RegressionTests/WalletTests.cs
WalletWasabi.Tests/RegressionTests/WalletTests.cs
@@ -189,14 +189,6 @@ namespace WalletWasabi.Tests.RegressionTests
var tipBlock = await rpc.GetBlockHeaderAsync(tip);
Assert.Equal(tipBlock.HashPrevBlock, bitcoinStore.SmartHeaderChain.GetChain().Select(x => x.header.BlockHash).ToArray()[bitcoinStore.SmartHeaderChain.HashCount - 2]);
- var utxoPath = global.IndexBuilderService.Bech32UtxoSetFilePath;
- var utxoLines = await File.ReadAllTextAsync(utxoPath);
- Assert.Contains(tx1.ToString(), utxoLines);
- Assert.Contains(tx2.ToString(), utxoLines);
- Assert.Contains(tx3.ToString(), utxoLines);
- Assert.DoesNotContain(tx4.ToString(), utxoLines); // make sure only bech is recorded
- Assert.DoesNotContain(tx5.ToString(), utxoLines); // make sure only bech is recorded
-
// Test synchronization after fork.
await rpc.InvalidateBlockAsync(tip); // Reorg 1
tip = await rpc.GetBestBlockHashAsync();
@@ -206,14 +198,6 @@ namespace WalletWasabi.Tests.RegressionTests
await rpc.GenerateAsync(5);
await WaitForIndexesToSyncAsync(global, TimeSpan.FromSeconds(90), bitcoinStore);
- utxoLines = await File.ReadAllTextAsync(utxoPath);
- Assert.Contains(tx1bumpRes.TransactionId.ToString(), utxoLines); // assert the tx1bump is the correct tx
- Assert.DoesNotContain(tx1.ToString(), utxoLines); // assert tx1 is abandoned (despite it confirmed previously)
- Assert.Contains(tx2.ToString(), utxoLines);
- Assert.Contains(tx3.ToString(), utxoLines);
- Assert.DoesNotContain(tx4.ToString(), utxoLines);
- Assert.DoesNotContain(tx5.ToString(), utxoLines);
-
var hashes = bitcoinStore.SmartHeaderChain.GetChain().Select(x => x.header.BlockHash).ToArray();
Assert.DoesNotContain(tip, hashes);
Assert.DoesNotContain(tipBlock.HashPrevBlock, hashes);

MODIFY
WalletWasabi/Blockchain/BlockFilters/IndexBuilderService.cs
WalletWasabi/Blockchain/BlockFilters/IndexBuilderService.cs
using NBitcoin;
-using NBitcoin.Protocol;
using NBitcoin.RPC;
using Nito.AsyncEx;
using System;
@@ -10,39 +9,40 @@ using System.Threading;
using System.Threading.Tasks;
using WalletWasabi.Backend.Models;
using WalletWasabi.BitcoinCore;
-using WalletWasabi.Blockchain.BlockFilters.History;
+using WalletWasabi.BitcoinCore.RpcModels;
using WalletWasabi.Blockchain.Blocks;
using WalletWasabi.Helpers;
using WalletWasabi.Logging;
using WalletWasabi.Models;
-using WalletWasabi.Stores;
namespace WalletWasabi.Blockchain.BlockFilters
{
public class IndexBuilderService
{
+ private const long NotStarted = 0;
+ private const long Running = 1;
+ private const long Stopping = 2;
+ private const long Stopped = 3;
+
/// <summary>
/// 0: Not started, 1: Running, 2: Stopping, 3: Stopped
/// </summary>
- private long _running;
+ private long _serviceStatus;
- private long _runner;
+ private long _workerCount;
- public IndexBuilderService(IRPCClient rpc, BlockNotifier blockNotifier, string indexFilePath, string bech32UtxoSetFilePath)
+ public IndexBuilderService(IRPCClient rpc, BlockNotifier blockNotifier, string indexFilePath)
{
RpcClient = Guard.NotNull(nameof(rpc), rpc);
BlockNotifier = Guard.NotNull(nameof(blockNotifier), blockNotifier);
IndexFilePath = Guard.NotNullOrEmptyOrWhitespace(nameof(indexFilePath), indexFilePath);
- Bech32UtxoSetFilePath = Guard.NotNullOrEmptyOrWhitespace(nameof(bech32UtxoSetFilePath), bech32UtxoSetFilePath);
- Bech32UtxoSet = new Dictionary<OutPoint, UtxoEntry>();
- Bech32UtxoSetHistory = new List<ActionHistoryHelper>(capacity: 100);
Index = new List<FilterModel>();
IndexLock = new AsyncLock();
StartingHeight = SmartHeader.GetStartingHeader(RpcClient.Network).Height;
- _running = 0;
+ _serviceStatus = NotStarted;
IoHelpers.EnsureContainingDirectoryExists(IndexFilePath);
if (File.Exists(IndexFilePath))
@@ -61,29 +61,6 @@ namespace WalletWasabi.Blockchain.BlockFilters
}
}
- IoHelpers.EnsureContainingDirectoryExists(bech32UtxoSetFilePath);
- if (File.Exists(bech32UtxoSetFilePath))
- {
- if (RpcClient.Network == Network.RegTest)
- {
- File.Delete(bech32UtxoSetFilePath); // RegTest is not a global ledger, better to delete it.
- }
- else
- {
- foreach (var line in File.ReadAllLines(Bech32UtxoSetFilePath))
- {
- var parts = line.Split(':');
-
- var txHash = new uint256(parts[0]);
- var nIn = int.Parse(parts[1]);
- var script = new Script(ByteHelpers.FromHex(parts[2]), true);
- OutPoint outPoint = new OutPoint(txHash, nIn);
- var utxoEntry = new UtxoEntry(outPoint, script);
- Bech32UtxoSet.Add(outPoint, utxoEntry);
- }
- }
- }
-
BlockNotifier.OnBlock += BlockNotifier_OnBlock;
}
@@ -92,15 +69,10 @@ namespace WalletWasabi.Blockchain.BlockFilters
public IRPCClient RpcClient { get; }
public BlockNotifier BlockNotifier { get; }
public string IndexFilePath { get; }
- public string Bech32UtxoSetFilePath { get; }
-
private List<FilterModel> Index { get; }
private AsyncLock IndexLock { get; }
public uint StartingHeight { get; }
- private Dictionary<OutPoint, UtxoEntry> Bech32UtxoSet { get; }
- private List<ActionHistoryHelper> Bech32UtxoSetHistory { get; }
-
- public bool IsRunning => Interlocked.Read(ref _running) == 1;
+ public bool IsRunning => Interlocked.Read(ref _serviceStatus) == Running;
public static GolombRiceFilter CreateDummyEmptyFilter(uint256 blockHash)
{
@@ -118,148 +90,105 @@ namespace WalletWasabi.Blockchain.BlockFilters
{
try
{
- if (Interlocked.Read(ref _runner) >= 2)
+ if (Interlocked.Read(ref _workerCount) >= 2)
{
return;
}
- Interlocked.Increment(ref _runner);
- while (Interlocked.Read(ref _runner) != 1)
+ Interlocked.Increment(ref _workerCount);
+ while (Interlocked.Read(ref _workerCount) != 1)
{
await Task.Delay(100);
}
- if (Interlocked.Read(ref _running) >= 2)
+ if (Interlocked.Read(ref _serviceStatus) >= Stopping)
{
return;
}
try
{
- Interlocked.Exchange(ref _running, 1);
+ Interlocked.Exchange(ref _serviceStatus, Running);
- var isImmature = false; // The last 100 blocks are reorgable. (Assume it is mature at first.)
SyncInfo syncInfo = null;
while (IsRunning)
{
try
{
// If we did not yet initialized syncInfo, do so.
- if (syncInfo is null)
+ if (syncInfo == null)
{
syncInfo = await GetSyncInfoAsync();
}
- uint heightToRequest = StartingHeight;
+ uint currentHeight = 0;
uint256 currentHash = null;
using (await IndexLock.LockAsync())
{
if (Index.Count != 0)
{
var lastIndex = Index[^1];
- heightToRequest = lastIndex.Header.Height + 1;
+ currentHeight = lastIndex.Header.Height;
currentHash = lastIndex.Header.BlockHash;
}
- }
-
- // If not synchronized or already 5 min passed since last update, get the latest blockchain info.
- if (!syncInfo.IsCoreSynchornized || DateTimeOffset.UtcNow - syncInfo.BlockchainInfoUpdated > TimeSpan.FromMinutes(5))
+ else
{
- syncInfo = await GetSyncInfoAsync();
+ currentHash = StartingHeight == 0 ? uint256.Zero :
+ await RpcClient.GetBlockHashAsync((int)StartingHeight - 1);
+ currentHeight = StartingHeight - 1;
+ }
}
- if (syncInfo.BlockCount - heightToRequest <= 100)
- {
- // Both Wasabi and our Core node is in sync. Start doing stuff through P2P from now on.
- if (syncInfo.IsCoreSynchornized && syncInfo.BlockCount == heightToRequest - 1)
+ var coreNotSynced = !syncInfo.IsCoreSynchornized;
+ var tipReached = syncInfo.BlockCount == currentHeight;
+ var refresh = DateTimeOffset.UtcNow - syncInfo.BlockchainInfoUpdated > TimeSpan.FromMinutes(5);
+ if (coreNotSynced || tipReached || refresh)
{
syncInfo = await GetSyncInfoAsync();
- // Double it to make sure not to accidentally miss any notification.
- if (syncInfo.IsCoreSynchornized && syncInfo.BlockCount == heightToRequest - 1)
- {
- // Mark the process notstarted, so it can be started again and finally block can mark it is stopped.
- Interlocked.Exchange(ref _running, 0);
- return;
- }
- }
-
- // Mark the synchronizing process is working with immature blocks from now on.
- isImmature = true;
}
- Block block = await RpcClient.GetBlockAsync(heightToRequest);
-
- // Reorg check, except if we're requesting the starting height, because then the "currentHash" wouldn't exist.
- if (heightToRequest != StartingHeight && currentHash != block.Header.HashPrevBlock)
+ // If wasabi filter height is the same as core we may be done.
+ if (syncInfo.BlockCount == currentHeight)
{
- // Reorg can happen only when immature. (If it'd not be immature, that'd be a huge issue.)
- if (isImmature)
+ // Check that core is fully synced
+ if (syncInfo.IsCoreSynchornized && !syncInfo.InitialBlockDownload)
{
- await ReorgOneAsync();
+ // Mark the process notstarted, so it can be started again
+ // and finally block can mark it as stopped.
+ Interlocked.Exchange(ref _serviceStatus, NotStarted);
+ return;
}
else
{
- Logger.LogCritical("This is something serious! Over 100 block reorg is noticed! We cannot handle that!");
- }
-
- // Skip the current block.
+ // Core is catching up give it a few milliseconds
+ await Task.Delay(100);
continue;
}
-
- if (isImmature)
- {
- PrepareBech32UtxoSetHistory();
}
- var scripts = new HashSet<Script>();
+ uint nextHeight = currentHeight + 1;
+ uint256 blockHash = await RpcClient.GetBlockHashAsync((int)nextHeight);
+ VerboseBlockInfo block = await RpcClient.GetVerboseBlockAsync(blockHash);
- foreach (var tx in block.Transactions)
- {
- // If stop was requested return.
- // Because this tx iteration can take even minutes
- // It does not need to be accessed with a thread safe fashion with Interlocked through IsRunning, this may have some performance benefit
- if (_running != 1)
+ // Check if we are still on the best chain,
+ // if not rewind filters til we find the fork.
+ if (currentHash != block.PrevBlockHash)
{
- return;
- }
+ Logger.LogWarning("Reorg observed on the network");
- for (int i = 0; i < tx.Outputs.Count; i++)
- {
- var output = tx.Outputs[i];
- if (output.ScriptPubKey.IsScriptType(ScriptType.P2WPKH))
- {
- var outpoint = new OutPoint(tx.GetHash(), i);
- var utxoEntry = new UtxoEntry(outpoint, output.ScriptPubKey);
- Bech32UtxoSet.Add(outpoint, utxoEntry);
- if (isImmature)
- {
- Bech32UtxoSetHistory.Last().StoreAction(Operation.Add, outpoint, output.ScriptPubKey);
- }
- scripts.Add(output.ScriptPubKey);
- }
- }
+ await ReorgOneAsync();
- foreach (var input in tx.Inputs)
- {
- OutPoint prevOut = input.PrevOut;
- if (Bech32UtxoSet.TryGetValue(prevOut, out UtxoEntry foundUtxoEntry))
- {
- var foundScript = foundUtxoEntry.Script;
- Bech32UtxoSet.Remove(prevOut);
- if (isImmature)
- {
- Bech32UtxoSetHistory.Last().StoreAction(Operation.Remove, prevOut, foundScript);
- }
- scripts.Add(foundScript);
- }
- }
+ // Skip the current block.
+ continue;
}
+ var scripts = FetchScripts(block);
+
GolombRiceFilter filter;
if (scripts.Any())
{
filter = new GolombRiceFilterBuilder()
- .SetKey(block.GetHash())
+ .SetKey(block.Hash)
.SetP(20)
.SetM(1 << 20)
.AddEntries(scripts.Select(x => x.ToCompressedBytes()))
@@ -269,35 +198,27 @@ namespace WalletWasabi.Blockchain.BlockFilters
{
// We cannot have empty filters, because there was a bug in GolombRiceFilterBuilder that evaluates empty filters to true.
// And this must be fixed in a backwards compatible way, so we create a fake filter with a random scp instead.
- filter = CreateDummyEmptyFilter(block.GetHash());
+ filter = CreateDummyEmptyFilter(block.Hash);
}
- var smartHeader = new SmartHeader(block.GetHash(), block.Header.HashPrevBlock, heightToRequest, block.Header.BlockTime);
+ var smartHeader = new SmartHeader(block.Hash, block.PrevBlockHash, nextHeight, block.BlockTime);
var filterModel = new FilterModel(smartHeader, filter);
await File.AppendAllLinesAsync(IndexFilePath, new[] { filterModel.ToLine() });
+
using (await IndexLock.LockAsync())
{
Index.Add(filterModel);
}
- if (File.Exists(Bech32UtxoSetFilePath))
- {
- File.Delete(Bech32UtxoSetFilePath);
- }
- var bech32UtxoSetLines = Bech32UtxoSet.Select(entry => entry.Value.Line);
-
- // Keep it sync unless you fix the performance issue with async.
- File.WriteAllLines(Bech32UtxoSetFilePath, bech32UtxoSetLines);
// If not close to the tip, just log debug.
- // Use height.Value instead of simply height, because it cannot be negative height.
- if (syncInfo.BlockCount - heightToRequest <= 3 || heightToRequest % 100 == 0)
+ if (syncInfo.BlockCount - nextHeight <= 3 || nextHeight % 100 == 0)
{
- Logger.LogInfo($"Created filter for block: {heightToRequest}.");
+ Logger.LogInfo($"Created filter for block: {nextHeight}.");
}
else
{
- Logger.LogDebug($"Created filter for block: {heightToRequest}.");
+ Logger.LogDebug($"Created filter for block: {nextHeight}.");
}
}
catch (Exception ex)
@@ -308,8 +229,8 @@ namespace WalletWasabi.Blockchain.BlockFilters
}
finally
{
- Interlocked.CompareExchange(ref _running, 3, 2); // If IsStopping, make it stopped.
- Interlocked.Decrement(ref _runner);
+ Interlocked.CompareExchange(ref _serviceStatus, Stopped, Stopping); // If IsStopping, make it stopped.
+ Interlocked.Decrement(ref _workerCount);
}
}
catch (Exception ex)
@@ -319,25 +240,31 @@ namespace WalletWasabi.Blockchain.BlockFilters
});
}
- private async Task<SyncInfo> GetSyncInfoAsync()
+ private HashSet<Script> FetchScripts(VerboseBlockInfo block)
{
- var bcinfo = await RpcClient.GetBlockchainInfoAsync();
- var pbcinfo = new SyncInfo(bcinfo);
- return pbcinfo;
- }
+ var scripts = new HashSet<Script>();
- private void BlockNotifier_OnBlock(object sender, Block e)
+ foreach (var tx in block.Transactions)
{
- try
+ foreach (var input in tx.Inputs)
{
- // Run sync every time a block notification arrives. Synchronizer will stop when it finishes.
- Synchronize();
+ if (input.PrevOutput?.PubkeyType == RpcPubkeyType.TxWitnessV0Keyhash)
+ {
+ scripts.Add(input.PrevOutput.ScriptPubKey);
}
- catch (Exception ex)
+ }
+
+ foreach (var output in tx.Outputs)
{
- Logger.LogError(ex);
+ if (output.PubkeyType == RpcPubkeyType.TxWitnessV0Keyhash)
+ {
+ scripts.Add(output.ScriptPubKey);
}
}
+ }
+
+ return scripts;
+ }
private async Task ReorgOneAsync()
{
@@ -351,25 +278,26 @@ namespace WalletWasabi.Blockchain.BlockFilters
// 2. Serialize Index. (Remove last line.)
var lines = await File.ReadAllLinesAsync(IndexFilePath);
await File.WriteAllLinesAsync(IndexFilePath, lines.Take(lines.Length - 1).ToArray());
+ }
- // 3. Rollback Bech32UtxoSet
- if (Bech32UtxoSetHistory.Count != 0)
+ private async Task<SyncInfo> GetSyncInfoAsync()
{
- Bech32UtxoSetHistory.Last().Rollback(Bech32UtxoSet); // The Bech32UtxoSet MUST be recovered to its previous state.
- Bech32UtxoSetHistory.RemoveLast();
-
- // 4. Serialize Bech32UtxoSet.
- await File.WriteAllLinesAsync(Bech32UtxoSetFilePath, Bech32UtxoSet.Select(entry => entry.Value.Line));
- }
+ var bcinfo = await RpcClient.GetBlockchainInfoAsync();
+ var pbcinfo = new SyncInfo(bcinfo);
+ return pbcinfo;
}
- private void PrepareBech32UtxoSetHistory()
+ private void BlockNotifier_OnBlock(object sender, Block e)
+ {
+ try
{
- if (Bech32UtxoSetHistory.Count >= 100)
+ // Run sync every time a block notification arrives. Synchronizer will stop when it finishes.
+ Synchronize();
+ }
+ catch (Exception ex)
{
- Bech32UtxoSetHistory.RemoveFirst();
+ Logger.LogError(ex);
}
- Bech32UtxoSetHistory.Add(new ActionHistoryHelper());
}
public (Height bestHeight, IEnumerable<FilterModel> filters) GetFilterLinesExcluding(uint256 bestKnownBlockHash, int count, out bool found)
@@ -423,9 +351,9 @@ namespace WalletWasabi.Blockchain.BlockFilters
BlockNotifier.OnBlock -= BlockNotifier_OnBlock;
}
- Interlocked.CompareExchange(ref _running, 2, 1); // If running, make it stopping.
+ Interlocked.CompareExchange(ref _serviceStatus, Stopping, Running); // If running, make it stopping.
- while (Interlocked.CompareExchange(ref _running, 3, 0) == 2)
+ while (Interlocked.CompareExchange(ref _serviceStatus, Stopped, NotStarted) == 2)
{
await Task.Delay(50);
}

MODIFY
WalletWasabi/Blockchain/BlockFilters/SyncInfo.cs
WalletWasabi/Blockchain/BlockFilters/SyncInfo.cs
@@ -15,11 +15,14 @@ namespace WalletWasabi.Blockchain.BlockFilters
int headerCount = (int)bcinfo.Headers;
BlockchainInfoUpdated = DateTimeOffset.UtcNow;
IsCoreSynchornized = BlockCount == headerCount;
+ InitialBlockDownload = bcinfo.InitialBlockDownload;
}
public BlockchainInfo BlockchainInfo { get; }
public int BlockCount { get; }
public DateTimeOffset BlockchainInfoUpdated { get; }
public bool IsCoreSynchornized { get; }
+
+ public bool InitialBlockDownload { get; }
}
}

Change the filter index builder to use bitcoin knots getblockverbose rpc