perf(net): batched writes, zero-alloc reads, bounded in-flight requests

Rework the ElectrumClient transport to cut latency and allocations
during sync:

- Channel-based single-reader write loop: drains the whole queue into a
  single WriteAsync+FlushAsync, replacing the _writeLock that forced one
  flush per message. N queued requests now travel in one TCP segment.
- PipeReader read loop parsing via Utf8JsonReader over pooled spans: no
  StreamReader/ReadLineAsync and no intermediate string per response.
- TcpClient.NoDelay = true: no Nagle wait on small packets.
- Concurrency gate (SemaphoreSlim, MaxInFlight=32) in RequestAsync:
  caps requests in flight to the server without serializing writes,
  avoiding floods (-101/-102) and connection drops on wallets with large
  history.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

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,25 @@ public sealed class ElectrumClient : IAsyncDisposable
 
     private readonly TcpClient _tcp;
     private readonly Stream _stream;
-    private readonly SemaphoreSlim _writeLock = new(1, 1);
     private readonly ConcurrentDictionary<long, TaskCompletionSource<JsonElement>> _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<byte[]> _outgoing = Channel.CreateUnbounded<byte[]>(
+        new UnboundedChannelOptions { SingleReader = true, AllowSynchronousContinuations = false });
+
+    // Tetto alle richieste in volo (non in coda di scrittura, ma in attesa di
+    // risposta sul server). Il write-loop batcha già l'invio in un solo segmento;
+    // questo gate evita di sommergere il server con migliaia di richieste
+    // simultanee su wallet grandi → niente -101/-102 a raffica né drop della
+    // connessione. Le scritture restano comunque pipelinate fino a questo grado.
+    private const int MaxInFlight = 32;
+    private readonly SemaphoreSlim _inFlight = new(MaxInFlight, MaxInFlight);
+
     private long _nextId;
 
     public string Host { get; }
@@ -29,10 +46,7 @@ public sealed class ElectrumClient : IAsyncDisposable
     public bool UseSsl { get; }
     public bool IsConnected => _tcp.Connected && !_cts.IsCancellationRequested;
 
-    /// <summary>(metodo, parametri) per le notifiche di subscription.</summary>
     public event Action<string, JsonElement>? NotificationReceived;
-
-    /// <summary>Scatta quando la connessione cade (errore di lettura o chiusura remota).</summary>
     public event Action<Exception?>? Disconnected;
 
     private ElectrumClient(TcpClient tcp, Stream stream, string host, int port, bool useSsl)
@@ -43,17 +57,13 @@ public sealed class ElectrumClient : IAsyncDisposable
         Port = port;
         UseSsl = useSsl;
         _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,
         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 +74,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 +91,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;
         }
@@ -97,6 +103,11 @@ public sealed class ElectrumClient : IAsyncDisposable
 
     public async Task<JsonElement> RequestAsync(string method, CancellationToken ct = default,
         params object?[] parameters)
+    {
+        // Gate prima di mettere in volo: oltre MaxInFlight richieste in attesa
+        // si attende che una risposta liberi uno slot, invece di sommergere il server.
+        await _inFlight.WaitAsync(ct);
+        try
         {
             var id = Interlocked.Increment(ref _nextId);
             var tcs = new TaskCompletionSource<JsonElement>(TaskCreationOptions.RunContinuationsAsynchronously);
@@ -110,45 +121,138 @@ public sealed class ElectrumClient : IAsyncDisposable
                 @params = parameters,
             });
 
-        await _writeLock.WaitAsync(ct);
-        try
+            _outgoing.Writer.TryWrite(payload);
+
+            await using var registration = ct.Register(() =>
             {
-            await _stream.WriteAsync(payload, ct);
-            await _stream.WriteAsync("\n"u8.ToArray(), ct);
-            await _stream.FlushAsync(ct);
+                _pending.TryRemove(id, out _);
+                tcs.TrySetCanceled(ct);
+            });
+            return await tcs.Task;
         }
         finally
         {
-            _writeLock.Release();
+            _inFlight.Release();
+        }
     }
 
-        await using var registration = ct.Register(() => tcs.TrySetCanceled(ct));
-        return await tcs.Task;
+    /// <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()
+    {
+        try
+        {
+            while (await _outgoing.Reader.WaitToReadAsync(_cts.Token))
+            {
+                using var ms = new MemoryStream(512);
+                while (_outgoing.Reader.TryRead(out var data))
+                {
+                    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);
+        }
     }
 
+    /// <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
         {
-            using var reader = new StreamReader(_stream, Encoding.UTF8, leaveOpen: true);
-            while (!_cts.IsCancellationRequested)
+            while (true)
             {
-                var line = await reader.ReadLineAsync(_cts.Token);
-                if (line is null)
-                    break; // chiusura remota
-                if (string.IsNullOrWhiteSpace(line))
-                    continue;
+                ReadResult result;
+                try { result = await pipe.ReadAsync(_cts.Token); }
+                catch (OperationCanceledException) { break; }
 
-                using var doc = JsonDocument.Parse(line);
+                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();
+            Disconnected?.Invoke(failure);
+        }
+    }
+
+    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))
-                        continue;
-                    if (root.TryGetProperty("error", out var error) && error.ValueKind != JsonValueKind.Null)
-                        tcs.TrySetException(new ElectrumServerException(error.ToString()));
+            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());
         }
@@ -159,28 +263,16 @@ public sealed class ElectrumClient : IAsyncDisposable
                 root.GetProperty("params").Clone());
         }
     }
-        }
-        catch (OperationCanceledException) { }
-        catch (Exception ex)
-        {
-            failure = ex;
-        }
-        finally
-        {
-            foreach (var (_, tcs) in _pending)
-                tcs.TrySetException(failure ?? new IOException("Connessione al server chiusa."));
-            _pending.Clear();
-            Disconnected?.Invoke(failure);
-        }
-    }
 
     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();
+        _inFlight.Dispose();
     }
 }