From 73a2d5356dcedd77b30f752619f2dfc5cf46c54d Mon Sep 17 00:00:00 2001 From: Davide Grilli Date: Thu, 11 Jun 2026 13:37:31 +0200 Subject: [PATCH] feat(spv): Electrum-style continuous updates with incremental parallel sync WalletSynchronizer now pipelines JSON-RPC requests (address subscribes, histories, tx downloads and Merkle proofs run in parallel batches) and keeps per-connection caches of downloaded transactions and verified proofs, so notification-triggered resyncs only cost the delta. The app connects automatically when a wallet opens, keeps one synchronizer per connection, and queues notifications that arrive during a sync instead of dropping them: incoming transactions now show up within seconds. --- src/App/ViewModels/MainWindowViewModel.cs | 71 +++++++++----- src/Core/Spv/WalletSynchronizer.cs | 108 ++++++++++++++-------- 2 files changed, 118 insertions(+), 61 deletions(-) diff --git a/src/App/ViewModels/MainWindowViewModel.cs b/src/App/ViewModels/MainWindowViewModel.cs index 906e43c..5b39949 100644 --- a/src/App/ViewModels/MainWindowViewModel.cs +++ b/src/App/ViewModels/MainWindowViewModel.cs @@ -35,9 +35,13 @@ public partial class MainWindowViewModel : ViewModelBase private string? _walletPath; private string? _password; private ElectrumClient? _client; + private WalletSynchronizer? _synchronizer; private IReadOnlyDictionary? _lastTransactions; private BuiltTransaction? _pendingSend; + /// Notifica arrivata durante una sync: si risincronizza appena finita. + private bool _resyncRequested; + /// File in attesa di password (apertura da menu File → Apri). private string? _pendingOpenPath; @@ -308,9 +312,10 @@ public partial class MainWindowViewModel : ViewModelBase + (doc.IsWatchOnly ? " · watch-only" : ""); ApplyCache(doc.Cache); IsWalletOpen = true; - StatusMessage = doc.Cache is null - ? "Wallet aperto. Connettiti a un server per sincronizzare." - : "Wallet aperto (dati dell'ultima sincronizzazione)."; + StatusMessage = "Wallet aperto: connessione al server…"; + // Come Electrum: ci si connette da soli al server selezionato, + // senza aspettare un click. + _ = ConnectAndSync(); } private void ApplyCache(SyncCache? cache) @@ -364,6 +369,11 @@ public partial class MainWindowViewModel : ViewModelBase { if (_account is null || _doc is null) return; + if (IsSyncing) + { + _resyncRequested = true; + return; + } IsSyncing = true; StatusMessage = ""; try @@ -383,28 +393,36 @@ public partial class MainWindowViewModel : ViewModelBase IsConnected = true; _autoReconnect = true; ConnectionStatus = $"connesso a {host}:{port}{(UseSsl ? " (TLS)" : "")}"; + // Synchronizer per connessione: conserva la cache di tx e + // prove verificate, così le risincronizzazioni sono incrementali. + _synchronizer = new WalletSynchronizer(_account, _client, _doc.GapLimit); + _synchronizer.Progress += msg => Dispatcher.UIThread.Post(() => StatusMessage = msg); } - var sync = new WalletSynchronizer(_account, _client, _doc.GapLimit); - sync.Progress += msg => Dispatcher.UIThread.Post(() => StatusMessage = msg); - var result = await sync.SyncOnceAsync(); - _lastTransactions = result.Transactions; - - _doc.Cache = new SyncCache + // Se durante la sync arrivano notifiche, si ripete subito: nessun + // aggiornamento del server va perso (modello Electrum). + do { - TipHeight = result.TipHeight, - ConfirmedSats = result.ConfirmedSats, - UnconfirmedSats = result.UnconfirmedSats, - NextReceiveIndex = result.NextReceiveIndex, - NextChangeIndex = result.NextChangeIndex, - History = [.. result.History], - Utxos = [.. result.Utxos], - Addresses = [.. result.AddressRows], - }; - WalletStore.Save(_doc, _walletPath!, _password); - ApplyCache(_doc.Cache); - StatusMessage = $"Sincronizzato: altezza {result.TipHeight}, " + - $"{result.History.Count} transazioni verificate SPV."; + _resyncRequested = false; + var result = await _synchronizer!.SyncOnceAsync(); + _lastTransactions = result.Transactions; + + _doc.Cache = new SyncCache + { + TipHeight = result.TipHeight, + ConfirmedSats = result.ConfirmedSats, + UnconfirmedSats = result.UnconfirmedSats, + NextReceiveIndex = result.NextReceiveIndex, + NextChangeIndex = result.NextChangeIndex, + History = [.. result.History], + Utxos = [.. result.Utxos], + Addresses = [.. result.AddressRows], + }; + WalletStore.Save(_doc, _walletPath!, _password); + ApplyCache(_doc.Cache); + StatusMessage = $"Sincronizzato: altezza {result.TipHeight}, " + + $"{result.History.Count} transazioni verificate SPV. Aggiornamento in tempo reale attivo."; + } while (_resyncRequested); } catch (CertificatePinMismatchException ex) { @@ -453,10 +471,14 @@ public partial class MainWindowViewModel : ViewModelBase private void OnServerNotification(string method, System.Text.Json.JsonElement payload) { // Cambiamento su un nostro indirizzo o nuovo blocco: risincronizza. + // Se una sync è già in corso, si accoda (il loop in ConnectAndSync la + // ripete subito dopo): nessuna notifica viene persa. if (method is "blockchain.scripthash.subscribe" or "blockchain.headers.subscribe") Dispatcher.UIThread.Post(() => { - if (!IsSyncing) + if (IsSyncing) + _resyncRequested = true; + else _ = ConnectAndSync(); }); } @@ -541,6 +563,9 @@ public partial class MainWindowViewModel : ViewModelBase { _ = _client?.DisposeAsync().AsTask(); _client = null; + _synchronizer = null; + _autoReconnect = false; + _resyncRequested = false; _doc = null; _account = null; _lastTransactions = null; diff --git a/src/Core/Spv/WalletSynchronizer.cs b/src/Core/Spv/WalletSynchronizer.cs index 5ab741b..fb70016 100644 --- a/src/Core/Spv/WalletSynchronizer.cs +++ b/src/Core/Spv/WalletSynchronizer.cs @@ -40,6 +40,13 @@ public sealed class WalletSynchronizer(HdAccount account, ElectrumClient client, /// Avanzamento leggibile (per CLI e barra di stato GUI). public event Action? Progress; + // Cache tra le passate (stesso synchronizer per tutta la vita della + // connessione): le tx già scaricate e le prove di Merkle già verificate a + // una data altezza non si rifanno — le risincronizzazioni da notifica + // costano solo ciò che è cambiato (modello Electrum). + private readonly Dictionary _txCache = []; + private readonly Dictionary _verifiedAtHeight = []; + public async Task SyncOnceAsync(CancellationToken ct = default) { var tip = await client.SubscribeHeadersAsync(ct); @@ -56,37 +63,45 @@ public sealed class WalletSynchronizer(HdAccount account, ElectrumClient client, foreach (var item in historyByAddress.Values.SelectMany(h => h)) txHeights[item.TxHash] = item.Height; - // 4. Scarica le transazioni. - Progress?.Invoke($"scarico {txHeights.Count} transazioni…"); + // 4. Scarica in parallelo le sole transazioni nuove. var network = PalladiumNetworks.For(account.Profile.Kind); - var transactions = new Dictionary(); - foreach (var txid in txHeights.Keys) + var missing = txHeights.Keys.Where(txid => !_txCache.ContainsKey(txid)).ToList(); + if (missing.Count > 0) { - var hex = await client.GetTransactionAsync(txid, ct); - transactions[txid] = Transaction.Parse(hex, network); + Progress?.Invoke($"scarico {missing.Count} transazioni…"); + var downloaded = await Task.WhenAll(missing.Select(async txid => + (txid, Transaction.Parse(await client.GetTransactionAsync(txid, ct), network)))); + foreach (var (txid, tx) in downloaded) + _txCache[txid] = tx; } + var transactions = txHeights.Keys.ToDictionary(txid => txid, txid => _txCache[txid]); - // 5. Verifica Merkle delle confermate (§7.4 punto 4). - var verified = new Dictionary(); - foreach (var (txid, height) in txHeights) + // 5. Verifica Merkle delle confermate (§7.4 punto 4), in parallelo e + // solo per le tx non ancora verificate a quell'altezza. + var toVerify = txHeights + .Where(kv => kv.Value > 0 + && (!_verifiedAtHeight.TryGetValue(kv.Key, out var h) || h != kv.Value)) + .ToList(); + if (toVerify.Count > 0) { - if (height <= 0) + Progress?.Invoke($"verifico {toVerify.Count} prove di Merkle…"); + await Task.WhenAll(toVerify.Select(async kv => { - verified[txid] = false; // in mempool: nessuna prova possibile - continue; - } - var proof = await client.GetMerkleAsync(txid, height, ct); - var header = BlockHeaderInfo.Parse(await client.GetBlockHeaderAsync(height, ct)); - var ok = MerkleProof.Verify( - uint256.Parse(txid), - proof.Pos, - proof.Merkle.Select(uint256.Parse), - header.MerkleRoot); - if (!ok) - throw new SpvVerificationException( - $"Prova di Merkle non valida per {txid} (blocco {height}): server non affidabile."); - verified[txid] = true; + var (txid, height) = kv; + var proofTask = client.GetMerkleAsync(txid, height, ct); + var headerTask = client.GetBlockHeaderAsync(height, ct); + var proof = await proofTask; + var header = BlockHeaderInfo.Parse(await headerTask); + if (!MerkleProof.Verify( + uint256.Parse(txid), proof.Pos, + proof.Merkle.Select(uint256.Parse), header.MerkleRoot)) + throw new SpvVerificationException( + $"Prova di Merkle non valida per {txid} (blocco {height}): server non affidabile."); + })); + foreach (var (txid, height) in toVerify) + _verifiedAtHeight[txid] = height; } + var verified = txHeights.ToDictionary(kv => kv.Key, kv => kv.Value > 0); // 6. Ricostruzione locale degli UTXO: accrediti = output verso nostri // script; spesi = outpoint consumati da una qualunque tx del wallet. @@ -179,7 +194,9 @@ public sealed class WalletSynchronizer(HdAccount account, ElectrumClient client, /// /// Scansiona una catena (receiving o change) finché trova gapLimit indirizzi - /// vuoti consecutivi (§5). Ritorna il primo indice non usato. + /// vuoti consecutivi (§5), procedendo a batch paralleli di gapLimit + /// subscribe per volta (le richieste JSON-RPC sono pipelinabili). + /// Ritorna il primo indice non usato. /// private async Task ScanChainAsync(bool isChange, List tracked, Dictionary> historyByAddress, CancellationToken ct) @@ -187,23 +204,38 @@ public sealed class WalletSynchronizer(HdAccount account, ElectrumClient client, var consecutiveEmpty = 0; var index = 0; var firstUnused = 0; - for (; consecutiveEmpty < gapLimit; index++) + while (consecutiveEmpty < gapLimit) { - var address = account.GetAddress(isChange, index); - var scripthash = Scripthash.FromAddress(address); - tracked.Add(new TrackedAddress(address, scripthash, isChange, index)); + var batch = Enumerable.Range(index, gapLimit).Select(i => + { + var address = account.GetAddress(isChange, i); + return new TrackedAddress(address, Scripthash.FromAddress(address), isChange, i); + }).ToList(); + index += batch.Count; + tracked.AddRange(batch); // La subscribe registra anche la notifica push per i cambi futuri. - var status = await client.SubscribeScripthashAsync(scripthash, ct); - if (status is null) - { - consecutiveEmpty++; - continue; - } + var statuses = await Task.WhenAll( + batch.Select(t => client.SubscribeScripthashAsync(t.ScriptHash, ct))); - historyByAddress[scripthash] = await client.GetHistoryAsync(scripthash, ct); - consecutiveEmpty = 0; - firstUnused = index + 1; + var used = batch.Where((t, i) => statuses[i] is not null).ToList(); + var histories = await Task.WhenAll( + used.Select(t => client.GetHistoryAsync(t.ScriptHash, ct))); + for (var i = 0; i < used.Count; i++) + historyByAddress[used[i].ScriptHash] = histories[i]; + + for (var i = 0; i < batch.Count && consecutiveEmpty < gapLimit; i++) + { + if (statuses[i] is null) + { + consecutiveEmpty++; + } + else + { + consecutiveEmpty = 0; + firstUnused = batch[i].Index + 1; + } + } } return firstUnused; }