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.
2026-06-11 13:37:31 +02:00
parent ece634f42e
commit 73a2d5356d
2 changed files with 118 additions and 61 deletions
@@ -35,9 +35,13 @@ public partial class MainWindowViewModel : ViewModelBase
    private string? _walletPath;
    private string? _password;
    private ElectrumClient? _client;
+    private WalletSynchronizer? _synchronizer;
    private IReadOnlyDictionary<string, Transaction>? _lastTransactions;
    private BuiltTransaction? _pendingSend;

+    /// <summary>Notifica arrivata durante una sync: si risincronizza appena finita.</summary>
+    private bool _resyncRequested;
+
    /// <summary>File in attesa di password (apertura da menu File → Apri).</summary>
    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;
@@ -40,6 +40,13 @@ public sealed class WalletSynchronizer(HdAccount account, ElectrumClient client,
    /// <summary>Avanzamento leggibile (per CLI e barra di stato GUI).</summary>
    public event Action<string>? 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<string, Transaction> _txCache = [];
+    private readonly Dictionary<string, int> _verifiedAtHeight = [];
+
    public async Task<SyncResult> 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<string, Transaction>();
-        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<string, bool>();
-        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,

    /// <summary>
    /// 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.
    /// </summary>
    private async Task<int> ScanChainAsync(bool isChange, List<TrackedAddress> tracked,
        Dictionary<string, IReadOnlyList<HistoryItem>> 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;
    }