Survive oversized Electrum replies and ignore dust deposits (B-75, ELC-20, DEP-01, DAT-10)

A reply line over asyncio's default 64 KiB StreamReader limit raised out of
_read_loop and closed the only connection; the pool address's get_history passes
that after ~800 transactions, and dust sent to any public address does it on
purpose.

- ElectrumClient opens the stream with a 16 MiB limit and frames lines itself:
  an oversized line is discarded up to its newline, a non-JSON line is dropped,
  and the request whose id is readable in the first/last 256 bytes fails at once;
  the connection stays up either way.
- Coins below DUST_LIMIT_SATS are never offered as credit candidates, never
  credited, and select_utxos never selects them.
- Spec (ELC-20, DEP-01, DAT-10, ARC), storico, BUGS.md and test counts updated.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

app/electrum/client.py

@@ -1,10 +1,15 @@
 import asyncio
 import itertools
 import json
+import logging
+import re
 import ssl
 from typing import NamedTuple
 
 
+logger = logging.getLogger(__name__)
+
+
 class ElectrumEndpoint(NamedTuple):
     """One server to connect to. The listener rotates over a list of these so a
     single unreachable or misbehaving server doesn't take the platform down —
@@ -64,6 +69,18 @@ def parse_endpoints(
 # and the confirmation poller could stop polling permanently.
 _REQUEST_TIMEOUT_SECONDS = 15
 
+# Longest reply line the client accepts (B-75, ELC-20). asyncio's default StreamReader
+# limit is 64 KiB, and a longer line used to raise out of the read loop and close the
+# only connection: the pool address's get_history passed that after ~800 transactions
+# (and a few hundred dust outputs sent to any public address do it on purpose), after
+# which every confirmation poll dropped the connection and payouts never confirmed.
+_MAX_LINE_BYTES = 16 * 1024 * 1024
+_READ_CHUNK_BYTES = 64 * 1024
+# How much of each end of a discarded line is kept to find the reply's `id` in, so the
+# request it answered can fail at once instead of waiting out the 15s timeout.
+_ID_SNIFF_BYTES = 256
+_ID_PATTERN = re.compile(rb'"id"\s*:\s*(\d+)')
+
 
 class ElectrumError(Exception):
     pass
@@ -104,7 +121,9 @@ class ElectrumClient:
             ssl_context = ssl.create_default_context()
             ssl_context.check_hostname = False
             ssl_context.verify_mode = ssl.CERT_NONE
-        self._reader, self._writer = await asyncio.open_connection(self.host, self.port, ssl=ssl_context)
+        self._reader, self._writer = await asyncio.open_connection(
+            self.host, self.port, ssl=ssl_context, limit=_MAX_LINE_BYTES
+        )
         self._read_task = asyncio.create_task(self._read_loop())
         await self.request("server.version", ["plm-lottery", "1.4"])
 
@@ -187,16 +206,44 @@ class ElectrumClient:
         return await self.request("blockchain.transaction.get", [txid, verbose])
 
     async def _read_loop(self) -> None:
+        """Frame the stream into lines by hand rather than with StreamReader.readline:
+        readline raises on a line over the limit, and that exception used to end this
+        loop and with it the connection. An oversized line is now dropped up to its
+        newline (B-75, ELC-20) and the connection stays up."""
         assert self._reader is not None
+        buffer = bytearray()
+        discarding = False
+        head = b""
+        tail = b""
         try:
             while True:
-                line = await self._reader.readline()
-                if not line:
+                chunk = await self._reader.read(_READ_CHUNK_BYTES)
+                if not chunk:
                     break
-                message = json.loads(line)
-                self._dispatch(message)
+                while chunk:
+                    newline = chunk.find(b"\n")
+                    piece = chunk if newline < 0 else chunk[:newline]
+                    chunk = b"" if newline < 0 else chunk[newline + 1 :]
+                    if discarding:
+                        tail = (tail + piece)[-_ID_SNIFF_BYTES:]
+                    else:
+                        buffer += piece
+                        if len(buffer) > _MAX_LINE_BYTES:
+                            discarding = True
+                            head = bytes(buffer[:_ID_SNIFF_BYTES])
+                            tail = bytes(buffer[-_ID_SNIFF_BYTES:])
+                            buffer.clear()
+                    if newline >= 0:
+                        if discarding:
+                            self._fail_unreadable_reply(head + b" " + tail, "reply exceeds the line limit")
+                            discarding = False
+                        else:
+                            self._handle_line(bytes(buffer))
+                        buffer.clear()
+            if buffer and not discarding:
+                self._handle_line(bytes(buffer))  # a final line the server never terminated
         finally:
-            # Whatever ended this loop — clean EOF, protocol error, TLS failure — the
+            # Whatever ended this loop — clean EOF, TLS failure — the
             # connection is unusable from here on. Setting this is what makes the
             # death observable to wait_closed(), and so to the listener's reconnect
             # logic; without it the listener waited on notification queues nobody
@@ -208,6 +255,33 @@ class ElectrumClient:
                     future.set_exception(error)
             self._pending.clear()
 
+    def _handle_line(self, line: bytes) -> None:
+        if not line.strip():
+            return
+        try:
+            message = json.loads(line)
+        except ValueError:
+            # A reply that is not JSON is a problem of that one request, not of the
+            # connection (B-75).
+            self._fail_unreadable_reply(line[:_ID_SNIFF_BYTES] + b" " + line[-_ID_SNIFF_BYTES:], "reply is not valid JSON")
+            return
+        if not isinstance(message, dict):
+            logger.warning("ignoring an Electrum message that is not a JSON object")
+            return
+        self._dispatch(message)
+
+    def _fail_unreadable_reply(self, sample: bytes, reason: str) -> None:
+        """Fail the request an unusable reply belongs to, when its `id` can be read
+        from `sample` (the first and last bytes of the line); otherwise that request
+        fails by its own timeout (ELC-04)."""
+        logger.warning("dropping an unusable Electrum reply: %s", reason)
+        for match in _ID_PATTERN.finditer(sample):
+            future = self._pending.pop(int(match.group(1)), None)
+            if future is not None:
+                if not future.done():
+                    future.set_exception(ElectrumError(reason))
+                return
+
     def _dispatch(self, message: dict) -> None:
         message_id = message.get("id")
         if message_id is not None and message_id in self._pending: