davideandClaude Opus 5
7c4e9983ea
A dropped connection used to hang the whole platform permanently, and three defects composed to do it (BUGS.md B-01): The read loop's death was invisible. When the socket closed, _read_loop broke out and finished, but _run_once was blocked on gather() over two notification consumers waiting on queues nobody would ever fill again — it never returned and never raised, so the reconnect-with-backoff logic was unreachable. client.wait_closed() now resolves when the loop ends for any reason, and _run_once races it against the consumers and a keepalive with asyncio.wait(FIRST_COMPLETED). Nothing had a timeout. request() registered a future, wrote to a half-closed socket (drain() often doesn't raise) and awaited a reply that would never come. That hung a POST /bets *while holding the per-user lock*, and could stop the confirmation poller for good. Every request is now bounded at 15s, and a timeout tears the connection down rather than leaving a server that owes us a reply in rotation. There was no keepalive, so on a quiet instance the normal way this connection dies is an idle-timeout drop by the server (~10 minutes for many). A server.ping every 60s makes that observable within a minute. listener.client is also cleared before reconnecting, so callers stop treating a dead connection as live. On top of the finding, the listener now rotates over a list of servers: ELECTRUM_FALLBACK_SERVERS holds comma-separated host:port[:notls] extras, tried after the primary. Everything the platform does goes through this one connection — deposit credits, broadcasts, confirmations, the chain tip the draw waits on — which made a single hardcoded server its biggest point of failure. A failed or dropped session moves to the next server immediately and only sleeps on the backoff once every server has had a turn, so one dead server costs one attempt instead of an outage, while a genuinely offline network still backs off. A malformed entry fails at startup, not during the outage when the fallback is what you need. Also fixes B-19: header handling refuses a height below the current tip and applies height and hex together, since _wait_for_next_block waits for tip_height > tip_at_close (a regression silently added a block to the draw's wait) and that hex is the draw's entropy source, so a mismatched pair would be worse than a stale one. Verified in the live deployment: the log shows the endpoint list, then "Electrum connected to santantonio.sytes.net:50002", and the connection holds. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
158 lines
5.2 KiB
Python
158 lines
5.2 KiB
Python
import asyncio
|
|
import json
|
|
|
|
from app.electrum.client import ElectrumClient, ElectrumError
|
|
|
|
|
|
class FakeWriter:
|
|
def __init__(self):
|
|
self.written = b""
|
|
|
|
def write(self, data: bytes) -> None:
|
|
self.written += data
|
|
|
|
async def drain(self) -> None:
|
|
pass
|
|
|
|
def close(self) -> None:
|
|
pass
|
|
|
|
async def wait_closed(self) -> None:
|
|
pass
|
|
|
|
|
|
async def _client_with_fake_transport() -> tuple[ElectrumClient, asyncio.StreamReader, FakeWriter]:
|
|
client = ElectrumClient("localhost", 1234)
|
|
reader = asyncio.StreamReader()
|
|
writer = FakeWriter()
|
|
client._reader = reader
|
|
client._writer = writer
|
|
client._read_task = asyncio.create_task(client._read_loop())
|
|
return client, reader, writer
|
|
|
|
|
|
async def test_request_resolves_on_matching_response():
|
|
client, reader, writer = await _client_with_fake_transport()
|
|
|
|
task = asyncio.create_task(client.request("blockchain.headers.subscribe"))
|
|
await asyncio.sleep(0) # let request() write the payload
|
|
sent = json.loads(writer.written.decode())
|
|
assert sent["method"] == "blockchain.headers.subscribe"
|
|
|
|
reader.feed_data((json.dumps({"id": sent["id"], "result": {"height": 100}}) + "\n").encode())
|
|
result = await task
|
|
assert result == {"height": 100}
|
|
|
|
client._read_task.cancel()
|
|
|
|
|
|
async def test_error_response_raises_electrum_error():
|
|
client, reader, writer = await _client_with_fake_transport()
|
|
|
|
task = asyncio.create_task(client.request("blockchain.transaction.broadcast", ["deadbeef"]))
|
|
await asyncio.sleep(0)
|
|
sent = json.loads(writer.written.decode())
|
|
|
|
reader.feed_data((json.dumps({"id": sent["id"], "error": "bad tx"}) + "\n").encode())
|
|
try:
|
|
await task
|
|
assert False, "expected ElectrumError"
|
|
except ElectrumError:
|
|
pass
|
|
|
|
client._read_task.cancel()
|
|
|
|
|
|
async def test_notification_delivered_to_subscription_queue():
|
|
client, reader, writer = await _client_with_fake_transport()
|
|
queue = client.notifications("blockchain.scripthash.subscribe")
|
|
|
|
push = {"method": "blockchain.scripthash.subscribe", "params": ["abcd", "newstatus"]}
|
|
reader.feed_data((json.dumps(push) + "\n").encode())
|
|
await asyncio.sleep(0)
|
|
|
|
params = await asyncio.wait_for(queue.get(), timeout=1)
|
|
assert params == ["abcd", "newstatus"]
|
|
|
|
client._read_task.cancel()
|
|
|
|
|
|
async def test_request_times_out_instead_of_hanging_forever(monkeypatch):
|
|
"""B-01: a server that owes us a reply and never sends one used to hang the
|
|
caller permanently — which meant a POST /bets could hang while holding the
|
|
per-user lock, and the confirmation poller could stop polling for good."""
|
|
from app.electrum import client as client_module
|
|
|
|
monkeypatch.setattr(client_module, "_REQUEST_TIMEOUT_SECONDS", 0.05)
|
|
client, reader, writer = await _client_with_fake_transport()
|
|
|
|
try:
|
|
await client.request("blockchain.transaction.get", ["deadbeef"])
|
|
assert False, "expected ElectrumError"
|
|
except ElectrumError as exc:
|
|
assert "timed out" in str(exc)
|
|
|
|
# The connection is torn down, so callers stop reusing a server that owes us.
|
|
assert client._closed.is_set()
|
|
client._read_task.cancel()
|
|
|
|
|
|
async def test_wait_closed_resolves_when_the_read_loop_dies():
|
|
"""B-01: the read loop dying used to be invisible — the listener sat on its
|
|
notification queues forever and never reconnected."""
|
|
client, reader, writer = await _client_with_fake_transport()
|
|
|
|
waiter = asyncio.create_task(client.wait_closed())
|
|
reader.feed_eof() # peer closed the connection
|
|
|
|
await asyncio.wait_for(waiter, timeout=1)
|
|
client._read_task.cancel()
|
|
|
|
|
|
async def test_pending_request_fails_when_the_connection_drops():
|
|
client, reader, writer = await _client_with_fake_transport()
|
|
|
|
task = asyncio.create_task(client.request("server.ping"))
|
|
await asyncio.sleep(0)
|
|
reader.feed_eof()
|
|
|
|
try:
|
|
await asyncio.wait_for(task, timeout=1)
|
|
assert False, "expected ElectrumError"
|
|
except ElectrumError:
|
|
pass
|
|
|
|
|
|
def test_parse_endpoints_puts_the_primary_first_and_dedupes():
|
|
from app.electrum.client import ElectrumEndpoint, parse_endpoints
|
|
|
|
endpoints = parse_endpoints(
|
|
"primary.example", 50002, True, "second.example:50002, third.example:50001:notls, primary.example:50002"
|
|
)
|
|
assert endpoints == [
|
|
ElectrumEndpoint("primary.example", 50002, True),
|
|
ElectrumEndpoint("second.example", 50002, True),
|
|
ElectrumEndpoint("third.example", 50001, False),
|
|
]
|
|
|
|
|
|
def test_parse_endpoints_handles_an_empty_fallback_list():
|
|
from app.electrum.client import ElectrumEndpoint, parse_endpoints
|
|
|
|
assert parse_endpoints("only.example", 50002, True, "") == [ElectrumEndpoint("only.example", 50002, True)]
|
|
assert parse_endpoints("only.example", 50002, True, " , ") == [
|
|
ElectrumEndpoint("only.example", 50002, True)
|
|
]
|
|
|
|
|
|
def test_parse_endpoints_rejects_a_malformed_entry():
|
|
"""A typo in a fallback server must fail at startup, not during the outage when
|
|
the fallback is the thing that's needed."""
|
|
import pytest
|
|
|
|
from app.electrum.client import parse_endpoints
|
|
|
|
for bad in ["nohost:", "host:notaport", "host", "host:50002:weird"]:
|
|
with pytest.raises(ValueError):
|
|
parse_endpoints("primary.example", 50002, True, bad)
|