Merge pull request WalletWasabi#12939 from turbolay/reorgBlockFilterI…

…terator

Remove filters from BlockFilterIterator after a reorg

WalletWasabi/Wallets/FilterProcessor/BlockFilterIterator.cs

@@ -3,6 +3,7 @@
 using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
+using Nito.AsyncEx;
 using WalletWasabi.Backend.Models;
 using WalletWasabi.Stores;
 
@@ -20,7 +21,10 @@ public BlockFilterIterator(IIndexStore indexStore, int maxNumberFiltersInMemory
 	}
 
 	/// <remarks>Internal only to allow modifications in tests.</remarks>
-	internal Dictionary<uint, FilterModel> Cache { get; } = new();
+	internal Dictionary<uint, FilterModel> Cache { get; } = [];
+
+	/// <remarks>Lock object to guard <see cref="Cache"/>.</remarks>
+	private AsyncLock Lock { get; } = new();
 	private IIndexStore IndexStore { get; }
 	public int MaxNumberFiltersInMemory { get; }
 
@@ -30,14 +34,16 @@ public BlockFilterIterator(IIndexStore indexStore, int maxNumberFiltersInMemory
 	/// <remarks>Filter is immediately removed from the cache once the method returns. Repeated calls for single height are thus expensive.</remarks>
 	public async Task<FilterModel> GetAndRemoveAsync(uint height, CancellationToken cancellationToken)
 	{
+		using IDisposable _ = await Lock.LockAsync(cancellationToken).ConfigureAwait(false);
+
 		// Each block filter is to needed just once, so we can remove the block right now and free the memory sooner.
 		if (Cache.Remove(height, out FilterModel? result))
 		{
 			return result;
 		}
 
 		// We don't have the next filter to process, so fetch another batch of filters from the database.
-		Clear();
+		ClearNoLock();
 
 		FilterModel[] filtersBatch = await IndexStore.FetchBatchAsync(height, MaxNumberFiltersInMemory, cancellationToken).ConfigureAwait(false);
 
@@ -73,7 +79,32 @@ public async Task<FilterModel> GetAndRemoveAsync(uint height, CancellationToken
 		return result;
 	}
 
-	public void Clear()
+	public async Task RemoveNewerThanAsync(uint height, CancellationToken cancellationToken)
+	{
+		using IDisposable _ = await Lock.LockAsync(cancellationToken).ConfigureAwait(false);
+
+		List<uint> keysToRemove = Cache.Keys
+			.Where(key => key > height)
+			.ToList();
+
+		foreach (uint heightToRemove in keysToRemove)
+		{
+			if (!Cache.Remove(heightToRemove))
+			{
+				throw new UnreachableException($"Filter {heightToRemove} was already removed from the Cache.");
+			}
+		}
+	}
+
+	public async Task ClearAsync(CancellationToken cancellationToken)
+	{
+		using IDisposable _ = await Lock.LockAsync(cancellationToken).ConfigureAwait(false);
+
+		ClearNoLock();
+	}
+
+	/// <remarks>Needs to be guarded by <see cref="Lock"/></remarks>
+	private void ClearNoLock()
 	{
 		Cache.Clear();
 	}

WalletWasabi/Wallets/WalletFilterProcessor.cs

@@ -175,7 +175,7 @@ protected override async Task ExecuteAsync(CancellationToken cancellationToken)
 				{
 					foreach (SyncType syncType in Enum.GetValues<SyncType>())
 					{
-						FilterIteratorsBySyncType[syncType].Clear();
+						await FilterIteratorsBySyncType[syncType].ClearAsync(cancellationToken).ConfigureAwait(false);
 					}
 				}
 			}
@@ -265,13 +265,17 @@ private async void ReorgedAsync(object? sender, FilterModel invalidFilter)
 		try
 		{
 			uint256 invalidBlockHash = invalidFilter.Header.BlockHash;
+			uint newBestHeight = invalidFilter.Header.Height - 1;
 
 			using (await ReorgLock.LockAsync(CancellationToken.None).ConfigureAwait(false))
 			{
-				KeyManager.SetMaxBestHeight(new Height(invalidFilter.Header.Height - 1));
+				KeyManager.SetMaxBestHeight(new Height(newBestHeight));
 				TransactionProcessor.UndoBlock((int)invalidFilter.Header.Height);
 				BitcoinStore.TransactionStore.ReleaseToMempoolFromBlock(invalidBlockHash);
-
+				foreach (SyncType syncType in Enum.GetValues<SyncType>())
+				{
+					await FilterIteratorsBySyncType[syncType].RemoveNewerThanAsync(newBestHeight, CancellationToken.None).ConfigureAwait(false);
+				}
 				await BlockDownloadService.RemoveBlocksAsync(invalidFilter.Header.Height).ConfigureAwait(false);
 			}
 		}