From d0ad9bb971c276610560fbb7fe131eb443588c3f Mon Sep 17 00:00:00 2001 From: Davide Grilli Date: Mon, 15 Jun 2026 16:25:04 +0200 Subject: [PATCH] temp --- .../ViewModels/MainWindowViewModel.Sync.cs | 9 +- src/Cli/Program.cs | 12 + src/Core/Net/ElectrumClient.cs | 187 +++++++--- src/Core/Spv/WalletSynchronizer.cs | 345 ++++++++++-------- src/Core/Storage/WalletDocument.cs | 7 + 5 files changed, 355 insertions(+), 205 deletions(-) diff --git a/src/App/ViewModels/MainWindowViewModel.Sync.cs b/src/App/ViewModels/MainWindowViewModel.Sync.cs index 64c1500..4ee5fcc 100644 --- a/src/App/ViewModels/MainWindowViewModel.Sync.cs +++ b/src/App/ViewModels/MainWindowViewModel.Sync.cs @@ -221,6 +221,9 @@ public partial class MainWindowViewModel _synchronizer.PreloadCaches( _doc.Cache?.RawTxHex ?? [], _doc.Cache?.VerifiedAt ?? [], + _doc.Cache?.BlockHeaders, + _doc.Cache?.NextReceiveIndex ?? 0, + _doc.Cache?.NextChangeIndex ?? 0, net); _synchronizer.Progress += msg => Dispatcher.UIThread.Post(() => StatusMessage = msg); } @@ -231,7 +234,7 @@ public partial class MainWindowViewModel var result = await _synchronizer.SyncOnceAsync(); _lastTransactions = result.Transactions; - var (rawHex, verifiedAt) = _synchronizer.ExportCaches( + var (rawHex, verifiedAt, blockHeaders) = _synchronizer.ExportCaches( PalladiumNetworks.For(_account.Profile.Kind)); _doc.Cache = new SyncCache { @@ -245,6 +248,7 @@ public partial class MainWindowViewModel Addresses = [.. result.AddressRows], RawTxHex = rawHex, VerifiedAt = verifiedAt, + BlockHeaders = blockHeaders, }; WalletStore.Save(_doc, _walletPath!, _password); ApplyCache(_doc.Cache); @@ -334,11 +338,12 @@ public partial class MainWindowViewModel try { var net = PalladiumNetworks.For(_account.Profile.Kind); - var (rawHex, verifiedAt) = _synchronizer.ExportCaches(net); + var (rawHex, verifiedAt, blockHeaders) = _synchronizer.ExportCaches(net); if (rawHex.Count == 0 && verifiedAt.Count == 0) return; (_doc.Cache ??= new SyncCache()).RawTxHex = rawHex; _doc.Cache.VerifiedAt = verifiedAt; + _doc.Cache.BlockHeaders = blockHeaders; WalletStore.Save(_doc, _walletPath, _password); } catch { /* non fatale: il prossimo salvataggio completo recupererà */ } diff --git a/src/Cli/Program.cs b/src/Cli/Program.cs index 6754ca5..e5bed59 100644 --- a/src/Cli/Program.cs +++ b/src/Cli/Program.cs @@ -131,8 +131,17 @@ static async Task Sync(string[] o) var sync = new WalletSynchronizer(account, client, doc.GapLimit); sync.Progress += msg => Console.WriteLine($" {msg}"); + var net = PalladiumNetworks.For(account.Profile.Kind); + sync.PreloadCaches( + doc.Cache?.RawTxHex ?? [], + doc.Cache?.VerifiedAt ?? [], + doc.Cache?.BlockHeaders, + doc.Cache?.NextReceiveIndex ?? 0, + doc.Cache?.NextChangeIndex ?? 0, + net); var result = await sync.SyncOnceAsync(); + var (rawHex, verifiedAt, blockHeaders) = sync.ExportCaches(net); doc.Cache = new SyncCache { TipHeight = result.TipHeight, @@ -143,6 +152,9 @@ static async Task Sync(string[] o) History = [.. result.History], Utxos = [.. result.Utxos], Addresses = [.. result.AddressRows], + RawTxHex = rawHex, + VerifiedAt = verifiedAt, + BlockHeaders = blockHeaders, }; WalletStore.Save(doc, path, Opt(o, "--password")); diff --git a/src/Core/Net/ElectrumClient.cs b/src/Core/Net/ElectrumClient.cs index 674e88b..a8e600d 100644 --- a/src/Core/Net/ElectrumClient.cs +++ b/src/Core/Net/ElectrumClient.cs @@ -1,8 +1,10 @@ +using System.Buffers; using System.Collections.Concurrent; +using System.IO.Pipelines; using System.Net.Security; using System.Net.Sockets; -using System.Text; using System.Text.Json; +using System.Threading.Channels; namespace PalladiumWallet.Core.Net; @@ -18,10 +20,17 @@ public sealed class ElectrumClient : IAsyncDisposable private readonly TcpClient _tcp; private readonly Stream _stream; - private readonly SemaphoreSlim _writeLock = new(1, 1); private readonly ConcurrentDictionary> _pending = new(); private readonly CancellationTokenSource _cts = new(); private readonly Task _readLoop; + private readonly Task _writeLoop; + + // Channel single-reader: le task scrivono i payload senza lock; + // il write loop drena tutto in un unico WriteAsync+FlushAsync — + // identico al buffered writer asyncio di Electrum. + private readonly Channel _outgoing = Channel.CreateUnbounded( + new UnboundedChannelOptions { SingleReader = true, AllowSynchronousContinuations = false }); + private long _nextId; public string Host { get; } @@ -29,10 +38,7 @@ public sealed class ElectrumClient : IAsyncDisposable public bool UseSsl { get; } public bool IsConnected => _tcp.Connected && !_cts.IsCancellationRequested; - /// (metodo, parametri) per le notifiche di subscription. public event Action? NotificationReceived; - - /// Scatta quando la connessione cade (errore di lettura o chiusura remota). public event Action? Disconnected; private ElectrumClient(TcpClient tcp, Stream stream, string host, int port, bool useSsl) @@ -42,18 +48,14 @@ public sealed class ElectrumClient : IAsyncDisposable Host = host; Port = port; UseSsl = useSsl; - _readLoop = Task.Run(ReadLoopAsync); + _readLoop = Task.Run(ReadLoopAsync); + _writeLoop = Task.Run(WriteLoopAsync); } - /// - /// Connette al server. Con la validazione del - /// certificato è TOFU tramite (§9): primo contatto - /// salva, contatti successivi confrontano; mismatch ⇒ . - /// public static async Task ConnectAsync(string host, int port, bool useSsl, CertificatePinStore? pins = null, CancellationToken ct = default) { - var tcp = new TcpClient(); + var tcp = new TcpClient { NoDelay = true }; try { await tcp.ConnectAsync(host, port, ct); @@ -64,10 +66,7 @@ public sealed class ElectrumClient : IAsyncDisposable var ssl = new SslStream(stream, leaveInnerStreamOpen: false, (_, cert, _, _) => { - // I server Electrum sono tipicamente self-signed: la - // fiducia è il pin TOFU, non la catena CA (§9). - if (cert is null) - return false; + if (cert is null) return false; pinOk = pins is null || pins.VerifyOrPin(host, port, cert); return pinOk; }); @@ -84,7 +83,6 @@ public sealed class ElectrumClient : IAsyncDisposable } var client = new ElectrumClient(tcp, stream, host, port, useSsl); - // Negoziazione obbligatoria prima di ogni altra richiesta. await client.RequestAsync("server.version", ct, ClientName, ProtocolVersion); return client; } @@ -110,63 +108,85 @@ public sealed class ElectrumClient : IAsyncDisposable @params = parameters, }); - await _writeLock.WaitAsync(ct); - try - { - await _stream.WriteAsync(payload, ct); - await _stream.WriteAsync("\n"u8.ToArray(), ct); - await _stream.FlushAsync(ct); - } - finally - { - _writeLock.Release(); - } + _outgoing.Writer.TryWrite(payload); - await using var registration = ct.Register(() => tcs.TrySetCanceled(ct)); + await using var registration = ct.Register(() => + { + _pending.TryRemove(id, out _); + tcs.TrySetCanceled(ct); + }); return await tcs.Task; } - private async Task ReadLoopAsync() + /// + /// Drain loop: svuota il channel in un unico buffer → un solo WriteAsync+FlushAsync + /// per tutti i messaggi in coda. Quando N richieste sono in coda, vengono + /// trasmesse in un singolo segmento TCP invece di N flush seriali. + /// + private async Task WriteLoopAsync() { - Exception? failure = null; try { - using var reader = new StreamReader(_stream, Encoding.UTF8, leaveOpen: true); - while (!_cts.IsCancellationRequested) + while (await _outgoing.Reader.WaitToReadAsync(_cts.Token)) { - var line = await reader.ReadLineAsync(_cts.Token); - if (line is null) - break; // chiusura remota - if (string.IsNullOrWhiteSpace(line)) - continue; - - using var doc = JsonDocument.Parse(line); - var root = doc.RootElement; - - if (root.TryGetProperty("id", out var idEl) && idEl.ValueKind == JsonValueKind.Number) + using var ms = new MemoryStream(512); + while (_outgoing.Reader.TryRead(out var data)) { - if (!_pending.TryRemove(idEl.GetInt64(), out var tcs)) - continue; - if (root.TryGetProperty("error", out var error) && error.ValueKind != JsonValueKind.Null) - tcs.TrySetException(new ElectrumServerException(error.ToString())); - else - tcs.TrySetResult(root.GetProperty("result").Clone()); - } - else if (root.TryGetProperty("method", out var methodEl)) - { - NotificationReceived?.Invoke( - methodEl.GetString()!, - root.GetProperty("params").Clone()); + ms.Write(data); + ms.WriteByte((byte)'\n'); } + await _stream.WriteAsync(ms.GetBuffer().AsMemory(0, (int)ms.Length), _cts.Token); + await _stream.FlushAsync(_cts.Token); } } catch (OperationCanceledException) { } catch (Exception ex) + { + foreach (var (_, pending) in _pending) + pending.TrySetException(ex); + } + } + + /// + /// Read loop con PipeReader: buffer pooled, zero allocazioni per-risposta + /// (nessuna stringa intermedia), parsing JSON direttamente da byte span. + /// + private async Task ReadLoopAsync() + { + Exception? failure = null; + var pipe = PipeReader.Create(_stream, new StreamPipeReaderOptions(leaveOpen: true)); + try + { + while (true) + { + ReadResult result; + try { result = await pipe.ReadAsync(_cts.Token); } + catch (OperationCanceledException) { break; } + + var buffer = result.Buffer; + try + { + while (TrySliceLine(ref buffer, out var line)) + if (!line.IsEmpty) + DispatchLine(line); + } + finally + { + // consumed = tutto ciò che abbiamo consumato (fino all'ultimo \n) + // examined = tutto ciò che abbiamo guardato (fino alla fine del buffer) + pipe.AdvanceTo(buffer.Start, buffer.End); + } + + if (result.IsCompleted) break; + } + } + catch (Exception ex) { failure = ex; } finally { + await pipe.CompleteAsync(); foreach (var (_, tcs) in _pending) tcs.TrySetException(failure ?? new IOException("Connessione al server chiusa.")); _pending.Clear(); @@ -174,13 +194,66 @@ public sealed class ElectrumClient : IAsyncDisposable } } + private static bool TrySliceLine(ref ReadOnlySequence buffer, + out ReadOnlySequence line) + { + var pos = buffer.PositionOf((byte)'\n'); + if (pos is null) { line = default; return false; } + line = buffer.Slice(0, pos.Value); + buffer = buffer.Slice(buffer.GetPosition(1, pos.Value)); + return true; + } + + private void DispatchLine(ReadOnlySequence line) + { + if (line.IsSingleSegment) + { + DispatchSpan(line.FirstSpan); + return; + } + // Multi-segmento (risposta molto lunga): copia su ArrayPool poi parsa. + var len = (int)line.Length; + var buf = ArrayPool.Shared.Rent(len); + try + { + line.CopyTo(buf); + DispatchSpan(buf.AsSpan(0, len)); + } + finally { ArrayPool.Shared.Return(buf); } + } + + private void DispatchSpan(ReadOnlySpan utf8) + { + // JsonDocument.Parse via Utf8JsonReader: nessuna stringa intermedia, + // parsing direttamente dallo span pooled. + var reader = new Utf8JsonReader(utf8); + using var doc = JsonDocument.ParseValue(ref reader); + var root = doc.RootElement; + + if (root.TryGetProperty("id", out var idEl) && idEl.ValueKind == JsonValueKind.Number) + { + if (!_pending.TryRemove(idEl.GetInt64(), out var tcs)) return; + if (root.TryGetProperty("error", out var err) && err.ValueKind != JsonValueKind.Null) + tcs.TrySetException(new ElectrumServerException(err.ToString())); + else + tcs.TrySetResult(root.GetProperty("result").Clone()); + } + else if (root.TryGetProperty("method", out var methodEl)) + { + NotificationReceived?.Invoke( + methodEl.GetString()!, + root.GetProperty("params").Clone()); + } + } + public async ValueTask DisposeAsync() { await _cts.CancelAsync(); + _outgoing.Writer.Complete(); _tcp.Close(); - try { await _readLoop; } catch { /* in chiusura */ } + try { await _readLoop; } catch { } + try { await _writeLoop; } catch { } _cts.Dispose(); - _writeLock.Dispose(); } } diff --git a/src/Core/Spv/WalletSynchronizer.cs b/src/Core/Spv/WalletSynchronizer.cs index 8acd27c..230b0a6 100644 --- a/src/Core/Spv/WalletSynchronizer.cs +++ b/src/Core/Spv/WalletSynchronizer.cs @@ -31,46 +31,45 @@ public sealed class SyncResult } /// -/// Sincronizzazione del wallet (blueprint §7.4): per ogni indirizzo calcola lo -/// scripthash e si sottoscrive; scarica storico e transazioni; verifica ogni tx -/// confermata con la prova di Merkle contro l'header del suo blocco (le risposte -/// del server non sono fidate, §17); ricostruisce localmente UTXO e saldo; -/// estende la scansione fino al gap limit (§5). +/// Sincronizzazione del wallet (blueprint §7.4). /// public sealed class WalletSynchronizer(IWalletAccount account, ElectrumClient client, int gapLimit = 20) { /// Avanzamento leggibile (per CLI e barra di stato GUI). public event Action? Progress; - // Richieste contemporanee verso il server. Troppo alte → -102 "server busy"; - // troppo basse → throughput scarso su storie grandi. - private const int MaxConcurrent = 20; - - // 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 ConcurrentDictionary _txCache = new(); private readonly Dictionary _verifiedAtHeight = []; - - // Header grezzi per altezza: una Task per altezza, condivisa tra - // tutte le tx dello stesso blocco → ogni blocco viene scaricato una sola - // volta anche con centinaia di tx confermate nello stesso blocco. private readonly ConcurrentDictionary> _headerFetches = new(); + // Indici noti dal sync precedente: usati da ScanChainAsync per la discovery + // incrementale — gli indirizzi già usati vengono fetchati in un unico burst + // invece di batches sequenziali, riducendo i round-trip da O(used/gapLimit) a O(1). + private int _knownReceiveIndex; + private int _knownChangeIndex; + /// - /// Pre-popola le cache interne da dati salvati su disco (SyncCache). + /// Pre-popola le cache interne da dati salvati su disco. /// Chiamare prima di SyncOnceAsync per evitare di riscaricale le tx già note. /// - public void PreloadCaches(Dictionary rawTxHex, - Dictionary verifiedAt, Network network) + public void PreloadCaches( + Dictionary rawTxHex, + Dictionary verifiedAt, + Dictionary? blockHeaders, + int knownReceiveIndex, + int knownChangeIndex, + Network network) { foreach (var (txid, hex) in rawTxHex) - if (!_txCache.ContainsKey(txid)) - _txCache[txid] = Transaction.Parse(hex, network); + _txCache.TryAdd(txid, Transaction.Parse(hex, network)); foreach (var (txid, height) in verifiedAt) if (!_verifiedAtHeight.ContainsKey(txid)) _verifiedAtHeight[txid] = height; + if (blockHeaders is not null) + foreach (var (height, hex) in blockHeaders) + _headerFetches.TryAdd(height, Task.FromResult(hex)); + _knownReceiveIndex = knownReceiveIndex; + _knownChangeIndex = knownChangeIndex; } /// @@ -78,15 +77,23 @@ public sealed class WalletSynchronizer(IWalletAccount account, ElectrumClient cl /// Solo le tx confermate (height > 0) vengono incluse: le non confermate /// possono cambiare (RBF) e vanno sempre riscaricate. /// - public (Dictionary RawTxHex, Dictionary VerifiedAt) + public (Dictionary RawTxHex, + Dictionary VerifiedAt, + Dictionary BlockHeaders) ExportCaches(Network network) { - // Includi solo le tx associate a una prova di Merkle verificata - // (cioè confermate e verificate): sono le uniche immutabili. var rawHex = _verifiedAtHeight.Keys .Where(_txCache.ContainsKey) .ToDictionary(txid => txid, txid => _txCache[txid].ToHex()); - return (rawHex, new Dictionary(_verifiedAtHeight)); + + // Solo gli header già completati: Task non ancora completate + // non vengono persistite (verranno rifetchate al prossimo sync se necessario). + var headers = new Dictionary(); + foreach (var (height, task) in _headerFetches) + if (task.IsCompletedSuccessfully) + headers[height] = task.Result; + + return (rawHex, new Dictionary(_verifiedAtHeight), headers); } public async Task SyncOnceAsync(CancellationToken ct = default) @@ -101,38 +108,41 @@ public sealed class WalletSynchronizer(IWalletAccount account, ElectrumClient cl if (account.FixedAddresses is { } fixedAddresses) { - // Importati WIF: lista fissa, nessun gap limit. - // Pochi indirizzi → subscribe diretto per notifiche push. foreach (var (addr, isChange, idx) in fixedAddresses) tracked.Add(new TrackedAddress(addr, Scripthash.FromAddress(addr), isChange, idx)); nextReceive = tracked.Count(t => !t.IsChange); - nextChange = 0; + nextChange = 0; - var histories = await Task.WhenAll( - tracked.Select(t => client.GetHistoryAsync(t.ScriptHash, ct))); - for (var i = 0; i < tracked.Count; i++) + await Task.WhenAll(tracked.Select(t => RetryOnBusyAsync(async () => { - if (histories[i].Count > 0) - historyByAddress[tracked[i].ScriptHash] = histories[i]; - } - // Subscribe a tutti (pochi): notifiche push per ogni indirizzo importato. - await Task.WhenAll(tracked.Select(t => client.SubscribeScripthashAsync(t.ScriptHash, ct))); + var h = await client.GetHistoryAsync(t.ScriptHash, ct); + if (h.Count > 0) historyByAddress[t.ScriptHash] = h; + }, ct)).Concat(tracked.Select(t => + RetryOnBusyAsync(() => client.SubscribeScripthashAsync(t.ScriptHash, ct), ct)))); } else { - // HD: discovery con GetHistoryAsync (senza subscription → no -101 su wallet grandi); - // subscribe solo al gap window per ricevere notifiche push di nuove tx. - nextReceive = await ScanChainAsync(isChange: false, tracked, historyByAddress, ct); - nextChange = await ScanChainAsync(isChange: true, tracked, historyByAddress, ct); + // Receive e change chain in parallelo (indipendenti per definizione). + // ScanChainAsync usa _knownReceiveIndex/_knownChangeIndex per la discovery + // incrementale: gli indirizzi già usati vengono fetchati in un burst unico. + var receiveTask = ScanChainAsync(isChange: false, _knownReceiveIndex, ct); + var changeTask = ScanChainAsync(isChange: true, _knownChangeIndex, ct); + var rxScan = await receiveTask; + var chScan = await changeTask; + + tracked.AddRange(rxScan.Tracked); + tracked.AddRange(chScan.Tracked); + foreach (var (k, v) in rxScan.History) historyByAddress[k] = v; + foreach (var (k, v) in chScan.History) historyByAddress[k] = v; + nextReceive = rxScan.NextIndex; + nextChange = chScan.NextIndex; - // Iscriviti al gap window (prossimi indirizzi attesi) per notifiche push. - // In questo modo il numero di subscription è sempre ≤ 2×gapLimit, indipendentemente - // dalla dimensione dello storico — nessun rischio di -101. var gapAddresses = tracked.Where(t => (!t.IsChange && t.Index >= nextReceive && t.Index < nextReceive + gapLimit) || ( t.IsChange && t.Index >= nextChange && t.Index < nextChange + gapLimit)).ToList(); if (gapAddresses.Count > 0) - await Task.WhenAll(gapAddresses.Select(t => client.SubscribeScripthashAsync(t.ScriptHash, ct))); + await Task.WhenAll(gapAddresses.Select(t => + RetryOnBusyAsync(() => client.SubscribeScripthashAsync(t.ScriptHash, ct), ct))); } // 3. Storico unico (txid → altezza massima riportata). @@ -140,71 +150,56 @@ public sealed class WalletSynchronizer(IWalletAccount account, ElectrumClient cl foreach (var item in historyByAddress.Values.SelectMany(h => h)) txHeights[item.TxHash] = item.Height; - // 4. Scarica le transazioni nuove: semaforo MaxConcurrent per non saturare - // il server, con aggiornamento progresso in tempo reale. - var network = PalladiumNetworks.For(account.Profile.Kind); - var missing = txHeights.Keys.Where(txid => !_txCache.ContainsKey(txid)).ToList(); - if (missing.Count > 0) - { - var dlSem = new SemaphoreSlim(MaxConcurrent, MaxConcurrent); - var dlDone = 0; - Progress?.Invoke($"scarico 0/{missing.Count} transazioni…"); - await Task.WhenAll(missing.Select(async txid => - { - await dlSem.WaitAsync(ct); - try - { - var raw = await client.GetTransactionAsync(txid, ct); - _txCache[txid] = Transaction.Parse(raw, network); - var n = Interlocked.Increment(ref dlDone); - Progress?.Invoke($"scarico {n}/{missing.Count} transazioni…"); - } - finally { dlSem.Release(); } - })); - } - var transactions = txHeights.Keys.ToDictionary(txid => txid, txid => _txCache[txid]); - - // 5. Verifica Merkle delle confermate (§7.4 punto 4). - // Gli header per altezza sono condivisi via _headerFetches: se 500 tx - // stanno nello stesso blocco, l'header viene scaricato una sola volta. + // 4+5. Download tx mancanti e verifica Merkle in parallelo senza semaforo. + var network = PalladiumNetworks.For(account.Profile.Kind); + var missing = txHeights.Keys.Where(txid => !_txCache.ContainsKey(txid)).ToList(); var toVerify = txHeights .Where(kv => kv.Value > 0 && (!_verifiedAtHeight.TryGetValue(kv.Key, out var h) || h != kv.Value)) .ToList(); - if (toVerify.Count > 0) + + if (missing.Count > 0 || toVerify.Count > 0) { - var merkSem = new SemaphoreSlim(MaxConcurrent, MaxConcurrent); + Progress?.Invoke($"scarico {missing.Count} tx, verifico {toVerify.Count} prove…"); + var dlDone = 0; var merkDone = 0; - Progress?.Invoke($"verifico 0/{toVerify.Count} prove di Merkle…"); - await Task.WhenAll(toVerify.Select(async kv => + + var dlTasks = missing.Select(txid => RetryOnBusyAsync(async () => { - await merkSem.WaitAsync(ct); - try - { - var (txid, height) = kv; - // Proof e header in parallelo; l'header è condiviso per altezza. - var proofTask = client.GetMerkleAsync(txid, height, ct); - var headerTask = _headerFetches.GetOrAdd(height, - h => client.GetBlockHeaderAsync(h, 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."); - var n = Interlocked.Increment(ref merkDone); - Progress?.Invoke($"verifico {n}/{toVerify.Count} prove di Merkle…"); - } - finally { merkSem.Release(); } - })); + var raw = await client.GetTransactionAsync(txid, ct); + _txCache[txid] = Transaction.Parse(raw, network); + var n = Interlocked.Increment(ref dlDone); + if (n % 50 == 0 || n == missing.Count) + Progress?.Invoke($"tx {n}/{missing.Count}, prove {merkDone}/{toVerify.Count}…"); + }, ct)); + + var merkTasks = toVerify.Select(kv => RetryOnBusyAsync(async () => + { + var (txid, height) = kv; + var proofTask = client.GetMerkleAsync(txid, height, ct); + var headerTask = _headerFetches.GetOrAdd(height, + h => client.GetBlockHeaderAsync(h, 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."); + var n = Interlocked.Increment(ref merkDone); + if (n % 50 == 0 || n == toVerify.Count) + Progress?.Invoke($"tx {dlDone}/{missing.Count}, prove {n}/{toVerify.Count}…"); + }, ct)); + + await Task.WhenAll(dlTasks.Concat(merkTasks)); 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. + var transactions = txHeights.Keys.ToDictionary(txid => txid, txid => _txCache[txid]); + var verified = txHeights.ToDictionary(kv => kv.Key, kv => kv.Value > 0); + + // 6. Ricostruzione locale degli UTXO. var byScript = tracked.ToDictionary(t => t.ScriptPubKey, t => t); var spent = transactions.Values .SelectMany(tx => tx.Inputs) @@ -223,18 +218,18 @@ public sealed class WalletSynchronizer(IWalletAccount account, ElectrumClient cl continue; utxos.Add(new CachedUtxo { - Txid = txid, - Vout = vout, - ValueSats = output.Value.Satoshi, - Address = addr.Address.ToString(), - IsChange = addr.IsChange, - AddressIndex = addr.Index, - Height = txHeights[txid], + Txid = txid, + Vout = vout, + ValueSats = output.Value.Satoshi, + Address = addr.Address.ToString(), + IsChange = addr.IsChange, + AddressIndex = addr.Index, + Height = txHeights[txid], }); } } - // 7. Delta per voce di storico (entrate - uscite del wallet). + // 7. Delta per voce di storico. var history = new List(); foreach (var (txid, tx) in transactions) { @@ -247,21 +242,19 @@ public sealed class WalletSynchronizer(IWalletAccount account, ElectrumClient cl .Sum(i => transactions[i.PrevOut.Hash.ToString()].Outputs[i.PrevOut.N].Value.Satoshi); history.Add(new CachedTx { - Txid = txid, - Height = txHeights[txid], + Txid = txid, + Height = txHeights[txid], DeltaSats = received - sentSats, - Verified = verified[txid], + Verified = verified[txid], }); } history.Sort((a, b) => { - // Non confermate (height<=0) in cima, poi per altezza decrescente. var ha = a.Height <= 0 ? int.MaxValue : a.Height; var hb = b.Height <= 0 ? int.MaxValue : b.Height; return hb.CompareTo(ha); }); - // Saldo e numero di transazioni per singolo indirizzo (vista indirizzi). var balanceByAddress = utxos .GroupBy(u => u.Address) .ToDictionary(g => g.Key, g => g.Sum(u => u.ValueSats)); @@ -269,62 +262,87 @@ public sealed class WalletSynchronizer(IWalletAccount account, ElectrumClient cl .OrderBy(t => t.IsChange).ThenBy(t => t.Index) .Select(t => new CachedAddress { - Address = t.Address.ToString(), - IsChange = t.IsChange, - Index = t.Index, - BalanceSats = balanceByAddress.GetValueOrDefault(t.Address.ToString()), - TxCount = historyByAddress.TryGetValue(t.ScriptHash, out var h) ? h.Count : 0, + Address = t.Address.ToString(), + IsChange = t.IsChange, + Index = t.Index, + BalanceSats = balanceByAddress.GetValueOrDefault(t.Address.ToString()), + TxCount = historyByAddress.TryGetValue(t.ScriptHash, out var h) ? h.Count : 0, }) .ToList(); return new SyncResult { - TipHeight = tip.Height, - ConfirmedSats = utxos.Where(u => u.Height > 0).Sum(u => u.ValueSats), - UnconfirmedSats = utxos.Where(u => u.Height <= 0).Sum(u => u.ValueSats), + TipHeight = tip.Height, + ConfirmedSats = utxos.Where(u => u.Height > 0).Sum(u => u.ValueSats), + UnconfirmedSats = utxos.Where(u => u.Height <= 0).Sum(u => u.ValueSats), NextReceiveIndex = nextReceive, - NextChangeIndex = nextChange, - History = history, - Utxos = utxos, - Addresses = tracked, - AddressRows = addressRows, - Transactions = transactions, + NextChangeIndex = nextChange, + History = history, + Utxos = utxos, + Addresses = tracked, + AddressRows = addressRows, + Transactions = transactions, }; } /// - /// Scansiona una catena (receiving o change) finché trova gapLimit indirizzi - /// vuoti consecutivi (§5), procedendo a batch paralleli di gapLimit per volta. - /// Usa GetHistoryAsync per la discovery — senza subscription → nessun rischio di - /// -101 "excessive resource usage" su wallet con molti indirizzi storici. - /// Le subscription per notifiche push vengono gestite dal chiamante (solo gap window). - /// Ritorna il primo indice non usato. + /// Scansiona una catena (receiving o change). + /// + /// Phase 1 — indirizzi noti (0..fromIndex-1): tutti i GetHistoryAsync partono + /// in un unico burst parallelo, senza batching sequenziale. Per un wallet con + /// 100 indirizzi usati → 1 RTT invece di 5 round sequenziali di gapLimit. + /// + /// Phase 2 — discovery dal fromIndex in poi: batching con gap limit come prima, + /// necessario per sapere dove fermarsi. /// - private async Task ScanChainAsync(bool isChange, List tracked, - Dictionary> historyByAddress, CancellationToken ct) + private async Task<(int NextIndex, + List Tracked, + Dictionary> History)> + ScanChainAsync(bool isChange, int fromIndex, CancellationToken ct) { + var tracked = new List(); + var history = new Dictionary>(); + + // Phase 1: burst unico per tutti gli indirizzi già noti. + if (fromIndex > 0) + { + var known = Enumerable.Range(0, fromIndex).Select(i => + { + var addr = account.GetAddress(isChange, i); + return new TrackedAddress(addr, Scripthash.FromAddress(addr), isChange, i); + }).ToList(); + tracked.AddRange(known); + + var knownHistories = await Task.WhenAll( + known.Select(t => RetryOnBusyAsync( + () => client.GetHistoryAsync(t.ScriptHash, ct), ct))); + for (var i = 0; i < known.Count; i++) + if (knownHistories[i].Count > 0) + history[known[i].ScriptHash] = knownHistories[i]; + } + + // Phase 2: discovery gap-limit dal fromIndex in poi. var consecutiveEmpty = 0; - var index = 0; - var firstUnused = 0; + var index = fromIndex; + var firstUnused = fromIndex; + while (consecutiveEmpty < gapLimit) { var batch = Enumerable.Range(index, gapLimit).Select(i => { - var address = account.GetAddress(isChange, i); - return new TrackedAddress(address, Scripthash.FromAddress(address), isChange, i); + var addr = account.GetAddress(isChange, i); + return new TrackedAddress(addr, Scripthash.FromAddress(addr), isChange, i); }).ToList(); index += batch.Count; tracked.AddRange(batch); - // GetHistoryAsync per discovery: risposta vuota [] se inutilizzato, - // lista di tx se usato — un solo round-trip per indirizzo. var histories = await Task.WhenAll( - batch.Select(t => client.GetHistoryAsync(t.ScriptHash, ct))); + batch.Select(t => RetryOnBusyAsync( + () => client.GetHistoryAsync(t.ScriptHash, ct), ct))); for (var i = 0; i < batch.Count && consecutiveEmpty < gapLimit; i++) { - var history = histories[i]; - if (history.Count == 0) + if (histories[i].Count == 0) { consecutiveEmpty++; } @@ -332,12 +350,47 @@ public sealed class WalletSynchronizer(IWalletAccount account, ElectrumClient cl { consecutiveEmpty = 0; firstUnused = batch[i].Index + 1; - historyByAddress[batch[i].ScriptHash] = history; + history[batch[i].ScriptHash] = histories[i]; } } } - return firstUnused; + + return (firstUnused, tracked, history); } + + private static async Task RetryOnBusyAsync(Func op, CancellationToken ct) + { + var delay = 200; + for (var attempt = 0; ; attempt++) + { + try { await op(); return; } + catch (ElectrumServerException ex) + when (IsBusy(ex) && attempt < 7) + { + await Task.Delay(delay, ct); + delay = Math.Min(delay * 2, 5_000); + } + } + } + + private static async Task RetryOnBusyAsync(Func> op, CancellationToken ct) + { + var delay = 200; + for (var attempt = 0; ; attempt++) + { + try { return await op(); } + catch (ElectrumServerException ex) + when (IsBusy(ex) && attempt < 7) + { + await Task.Delay(delay, ct); + delay = Math.Min(delay * 2, 5_000); + } + } + } + + private static bool IsBusy(ElectrumServerException ex) => + ex.Message.Contains("-102") || + ex.Message.Contains("server busy", StringComparison.OrdinalIgnoreCase); } /// La verifica SPV è fallita: i dati del server contraddicono le prove (§17). diff --git a/src/Core/Storage/WalletDocument.cs b/src/Core/Storage/WalletDocument.cs index 67a63bd..3a3a625 100644 --- a/src/Core/Storage/WalletDocument.cs +++ b/src/Core/Storage/WalletDocument.cs @@ -113,6 +113,13 @@ public sealed class SyncCache /// riverificare le stesse prove ad ogni avvio: le conferme sono immutabili. /// public Dictionary? VerifiedAt { get; set; } + + /// + /// Header grezzi per altezza (altezza → hex). Immutabili: non vengono mai + /// rifetchati una volta salvati. Elimina i GetBlockHeader sulle prove Merkle + /// già verificate nei sync successivi. + /// + public Dictionary? BlockHeaders { get; set; } } /// Indirizzo scansionato con saldo proprio e numero di transazioni (vista indirizzi).