This commit is contained in:
2026-06-15 16:25:04 +02:00
parent 14ae39c5aa
commit d0ad9bb971
5 changed files with 355 additions and 205 deletions
@@ -221,6 +221,9 @@ public partial class MainWindowViewModel
_synchronizer.PreloadCaches( _synchronizer.PreloadCaches(
_doc.Cache?.RawTxHex ?? [], _doc.Cache?.RawTxHex ?? [],
_doc.Cache?.VerifiedAt ?? [], _doc.Cache?.VerifiedAt ?? [],
_doc.Cache?.BlockHeaders,
_doc.Cache?.NextReceiveIndex ?? 0,
_doc.Cache?.NextChangeIndex ?? 0,
net); net);
_synchronizer.Progress += msg => Dispatcher.UIThread.Post(() => StatusMessage = msg); _synchronizer.Progress += msg => Dispatcher.UIThread.Post(() => StatusMessage = msg);
} }
@@ -231,7 +234,7 @@ public partial class MainWindowViewModel
var result = await _synchronizer.SyncOnceAsync(); var result = await _synchronizer.SyncOnceAsync();
_lastTransactions = result.Transactions; _lastTransactions = result.Transactions;
var (rawHex, verifiedAt) = _synchronizer.ExportCaches( var (rawHex, verifiedAt, blockHeaders) = _synchronizer.ExportCaches(
PalladiumNetworks.For(_account.Profile.Kind)); PalladiumNetworks.For(_account.Profile.Kind));
_doc.Cache = new SyncCache _doc.Cache = new SyncCache
{ {
@@ -245,6 +248,7 @@ public partial class MainWindowViewModel
Addresses = [.. result.AddressRows], Addresses = [.. result.AddressRows],
RawTxHex = rawHex, RawTxHex = rawHex,
VerifiedAt = verifiedAt, VerifiedAt = verifiedAt,
BlockHeaders = blockHeaders,
}; };
WalletStore.Save(_doc, _walletPath!, _password); WalletStore.Save(_doc, _walletPath!, _password);
ApplyCache(_doc.Cache); ApplyCache(_doc.Cache);
@@ -334,11 +338,12 @@ public partial class MainWindowViewModel
try try
{ {
var net = PalladiumNetworks.For(_account.Profile.Kind); 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) if (rawHex.Count == 0 && verifiedAt.Count == 0)
return; return;
(_doc.Cache ??= new SyncCache()).RawTxHex = rawHex; (_doc.Cache ??= new SyncCache()).RawTxHex = rawHex;
_doc.Cache.VerifiedAt = verifiedAt; _doc.Cache.VerifiedAt = verifiedAt;
_doc.Cache.BlockHeaders = blockHeaders;
WalletStore.Save(_doc, _walletPath, _password); WalletStore.Save(_doc, _walletPath, _password);
} }
catch { /* non fatale: il prossimo salvataggio completo recupererà */ } catch { /* non fatale: il prossimo salvataggio completo recupererà */ }
+12
View File
@@ -131,8 +131,17 @@ static async Task<int> Sync(string[] o)
var sync = new WalletSynchronizer(account, client, doc.GapLimit); var sync = new WalletSynchronizer(account, client, doc.GapLimit);
sync.Progress += msg => Console.WriteLine($" {msg}"); 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 result = await sync.SyncOnceAsync();
var (rawHex, verifiedAt, blockHeaders) = sync.ExportCaches(net);
doc.Cache = new SyncCache doc.Cache = new SyncCache
{ {
TipHeight = result.TipHeight, TipHeight = result.TipHeight,
@@ -143,6 +152,9 @@ static async Task<int> Sync(string[] o)
History = [.. result.History], History = [.. result.History],
Utxos = [.. result.Utxos], Utxos = [.. result.Utxos],
Addresses = [.. result.AddressRows], Addresses = [.. result.AddressRows],
RawTxHex = rawHex,
VerifiedAt = verifiedAt,
BlockHeaders = blockHeaders,
}; };
WalletStore.Save(doc, path, Opt(o, "--password")); WalletStore.Save(doc, path, Opt(o, "--password"));
+130 -57
View File
@@ -1,8 +1,10 @@
using System.Buffers;
using System.Collections.Concurrent; using System.Collections.Concurrent;
using System.IO.Pipelines;
using System.Net.Security; using System.Net.Security;
using System.Net.Sockets; using System.Net.Sockets;
using System.Text;
using System.Text.Json; using System.Text.Json;
using System.Threading.Channels;
namespace PalladiumWallet.Core.Net; namespace PalladiumWallet.Core.Net;
@@ -18,10 +20,17 @@ public sealed class ElectrumClient : IAsyncDisposable
private readonly TcpClient _tcp; private readonly TcpClient _tcp;
private readonly Stream _stream; private readonly Stream _stream;
private readonly SemaphoreSlim _writeLock = new(1, 1);
private readonly ConcurrentDictionary<long, TaskCompletionSource<JsonElement>> _pending = new(); private readonly ConcurrentDictionary<long, TaskCompletionSource<JsonElement>> _pending = new();
private readonly CancellationTokenSource _cts = new(); private readonly CancellationTokenSource _cts = new();
private readonly Task _readLoop; 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<byte[]> _outgoing = Channel.CreateUnbounded<byte[]>(
new UnboundedChannelOptions { SingleReader = true, AllowSynchronousContinuations = false });
private long _nextId; private long _nextId;
public string Host { get; } public string Host { get; }
@@ -29,10 +38,7 @@ public sealed class ElectrumClient : IAsyncDisposable
public bool UseSsl { get; } public bool UseSsl { get; }
public bool IsConnected => _tcp.Connected && !_cts.IsCancellationRequested; public bool IsConnected => _tcp.Connected && !_cts.IsCancellationRequested;
/// <summary>(metodo, parametri) per le notifiche di subscription.</summary>
public event Action<string, JsonElement>? NotificationReceived; public event Action<string, JsonElement>? NotificationReceived;
/// <summary>Scatta quando la connessione cade (errore di lettura o chiusura remota).</summary>
public event Action<Exception?>? Disconnected; public event Action<Exception?>? Disconnected;
private ElectrumClient(TcpClient tcp, Stream stream, string host, int port, bool useSsl) private ElectrumClient(TcpClient tcp, Stream stream, string host, int port, bool useSsl)
@@ -42,18 +48,14 @@ public sealed class ElectrumClient : IAsyncDisposable
Host = host; Host = host;
Port = port; Port = port;
UseSsl = useSsl; UseSsl = useSsl;
_readLoop = Task.Run(ReadLoopAsync); _readLoop = Task.Run(ReadLoopAsync);
_writeLoop = Task.Run(WriteLoopAsync);
} }
/// <summary>
/// Connette al server. Con <paramref name="useSsl"/> la validazione del
/// certificato è TOFU tramite <paramref name="pins"/> (§9): primo contatto
/// salva, contatti successivi confrontano; mismatch ⇒ <see cref="CertificatePinMismatchException"/>.
/// </summary>
public static async Task<ElectrumClient> ConnectAsync(string host, int port, bool useSsl, public static async Task<ElectrumClient> ConnectAsync(string host, int port, bool useSsl,
CertificatePinStore? pins = null, CancellationToken ct = default) CertificatePinStore? pins = null, CancellationToken ct = default)
{ {
var tcp = new TcpClient(); var tcp = new TcpClient { NoDelay = true };
try try
{ {
await tcp.ConnectAsync(host, port, ct); await tcp.ConnectAsync(host, port, ct);
@@ -64,10 +66,7 @@ public sealed class ElectrumClient : IAsyncDisposable
var ssl = new SslStream(stream, leaveInnerStreamOpen: false, var ssl = new SslStream(stream, leaveInnerStreamOpen: false,
(_, cert, _, _) => (_, cert, _, _) =>
{ {
// I server Electrum sono tipicamente self-signed: la if (cert is null) return false;
// fiducia è il pin TOFU, non la catena CA (§9).
if (cert is null)
return false;
pinOk = pins is null || pins.VerifyOrPin(host, port, cert); pinOk = pins is null || pins.VerifyOrPin(host, port, cert);
return pinOk; return pinOk;
}); });
@@ -84,7 +83,6 @@ public sealed class ElectrumClient : IAsyncDisposable
} }
var client = new ElectrumClient(tcp, stream, host, port, useSsl); var client = new ElectrumClient(tcp, stream, host, port, useSsl);
// Negoziazione obbligatoria prima di ogni altra richiesta.
await client.RequestAsync("server.version", ct, ClientName, ProtocolVersion); await client.RequestAsync("server.version", ct, ClientName, ProtocolVersion);
return client; return client;
} }
@@ -110,63 +108,85 @@ public sealed class ElectrumClient : IAsyncDisposable
@params = parameters, @params = parameters,
}); });
await _writeLock.WaitAsync(ct); _outgoing.Writer.TryWrite(payload);
try
{
await _stream.WriteAsync(payload, ct);
await _stream.WriteAsync("\n"u8.ToArray(), ct);
await _stream.FlushAsync(ct);
}
finally
{
_writeLock.Release();
}
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; return await tcs.Task;
} }
private async Task ReadLoopAsync() /// <summary>
/// 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.
/// </summary>
private async Task WriteLoopAsync()
{ {
Exception? failure = null;
try try
{ {
using var reader = new StreamReader(_stream, Encoding.UTF8, leaveOpen: true); while (await _outgoing.Reader.WaitToReadAsync(_cts.Token))
while (!_cts.IsCancellationRequested)
{ {
var line = await reader.ReadLineAsync(_cts.Token); using var ms = new MemoryStream(512);
if (line is null) while (_outgoing.Reader.TryRead(out var data))
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)
{ {
if (!_pending.TryRemove(idEl.GetInt64(), out var tcs)) ms.Write(data);
continue; ms.WriteByte((byte)'\n');
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());
} }
await _stream.WriteAsync(ms.GetBuffer().AsMemory(0, (int)ms.Length), _cts.Token);
await _stream.FlushAsync(_cts.Token);
} }
} }
catch (OperationCanceledException) { } catch (OperationCanceledException) { }
catch (Exception ex) catch (Exception ex)
{
foreach (var (_, pending) in _pending)
pending.TrySetException(ex);
}
}
/// <summary>
/// Read loop con PipeReader: buffer pooled, zero allocazioni per-risposta
/// (nessuna stringa intermedia), parsing JSON direttamente da byte span.
/// </summary>
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; failure = ex;
} }
finally finally
{ {
await pipe.CompleteAsync();
foreach (var (_, tcs) in _pending) foreach (var (_, tcs) in _pending)
tcs.TrySetException(failure ?? new IOException("Connessione al server chiusa.")); tcs.TrySetException(failure ?? new IOException("Connessione al server chiusa."));
_pending.Clear(); _pending.Clear();
@@ -174,13 +194,66 @@ public sealed class ElectrumClient : IAsyncDisposable
} }
} }
private static bool TrySliceLine(ref ReadOnlySequence<byte> buffer,
out ReadOnlySequence<byte> 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<byte> 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<byte>.Shared.Rent(len);
try
{
line.CopyTo(buf);
DispatchSpan(buf.AsSpan(0, len));
}
finally { ArrayPool<byte>.Shared.Return(buf); }
}
private void DispatchSpan(ReadOnlySpan<byte> 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() public async ValueTask DisposeAsync()
{ {
await _cts.CancelAsync(); await _cts.CancelAsync();
_outgoing.Writer.Complete();
_tcp.Close(); _tcp.Close();
try { await _readLoop; } catch { /* in chiusura */ } try { await _readLoop; } catch { }
try { await _writeLoop; } catch { }
_cts.Dispose(); _cts.Dispose();
_writeLock.Dispose();
} }
} }
+199 -146
View File
@@ -31,46 +31,45 @@ public sealed class SyncResult
} }
/// <summary> /// <summary>
/// Sincronizzazione del wallet (blueprint §7.4): per ogni indirizzo calcola lo /// Sincronizzazione del wallet (blueprint §7.4).
/// 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).
/// </summary> /// </summary>
public sealed class WalletSynchronizer(IWalletAccount account, ElectrumClient client, int gapLimit = 20) public sealed class WalletSynchronizer(IWalletAccount account, ElectrumClient client, int gapLimit = 20)
{ {
/// <summary>Avanzamento leggibile (per CLI e barra di stato GUI).</summary> /// <summary>Avanzamento leggibile (per CLI e barra di stato GUI).</summary>
public event Action<string>? Progress; public event Action<string>? Progress;
// Richieste contemporanee verso il server. Troppo alte → -102 "server busy"; private readonly ConcurrentDictionary<string, Transaction> _txCache = new();
// 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<string, Transaction> _txCache = [];
private readonly Dictionary<string, int> _verifiedAtHeight = []; private readonly Dictionary<string, int> _verifiedAtHeight = [];
// Header grezzi per altezza: una Task<string> 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<int, Task<string>> _headerFetches = new(); private readonly ConcurrentDictionary<int, Task<string>> _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;
/// <summary> /// <summary>
/// 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. /// Chiamare prima di SyncOnceAsync per evitare di riscaricale le tx già note.
/// </summary> /// </summary>
public void PreloadCaches(Dictionary<string, string> rawTxHex, public void PreloadCaches(
Dictionary<string, int> verifiedAt, Network network) Dictionary<string, string> rawTxHex,
Dictionary<string, int> verifiedAt,
Dictionary<int, string>? blockHeaders,
int knownReceiveIndex,
int knownChangeIndex,
Network network)
{ {
foreach (var (txid, hex) in rawTxHex) foreach (var (txid, hex) in rawTxHex)
if (!_txCache.ContainsKey(txid)) _txCache.TryAdd(txid, Transaction.Parse(hex, network));
_txCache[txid] = Transaction.Parse(hex, network);
foreach (var (txid, height) in verifiedAt) foreach (var (txid, height) in verifiedAt)
if (!_verifiedAtHeight.ContainsKey(txid)) if (!_verifiedAtHeight.ContainsKey(txid))
_verifiedAtHeight[txid] = height; _verifiedAtHeight[txid] = height;
if (blockHeaders is not null)
foreach (var (height, hex) in blockHeaders)
_headerFetches.TryAdd(height, Task.FromResult(hex));
_knownReceiveIndex = knownReceiveIndex;
_knownChangeIndex = knownChangeIndex;
} }
/// <summary> /// <summary>
@@ -78,15 +77,23 @@ public sealed class WalletSynchronizer(IWalletAccount account, ElectrumClient cl
/// Solo le tx confermate (height > 0) vengono incluse: le non confermate /// Solo le tx confermate (height > 0) vengono incluse: le non confermate
/// possono cambiare (RBF) e vanno sempre riscaricate. /// possono cambiare (RBF) e vanno sempre riscaricate.
/// </summary> /// </summary>
public (Dictionary<string, string> RawTxHex, Dictionary<string, int> VerifiedAt) public (Dictionary<string, string> RawTxHex,
Dictionary<string, int> VerifiedAt,
Dictionary<int, string> BlockHeaders)
ExportCaches(Network network) 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 var rawHex = _verifiedAtHeight.Keys
.Where(_txCache.ContainsKey) .Where(_txCache.ContainsKey)
.ToDictionary(txid => txid, txid => _txCache[txid].ToHex()); .ToDictionary(txid => txid, txid => _txCache[txid].ToHex());
return (rawHex, new Dictionary<string, int>(_verifiedAtHeight));
// Solo gli header già completati: Task<string> non ancora completate
// non vengono persistite (verranno rifetchate al prossimo sync se necessario).
var headers = new Dictionary<int, string>();
foreach (var (height, task) in _headerFetches)
if (task.IsCompletedSuccessfully)
headers[height] = task.Result;
return (rawHex, new Dictionary<string, int>(_verifiedAtHeight), headers);
} }
public async Task<SyncResult> SyncOnceAsync(CancellationToken ct = default) public async Task<SyncResult> SyncOnceAsync(CancellationToken ct = default)
@@ -101,38 +108,41 @@ public sealed class WalletSynchronizer(IWalletAccount account, ElectrumClient cl
if (account.FixedAddresses is { } fixedAddresses) 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) foreach (var (addr, isChange, idx) in fixedAddresses)
tracked.Add(new TrackedAddress(addr, Scripthash.FromAddress(addr), isChange, idx)); tracked.Add(new TrackedAddress(addr, Scripthash.FromAddress(addr), isChange, idx));
nextReceive = tracked.Count(t => !t.IsChange); nextReceive = tracked.Count(t => !t.IsChange);
nextChange = 0; nextChange = 0;
var histories = await Task.WhenAll( await Task.WhenAll(tracked.Select(t => RetryOnBusyAsync(async () =>
tracked.Select(t => client.GetHistoryAsync(t.ScriptHash, ct)));
for (var i = 0; i < tracked.Count; i++)
{ {
if (histories[i].Count > 0) var h = await client.GetHistoryAsync(t.ScriptHash, ct);
historyByAddress[tracked[i].ScriptHash] = histories[i]; if (h.Count > 0) historyByAddress[t.ScriptHash] = h;
} }, ct)).Concat(tracked.Select(t =>
// Subscribe a tutti (pochi): notifiche push per ogni indirizzo importato. RetryOnBusyAsync(() => client.SubscribeScripthashAsync(t.ScriptHash, ct), ct))));
await Task.WhenAll(tracked.Select(t => client.SubscribeScripthashAsync(t.ScriptHash, ct)));
} }
else else
{ {
// HD: discovery con GetHistoryAsync (senza subscription → no -101 su wallet grandi); // Receive e change chain in parallelo (indipendenti per definizione).
// subscribe solo al gap window per ricevere notifiche push di nuove tx. // ScanChainAsync usa _knownReceiveIndex/_knownChangeIndex per la discovery
nextReceive = await ScanChainAsync(isChange: false, tracked, historyByAddress, ct); // incrementale: gli indirizzi già usati vengono fetchati in un burst unico.
nextChange = await ScanChainAsync(isChange: true, tracked, historyByAddress, ct); 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 => var gapAddresses = tracked.Where(t =>
(!t.IsChange && t.Index >= nextReceive && t.Index < nextReceive + gapLimit) || (!t.IsChange && t.Index >= nextReceive && t.Index < nextReceive + gapLimit) ||
( t.IsChange && t.Index >= nextChange && t.Index < nextChange + gapLimit)).ToList(); ( t.IsChange && t.Index >= nextChange && t.Index < nextChange + gapLimit)).ToList();
if (gapAddresses.Count > 0) 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). // 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)) foreach (var item in historyByAddress.Values.SelectMany(h => h))
txHeights[item.TxHash] = item.Height; txHeights[item.TxHash] = item.Height;
// 4. Scarica le transazioni nuove: semaforo MaxConcurrent per non saturare // 4+5. Download tx mancanti e verifica Merkle in parallelo senza semaforo.
// il server, con aggiornamento progresso in tempo reale. var network = PalladiumNetworks.For(account.Profile.Kind);
var network = PalladiumNetworks.For(account.Profile.Kind); var missing = txHeights.Keys.Where(txid => !_txCache.ContainsKey(txid)).ToList();
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.
var toVerify = txHeights var toVerify = txHeights
.Where(kv => kv.Value > 0 .Where(kv => kv.Value > 0
&& (!_verifiedAtHeight.TryGetValue(kv.Key, out var h) || h != kv.Value)) && (!_verifiedAtHeight.TryGetValue(kv.Key, out var h) || h != kv.Value))
.ToList(); .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; 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); var raw = await client.GetTransactionAsync(txid, ct);
try _txCache[txid] = Transaction.Parse(raw, network);
{ var n = Interlocked.Increment(ref dlDone);
var (txid, height) = kv; if (n % 50 == 0 || n == missing.Count)
// Proof e header in parallelo; l'header è condiviso per altezza. Progress?.Invoke($"tx {n}/{missing.Count}, prove {merkDone}/{toVerify.Count}…");
var proofTask = client.GetMerkleAsync(txid, height, ct); }, ct));
var headerTask = _headerFetches.GetOrAdd(height,
h => client.GetBlockHeaderAsync(h, ct)); var merkTasks = toVerify.Select(kv => RetryOnBusyAsync(async () =>
var proof = await proofTask; {
var header = BlockHeaderInfo.Parse(await headerTask); var (txid, height) = kv;
if (!MerkleProof.Verify( var proofTask = client.GetMerkleAsync(txid, height, ct);
uint256.Parse(txid), proof.Pos, var headerTask = _headerFetches.GetOrAdd(height,
proof.Merkle.Select(uint256.Parse), header.MerkleRoot)) h => client.GetBlockHeaderAsync(h, ct));
throw new SpvVerificationException( var proof = await proofTask;
$"Prova di Merkle non valida per {txid} (blocco {height}): server non affidabile."); var header = BlockHeaderInfo.Parse(await headerTask);
var n = Interlocked.Increment(ref merkDone); if (!MerkleProof.Verify(
Progress?.Invoke($"verifico {n}/{toVerify.Count} prove di Merkle…"); uint256.Parse(txid), proof.Pos,
} proof.Merkle.Select(uint256.Parse), header.MerkleRoot))
finally { merkSem.Release(); } 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) foreach (var (txid, height) in toVerify)
_verifiedAtHeight[txid] = height; _verifiedAtHeight[txid] = height;
} }
var verified = txHeights.ToDictionary(kv => kv.Key, kv => kv.Value > 0);
// 6. Ricostruzione locale degli UTXO: accrediti = output verso nostri var transactions = txHeights.Keys.ToDictionary(txid => txid, txid => _txCache[txid]);
// script; spesi = outpoint consumati da una qualunque tx del wallet. 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 byScript = tracked.ToDictionary(t => t.ScriptPubKey, t => t);
var spent = transactions.Values var spent = transactions.Values
.SelectMany(tx => tx.Inputs) .SelectMany(tx => tx.Inputs)
@@ -223,18 +218,18 @@ public sealed class WalletSynchronizer(IWalletAccount account, ElectrumClient cl
continue; continue;
utxos.Add(new CachedUtxo utxos.Add(new CachedUtxo
{ {
Txid = txid, Txid = txid,
Vout = vout, Vout = vout,
ValueSats = output.Value.Satoshi, ValueSats = output.Value.Satoshi,
Address = addr.Address.ToString(), Address = addr.Address.ToString(),
IsChange = addr.IsChange, IsChange = addr.IsChange,
AddressIndex = addr.Index, AddressIndex = addr.Index,
Height = txHeights[txid], Height = txHeights[txid],
}); });
} }
} }
// 7. Delta per voce di storico (entrate - uscite del wallet). // 7. Delta per voce di storico.
var history = new List<CachedTx>(); var history = new List<CachedTx>();
foreach (var (txid, tx) in transactions) 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); .Sum(i => transactions[i.PrevOut.Hash.ToString()].Outputs[i.PrevOut.N].Value.Satoshi);
history.Add(new CachedTx history.Add(new CachedTx
{ {
Txid = txid, Txid = txid,
Height = txHeights[txid], Height = txHeights[txid],
DeltaSats = received - sentSats, DeltaSats = received - sentSats,
Verified = verified[txid], Verified = verified[txid],
}); });
} }
history.Sort((a, b) => history.Sort((a, b) =>
{ {
// Non confermate (height<=0) in cima, poi per altezza decrescente.
var ha = a.Height <= 0 ? int.MaxValue : a.Height; var ha = a.Height <= 0 ? int.MaxValue : a.Height;
var hb = b.Height <= 0 ? int.MaxValue : b.Height; var hb = b.Height <= 0 ? int.MaxValue : b.Height;
return hb.CompareTo(ha); return hb.CompareTo(ha);
}); });
// Saldo e numero di transazioni per singolo indirizzo (vista indirizzi).
var balanceByAddress = utxos var balanceByAddress = utxos
.GroupBy(u => u.Address) .GroupBy(u => u.Address)
.ToDictionary(g => g.Key, g => g.Sum(u => u.ValueSats)); .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) .OrderBy(t => t.IsChange).ThenBy(t => t.Index)
.Select(t => new CachedAddress .Select(t => new CachedAddress
{ {
Address = t.Address.ToString(), Address = t.Address.ToString(),
IsChange = t.IsChange, IsChange = t.IsChange,
Index = t.Index, Index = t.Index,
BalanceSats = balanceByAddress.GetValueOrDefault(t.Address.ToString()), BalanceSats = balanceByAddress.GetValueOrDefault(t.Address.ToString()),
TxCount = historyByAddress.TryGetValue(t.ScriptHash, out var h) ? h.Count : 0, TxCount = historyByAddress.TryGetValue(t.ScriptHash, out var h) ? h.Count : 0,
}) })
.ToList(); .ToList();
return new SyncResult return new SyncResult
{ {
TipHeight = tip.Height, TipHeight = tip.Height,
ConfirmedSats = utxos.Where(u => u.Height > 0).Sum(u => u.ValueSats), ConfirmedSats = utxos.Where(u => u.Height > 0).Sum(u => u.ValueSats),
UnconfirmedSats = utxos.Where(u => u.Height <= 0).Sum(u => u.ValueSats), UnconfirmedSats = utxos.Where(u => u.Height <= 0).Sum(u => u.ValueSats),
NextReceiveIndex = nextReceive, NextReceiveIndex = nextReceive,
NextChangeIndex = nextChange, NextChangeIndex = nextChange,
History = history, History = history,
Utxos = utxos, Utxos = utxos,
Addresses = tracked, Addresses = tracked,
AddressRows = addressRows, AddressRows = addressRows,
Transactions = transactions, Transactions = transactions,
}; };
} }
/// <summary> /// <summary>
/// Scansiona una catena (receiving o change) finché trova gapLimit indirizzi /// Scansiona una catena (receiving o change).
/// vuoti consecutivi (§5), procedendo a batch paralleli di gapLimit per volta. ///
/// Usa GetHistoryAsync per la discovery — senza subscription → nessun rischio di /// Phase 1 — indirizzi noti (0..fromIndex-1): tutti i GetHistoryAsync partono
/// -101 "excessive resource usage" su wallet con molti indirizzi storici. /// in un unico burst parallelo, senza batching sequenziale. Per un wallet con
/// Le subscription per notifiche push vengono gestite dal chiamante (solo gap window). /// 100 indirizzi usati → 1 RTT invece di 5 round sequenziali di gapLimit.
/// Ritorna il primo indice non usato. ///
/// Phase 2 — discovery dal fromIndex in poi: batching con gap limit come prima,
/// necessario per sapere dove fermarsi.
/// </summary> /// </summary>
private async Task<int> ScanChainAsync(bool isChange, List<TrackedAddress> tracked, private async Task<(int NextIndex,
Dictionary<string, IReadOnlyList<HistoryItem>> historyByAddress, CancellationToken ct) List<TrackedAddress> Tracked,
Dictionary<string, IReadOnlyList<HistoryItem>> History)>
ScanChainAsync(bool isChange, int fromIndex, CancellationToken ct)
{ {
var tracked = new List<TrackedAddress>();
var history = new Dictionary<string, IReadOnlyList<HistoryItem>>();
// 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 consecutiveEmpty = 0;
var index = 0; var index = fromIndex;
var firstUnused = 0; var firstUnused = fromIndex;
while (consecutiveEmpty < gapLimit) while (consecutiveEmpty < gapLimit)
{ {
var batch = Enumerable.Range(index, gapLimit).Select(i => var batch = Enumerable.Range(index, gapLimit).Select(i =>
{ {
var address = account.GetAddress(isChange, i); var addr = account.GetAddress(isChange, i);
return new TrackedAddress(address, Scripthash.FromAddress(address), isChange, i); return new TrackedAddress(addr, Scripthash.FromAddress(addr), isChange, i);
}).ToList(); }).ToList();
index += batch.Count; index += batch.Count;
tracked.AddRange(batch); 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( 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++) for (var i = 0; i < batch.Count && consecutiveEmpty < gapLimit; i++)
{ {
var history = histories[i]; if (histories[i].Count == 0)
if (history.Count == 0)
{ {
consecutiveEmpty++; consecutiveEmpty++;
} }
@@ -332,12 +350,47 @@ public sealed class WalletSynchronizer(IWalletAccount account, ElectrumClient cl
{ {
consecutiveEmpty = 0; consecutiveEmpty = 0;
firstUnused = batch[i].Index + 1; 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<Task> 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<T> RetryOnBusyAsync<T>(Func<Task<T>> 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);
} }
/// <summary>La verifica SPV è fallita: i dati del server contraddicono le prove (§17).</summary> /// <summary>La verifica SPV è fallita: i dati del server contraddicono le prove (§17).</summary>
+7
View File
@@ -113,6 +113,13 @@ public sealed class SyncCache
/// riverificare le stesse prove ad ogni avvio: le conferme sono immutabili. /// riverificare le stesse prove ad ogni avvio: le conferme sono immutabili.
/// </summary> /// </summary>
public Dictionary<string, int>? VerifiedAt { get; set; } public Dictionary<string, int>? VerifiedAt { get; set; }
/// <summary>
/// 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.
/// </summary>
public Dictionary<int, string>? BlockHeaders { get; set; }
} }
/// <summary>Indirizzo scansionato con saldo proprio e numero di transazioni (vista indirizzi).</summary> /// <summary>Indirizzo scansionato con saldo proprio e numero di transazioni (vista indirizzi).</summary>