From a748db29aab4de6b09545e7b1ea351ad68e3d9ed Mon Sep 17 00:00:00 2001 From: BareBits Date: Mon, 27 Jul 2026 21:45:16 -0700 Subject: [PATCH] Fix Electrum hanging on shutdown: cancel-safe, resumable announcement PoW Electrum never fully shut down once the swap server had been enabled. The asyncio loop runs on a non-daemon thread ('EventLoop', electrum/util.py), so anything that blocks that thread keeps the interpreter alive forever. Root cause: electrum.util.gen_nostr_ann_pow deadlocks the loop thread when the awaiting task is cancelled mid-grind. with ProcessPoolExecutor(max_workers=max_workers) as executor: ... done, pending = await asyncio.wait(tasks, ...) # CancelledError executor.shutdown(wait=False, cancel_futures=True) # success path only On cancellation the context manager's __exit__ runs executor.shutdown(wait=True), a synchronous join, on the loop thread. The workers only exit when a shared shutdown Event is set, and that Event is only set by a worker that *finds* a solution, so the join never returns. The plugin reached that code from two tasks it cancels at shutdown: run_nostr_server and _one_shot_publish both begin with set_nostr_proof_of_work. stop_server() cancels both from the close_wallet / on_close_window hooks. With the default 30-bit target and nonce 0 a fresh server always grinds (~3.5 min on an 8-core box), so quitting in that window wedged the loop permanently. The two tasks also ground concurrently, doubling CPU use for identical work. Fix: own the proof of work (plugins/swapserver_gui/pow.py) and seed the nonce before starting the nostr transport, so upstream's set_nostr_proof_of_work always short-circuits on a cached nonce and never enters the broken function. * raw multiprocessing.Process instead of ProcessPoolExecutor, whose atexit hook is itself a hang vector; cancellation signals, terminates and hands joining to a daemon thread, never blocking the loop or the GUI thread * resumable: workers checkpoint their nonce cursor, persisted per lane in a new config var, so a grind cancelled by quitting continues into untested nonce space instead of re-scanning the range upstream deterministically re-tests. Tracking the best nonce seen keeps cursors valid for any target * thread fallback where fork is unavailable (Windows/Android), since the plugin package is not guaranteed importable in a spawned child Also fixed, all contributing to an unclean teardown: * stop_server() now unregisters the HTTP EventListener synchronously, so a cancelled cleanup coroutine cannot leave stale callbacks registered * the 4s refresh QTimer is stopped when the tab is removed; it was calling into get_asyncio_loop() while the loop was being torn down * publish_now waits on a PoW gate rather than computing its own The tab now shows proof-of-work progress and an estimated grind time next to the target, and confirms before raising the target past 30 bits. Tests: 52 pass. tests/test_shutdown_e2e.py reproduces the reported bug end to end, including a subprocess test asserting a whole interpreter actually exits while the proof of work is still running, and that the wallet closes cleanly mid-grind with progress preserved. Verified against the real upstream helper: cancelling gen_nostr_ann_pow wedges the loop and the process needs SIGKILL, while the replacement exits cleanly. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_013zETFC6MhnsCuxvQDGxqV5 --- contrib/make_zip.sh | 1 + plugins/swapserver_gui/__init__.py | 13 + plugins/swapserver_gui/pow.py | 470 +++++++++++++++++++++++ plugins/swapserver_gui/qt.py | 100 ++++- plugins/swapserver_gui/swapserver_gui.py | 147 ++++++- tests/test_pow.py | 319 +++++++++++++++ tests/test_shutdown_e2e.py | 374 ++++++++++++++++++ tests/test_swapserver_gui.py | 55 ++- 8 files changed, 1461 insertions(+), 18 deletions(-) create mode 100644 plugins/swapserver_gui/pow.py create mode 100644 tests/test_pow.py create mode 100644 tests/test_shutdown_e2e.py diff --git a/contrib/make_zip.sh b/contrib/make_zip.sh index d12d311..84d6d8a 100755 --- a/contrib/make_zip.sh +++ b/contrib/make_zip.sh @@ -9,6 +9,7 @@ # swapserver_gui/manifest.json # swapserver_gui/__init__.py # swapserver_gui/swapserver_gui.py +# swapserver_gui/pow.py # swapserver_gui/qt.py # # The zip is a plain archive: no signing key or secret is required to build it. diff --git a/plugins/swapserver_gui/__init__.py b/plugins/swapserver_gui/__init__.py index 8492d01..1841254 100644 --- a/plugins/swapserver_gui/__init__.py +++ b/plugins/swapserver_gui/__init__.py @@ -29,3 +29,16 @@ type_=bool, plugin='swapserver_gui', ) + +# Resumable proof-of-work search state, as JSON (see swapserver_gui/pow.py). +# Holds the nostr pubkey the search applies to, the per-lane nonce cursors, and +# the best nonce seen so far. Persisting this means a proof-of-work that gets +# cancelled (e.g. by quitting Electrum mid-grind) is not thrown away: the next +# run continues into untested nonce space instead of re-scanning the range it +# already rejected. +SimpleConfig.SWAPSERVER_GUI_POW_STATE = ConfigVar( + 'plugins.swapserver_gui.pow_state', + default='', + type_=str, + plugin='swapserver_gui', +) diff --git a/plugins/swapserver_gui/pow.py b/plugins/swapserver_gui/pow.py new file mode 100644 index 0000000..96b21d1 --- /dev/null +++ b/plugins/swapserver_gui/pow.py @@ -0,0 +1,470 @@ +#!/usr/bin/env python +# +# swapserver_gui - a Qt GUI plugin for Electrum's submarine swap server. +# This file is released into the public domain (The Unlicense); see LICENSE. +# +# Cancel-safe, resumable proof-of-work for the nostr swap announcement. +# +# WHY THIS MODULE EXISTS +# ---------------------- +# Electrum's ``electrum.util.gen_nostr_ann_pow`` deadlocks the asyncio event +# loop if the awaiting task is cancelled mid-grind: +# +# with ProcessPoolExecutor(max_workers=max_workers) as executor: +# ... +# done, pending = await asyncio.wait(tasks, ...) # <- CancelledError here +# executor.shutdown(wait=False, cancel_futures=True) # only on success +# +# On cancellation the ``with`` block's ``__exit__`` runs +# ``executor.shutdown(wait=True)``, a *synchronous* join, on the event loop +# thread. The workers (``electrum.util.nostr_pow_worker``) only ever exit when +# the shared ``shutdown`` Event is set, and that Event is only set by a worker +# that *finds* a solution. So the join never returns and the loop thread is +# wedged forever. Because Electrum's loop runs on a non-daemon thread +# (``electrum/util.py: create_and_start_event_loop``), the whole process then +# refuses to exit -- Electrum "hangs on shutdown". +# +# We therefore never let upstream grind: we compute the nonce here, store it in +# ``config.SWAPSERVER_ANN_POW_NONCE``, and upstream's +# ``SwapManager.set_nostr_proof_of_work`` short-circuits on the cached value. +# +# DESIGN NOTES +# ------------ +# * Raw ``multiprocessing.Process`` instead of ``ProcessPoolExecutor``: the +# executor installs an ``atexit`` hook that joins its workers at interpreter +# shutdown, which is itself a hang vector. Raw processes let us SIGTERM/SIGKILL. +# * Cancellation never blocks the event loop thread: we ``terminate()`` (instant) +# and hand the joining off to a short-lived *daemon* thread. +# * The search is a linear nonce scan, so while there is no such thing as being +# "partly done" with a PoW (each nonce is an independent 2^-target trial), the +# set of *already-rejected* nonces is worth keeping. Upstream restarts every +# worker from a deterministic start nonce, so a cancelled grind re-tests +# exactly the range it already rejected. We persist a per-lane cursor so work +# accumulates monotonically across restarts. +# * We also track the best nonce seen so far. That makes the persisted cursors +# valid for *any* target: if ``best_bits < target`` then no scanned nonce can +# meet ``target`` either (best is the maximum), so the scanned range stays +# exhausted even when the user lowers the target. + +import asyncio +import hashlib +import json +import multiprocessing +import threading +import time +from dataclasses import dataclass, field +from typing import Any, Callable, List, Optional, Sequence + +HASH_LEN_BITS = 256 +MAX_NONCE = (1 << (32 * 8)) - 1 # 32-byte nonce, as upstream + +# The nonce space is split into a fixed number of "lanes" so that a persisted +# cursor keeps its meaning across machines with different core counts. A lane +# is ~2^250 nonces wide, i.e. unreachable in practice. +NUM_LANES = 64 +LANE_SIZE = MAX_NONCE // NUM_LANES + +# Hashes a worker computes between two checkpoints. Checking the shutdown flag +# per hash would dominate the hot loop; per block it is free. At ~0.7 MH/s this +# bounds a worker's cancellation latency to roughly 1.5s (it is terminated with +# a signal anyway, so this only matters for the clean-exit path). +CHECKPOINT_BLOCK = 1_000_000 + + +def pow_bits(pubkey: bytes, nonce: Optional[int]) -> int: + """Leading-zero bits of ``sha256(b'electrum-' + pubkey + nonce)``. + + Mirrors ``electrum.util.get_nostr_ann_pow_amount`` exactly; kept local so + this module can be reasoned about (and unit-tested) on its own. + """ + if not nonce: + return 0 + digest = hashlib.sha256(b'electrum-' + pubkey + nonce.to_bytes(32, 'big')).digest() + return HASH_LEN_BITS - int.from_bytes(digest, 'big').bit_length() + + +def lane_start(lane_idx: int) -> int: + return lane_idx * LANE_SIZE + + +# --------------------------------------------------------------------- state +@dataclass +class PowState: + """Resumable search state, persisted as JSON in the plugin's config var.""" + + pubkey_hex: str = "" + target: int = 0 + offsets: List[int] = field(default_factory=list) # hashes done, per lane + best_nonce: Optional[int] = None + best_bits: int = 0 + + def hashes_done(self) -> int: + return sum(self.offsets) + + def offset_for(self, lane_idx: int) -> int: + if lane_idx < len(self.offsets): + return self.offsets[lane_idx] + return 0 + + def dumps(self) -> str: + return json.dumps({ + "pubkey": self.pubkey_hex, + "target": self.target, + "offsets": self.offsets, + # decimal string: the nonce is a 256-bit integer + "best_nonce": str(self.best_nonce) if self.best_nonce is not None else None, + "best_bits": self.best_bits, + }) + + @classmethod + def load(cls, raw: Optional[str], *, pubkey: bytes, target: int) -> 'PowState': + """Restore state for ``pubkey``, or a blank state if it does not apply. + + The cursors are only meaningful for the pubkey they were scanned + against (it is part of the hash preimage), so a wallet switch discards + them. A change of *target* does not invalidate them -- see the module + docstring. + """ + fresh = cls(pubkey_hex=pubkey.hex(), target=target) + if not raw: + return fresh + try: + data = json.loads(raw) + if not isinstance(data, dict): + return fresh + if data.get("pubkey") != pubkey.hex(): + return fresh # different nostr key -> different preimage + offsets = [int(o) for o in data.get("offsets") or []] + if any(o < 0 or o >= LANE_SIZE for o in offsets): + return fresh # corrupt + raw_nonce = data.get("best_nonce") + best_nonce = int(raw_nonce) if raw_nonce not in (None, "", "None") else None + if best_nonce is not None and not (0 <= best_nonce <= MAX_NONCE): + return fresh + best_bits = int(data.get("best_bits") or 0) + # Trust but verify: recompute the claimed best so a corrupted or + # hand-edited config can never make us publish a bad nonce. + if best_nonce is not None: + actual = pow_bits(pubkey, best_nonce) + if actual != best_bits: + best_bits = actual + return cls( + pubkey_hex=pubkey.hex(), + target=target, + offsets=offsets, + best_nonce=best_nonce, + best_bits=best_bits, + ) + except (ValueError, TypeError): + return fresh + + +@dataclass +class PowResult: + nonce: Optional[int] + bits: int + state: PowState + found: bool + + +# -------------------------------------------------------------------- worker +def _grind_worker( + lane_idx: int, + start_nonce: int, + initial_offset: int, + pubkey: bytes, + target_bits: int, + shutdown: Any, # multiprocessing.Event or threading.Event + progress: Any, # multiprocessing.Array('Q') or list + best_bits: Any, # multiprocessing.Array('i') or list + best_off: Any, # multiprocessing.Array('Q') or list + found: Any, # multiprocessing.Array('b') or list +) -> None: + """Scan a lane for a nonce meeting ``target_bits``. + + Shared by the process backend (via fork) and the thread fallback; the + shared-state arguments only need ``__getitem__``/``__setitem__``. + """ + sha256 = hashlib.sha256 + preimage = b'electrum-' + pubkey + threshold = 1 << (HASH_LEN_BITS - target_bits) + offset = initial_offset + nonce = start_nonce + offset + local_best_val = 1 << HASH_LEN_BITS # sentinel: worse than any real digest + local_best_off = offset + + while True: + for _ in range(CHECKPOINT_BLOCK): + digest = sha256(preimage + nonce.to_bytes(32, 'big')).digest() + val = int.from_bytes(digest, 'big') + if val < threshold: + progress[lane_idx] = offset + 1 + best_off[lane_idx] = offset + best_bits[lane_idx] = HASH_LEN_BITS - val.bit_length() + found[lane_idx] = 1 + shutdown.set() + return + if val < local_best_val: # single int compare: cheap in the hot loop + local_best_val = val + local_best_off = offset + nonce += 1 + offset += 1 + progress[lane_idx] = offset + bits = HASH_LEN_BITS - local_best_val.bit_length() + if bits > best_bits[lane_idx]: + best_bits[lane_idx] = bits + best_off[lane_idx] = local_best_off + if shutdown.is_set(): + return + + +# ------------------------------------------------------------------ backends +def _fork_available() -> bool: + try: + return "fork" in multiprocessing.get_all_start_methods() + except (AttributeError, ValueError, ImportError): + return False + + +def default_worker_count() -> int: + """All but one CPU, as upstream, so the UI stays responsive.""" + try: + return max(multiprocessing.cpu_count() - 1, 1) + except NotImplementedError: + return 1 + + +class _Backend: + """Owns the running workers and the shared progress state.""" + + def __init__(self, num_workers: int) -> None: + self.num_workers = num_workers + self.procs: List[Any] = [] + self.shutdown: Any = None + self.progress: Any = None + self.best_bits: Any = None + self.best_off: Any = None + self.found: Any = None + self.uses_processes = False + + def start(self, *, pubkey: bytes, target_bits: int, state: PowState) -> None: + n = self.num_workers + # 'fork' lets the children inherit the shared primitives (they cannot be + # pickled, which is why upstream needed a Manager process). Without it + # we fall back to a single in-process thread: the plugin dir is not + # guaranteed to be importable in a 'spawn'ed child (it may be loaded + # from a zip as electrum_external_plugins.swapserver_gui), so a process + # backend there would fail at unpickling. + # + # Python 3.12 warns that forking a multi-threaded process (which + # Electrum is) can deadlock the child. That is safe here because + # _grind_worker touches nothing that could hold an inherited lock: no + # imports, no logging, no allocator-heavy library calls -- only hashlib, + # int arithmetic, and the inherited shared primitives (which are POSIX + # semaphores / shared memory, not Python locks). + if _fork_available() and n > 1: + ctx = multiprocessing.get_context("fork") + self.shutdown = ctx.Event() + self.progress = ctx.Array('Q', n) # unsigned long long + self.best_bits = ctx.Array('i', n) + self.best_off = ctx.Array('Q', n) + self.found = ctx.Array('b', n) + self.uses_processes = True + factory: Callable[..., Any] = ctx.Process + else: + self.num_workers = n = 1 + self.shutdown = threading.Event() + self.progress = [0] + self.best_bits = [0] + self.best_off = [0] + self.found = [0] + self.uses_processes = False + factory = threading.Thread + + for lane_idx in range(n): + offset = state.offset_for(lane_idx) + self.progress[lane_idx] = offset + proc = factory( + target=_grind_worker, + args=(lane_idx, lane_start(lane_idx), offset, pubkey, target_bits, + self.shutdown, self.progress, self.best_bits, + self.best_off, self.found), + daemon=True, # must never keep the interpreter alive + name=f"swap-pow-{lane_idx}", + ) + proc.start() + self.procs.append(proc) + + def any_alive(self) -> bool: + return any(p.is_alive() for p in self.procs) + + def winner(self) -> Optional[int]: + """Lane index that found a solution, if any.""" + for lane_idx in range(self.num_workers): + if self.found[lane_idx]: + return lane_idx + return None + + def snapshot(self, state: PowState) -> PowState: + """Fold the live shared counters back into a persistable state.""" + offsets = list(state.offsets) + while len(offsets) < self.num_workers: + offsets.append(0) + for lane_idx in range(self.num_workers): + offsets[lane_idx] = max(offsets[lane_idx], int(self.progress[lane_idx])) + bits = int(self.best_bits[lane_idx]) + if bits > state.best_bits: + state.best_bits = bits + state.best_nonce = lane_start(lane_idx) + int(self.best_off[lane_idx]) + state.offsets = offsets + return state + + def stop(self) -> None: + """Tear the workers down WITHOUT ever blocking the caller. + + This runs on the asyncio loop thread during cancellation, so it must + return immediately -- that is the entire point of this module. + """ + if self.shutdown is not None: + self.shutdown.set() + procs = list(self.procs) + self.procs = [] + if not procs: + return + if self.uses_processes: + for p in procs: + try: + if p.is_alive(): + p.terminate() # SIGTERM, returns immediately + except (OSError, ValueError, AttributeError): + pass + + def _reap() -> None: + for p in procs: + try: + p.join(timeout=2.0) + if self.uses_processes and p.is_alive(): + p.kill() + p.join(timeout=2.0) + except (OSError, ValueError, AttributeError): + pass + + # daemon thread: reaping must not delay shutdown either + threading.Thread(target=_reap, name="swap-pow-reap", daemon=True).start() + + +# --------------------------------------------------------------------- grind +async def grind( + *, + pubkey: bytes, + target_bits: int, + state: PowState, + num_workers: Optional[int] = None, + on_progress: Optional[Callable[[PowState], None]] = None, + poll_interval: float = 0.25, + persist_interval: float = 10.0, +) -> PowResult: + """Search for a nonce with at least ``target_bits`` of work. + + Fully cancellable: on ``CancelledError`` the workers are signalled and + terminated, the search cursors are folded back into ``state`` and handed to + ``on_progress`` (so the caller can persist them), and the error is re-raised. + The event loop thread is never blocked, on any path. + """ + if num_workers is None: + num_workers = default_worker_count() + num_workers = max(1, min(num_workers, NUM_LANES)) + + backend = _Backend(num_workers) + last_persist = 0.0 + + def _publish(force: bool = False) -> None: + nonlocal last_persist + backend.snapshot(state) + if on_progress is None: + return + now = time.monotonic() + if force or (now - last_persist) >= persist_interval: + last_persist = now + on_progress(state) + + try: + backend.start(pubkey=pubkey, target_bits=target_bits, state=state) + last_persist = time.monotonic() + while True: + await asyncio.sleep(poll_interval) + lane = backend.winner() + if lane is not None: + break + if not backend.any_alive(): + # every worker exited without a hit (only possible if something + # external killed them); report what we scanned and give up. + _publish(force=True) + return PowResult(nonce=None, bits=state.best_bits, state=state, found=False) + _publish() + except asyncio.CancelledError: + backend.stop() + _publish(force=True) + raise + except BaseException: + backend.stop() + _publish(force=True) + raise + + # success path + nonce = lane_start(lane) + int(backend.best_off[lane]) + bits = pow_bits(pubkey, nonce) + backend.stop() + if bits > state.best_bits: + state.best_bits = bits + state.best_nonce = nonce + _publish(force=True) + return PowResult(nonce=nonce, bits=bits, state=state, found=True) + + +# ------------------------------------------------------------------ estimates +_HASH_RATE: Optional[float] = None + + +def benchmark_hash_rate(samples: int = 40_000) -> float: + """Single-core hashes/second, measured once and cached. + + Used only to show the user an estimate before they commit to a target. + """ + global _HASH_RATE + if _HASH_RATE is not None: + return _HASH_RATE + preimage = b'electrum-' + bytes(32) + sha256 = hashlib.sha256 + nonce = 0 + t0 = time.perf_counter() + for _ in range(samples): + sha256(preimage + nonce.to_bytes(32, 'big')).digest() + nonce += 1 + elapsed = time.perf_counter() - t0 + _HASH_RATE = samples / elapsed if elapsed > 0 else 1e6 + return _HASH_RATE + + +def estimate_seconds(target_bits: int, *, num_workers: Optional[int] = None, + hash_rate: Optional[float] = None) -> float: + """Expected seconds to find a nonce with ``target_bits`` of work.""" + if target_bits <= 0: + return 0.0 + if num_workers is None: + num_workers = default_worker_count() + if hash_rate is None: + hash_rate = benchmark_hash_rate() + return (1 << target_bits) / max(hash_rate * num_workers, 1.0) + + +def format_duration(seconds: float) -> str: + if seconds < 1: + return "< 1s" + if seconds < 90: + return f"{seconds:.0f}s" + if seconds < 90 * 60: + return f"{seconds / 60:.1f} min" + if seconds < 48 * 3600: + return f"{seconds / 3600:.1f} hours" + return f"{seconds / 86400:.1f} days" diff --git a/plugins/swapserver_gui/qt.py b/plugins/swapserver_gui/qt.py index 54a0ec9..4754ad6 100644 --- a/plugins/swapserver_gui/qt.py +++ b/plugins/swapserver_gui/qt.py @@ -22,6 +22,11 @@ from .swapserver_gui import ( SwapServerGuiPlugin, SwapServerError, get_swap_history, get_swap_summary, ) +from . import pow as swap_pow + +# Above this many bits a proof-of-work grind stops being a "wait a few minutes" +# affair (see the estimate shown next to the spinbox), so we ask for confirmation. +POW_TARGET_WARN_ABOVE = 30 if TYPE_CHECKING: from electrum.wallet import Abstract_Wallet @@ -65,7 +70,7 @@ def __init__(self, plugin: 'Plugin', window: 'ElectrumWindow') -> None: body.addWidget(self._build_output_group(), 1) # ---- periodic refresh -------------------------------------------- - self._timer = QTimer(self) + self._timer: Optional[QTimer] = QTimer(self) self._timer.timeout.connect(self.refresh) self._timer.start(4000) @@ -98,8 +103,17 @@ def _build_settings_group(self) -> QGroupBox: self.pow_spin = QSpinBox() self.pow_spin.setRange(0, 40) - self.pow_spin.setToolTip(_("Proof-of-work target (in bits) for the nostr announcement.")) - form.addRow(_("Nostr PoW target:"), self.pow_spin) + self.pow_spin.setToolTip(_( + "Proof-of-work target (in bits) for the nostr announcement.\n" + "Each extra bit doubles the expected computation time.")) + self.pow_est_label = QLabel() + self.pow_spin.valueChanged.connect(self._update_pow_estimate) + pow_row = QHBoxLayout() + pow_row.addWidget(self.pow_spin) + pow_row.addWidget(self.pow_est_label) + pow_wrap = QWidget() + pow_wrap.setLayout(pow_row) + form.addRow(_("Nostr PoW target:"), pow_wrap) self.relays_edit = QPlainTextEdit() self.relays_edit.setPlaceholderText("wss://relay.example.com, wss://relay2.example.com") @@ -122,6 +136,7 @@ def _build_output_group(self) -> QGroupBox: rows = [ ("http", _("HTTP endpoint:")), ("nostr", _("Nostr announcement:")), + ("pow", _("Proof of work:")), ("percentage", _("Fee percentage:")), ("min_amount", _("Min amount:")), ("max_forward", _("Max forward (normal):")), @@ -155,10 +170,27 @@ def load_settings_into_widgets(self) -> None: self.pow_spin.setValue(int(self.config.SWAPSERVER_POW_TARGET)) self.relays_edit.setPlainText((self.config.NOSTR_RELAYS or "").replace(",", ",\n")) self._update_fee_label() + self._update_pow_estimate() def _update_fee_label(self) -> None: self.fee_pct_label.setText("= {:.4f} %".format(self.fee_spin.value() / 10000)) + def _update_pow_estimate(self) -> None: + """Show what the selected PoW target actually costs on this machine.""" + bits = self.pow_spin.value() + if bits <= 0: + self.pow_est_label.setText(_("disabled")) + return + try: + seconds = swap_pow.estimate_seconds(bits) + except Exception: + self.pow_est_label.setText("") + return + text = _("~{} to compute").format(swap_pow.format_duration(seconds)) + if bits > POW_TARGET_WARN_ABOVE: + text = "" + text + "" + self.pow_est_label.setText(text) + def _relays_from_widget(self) -> str: raw = self.relays_edit.toPlainText().replace("\n", ",") parts = [p.strip() for p in raw.split(",") if p.strip()] @@ -167,18 +199,35 @@ def _relays_from_widget(self) -> str: def on_save(self) -> None: new_port = self.port_spin.value() or None new_relays = self._relays_from_widget() + new_pow = self.pow_spin.value() old_port = self.config.SWAPSERVER_PORT old_relays = self.config.NOSTR_RELAYS or "" + old_pow = int(self.config.SWAPSERVER_POW_TARGET or 0) + + # A higher target means the stored nonce no longer qualifies and the + # server has to grind a new one, which can take a long time. Make that + # explicit rather than silently burning CPU for hours. + if new_pow > old_pow and new_pow > POW_TARGET_WARN_ABOVE: + est = swap_pow.format_duration(swap_pow.estimate_seconds(new_pow)) + if not self.window.question( + _("A proof-of-work target of {} bits is expected to take about {} " + "to compute on this machine.").format(new_pow, est) + "\n\n" + + _("The swap server cannot announce over nostr until it finishes. " + "Continue?")): + return + # persist self.config.SWAPSERVER_PORT = new_port self.config.SWAPSERVER_FEE_MILLIONTHS = self.fee_spin.value() - self.config.SWAPSERVER_POW_TARGET = self.pow_spin.value() + self.config.SWAPSERVER_POW_TARGET = new_pow self.config.NOSTR_RELAYS = new_relays self.load_settings_into_widgets() - # Only a port/relay change needs a server restart; the fee and PoW target - # are read live on each request/announcement. Restarting on every save - # would needlessly bounce the (busy) nostr transport. - transport_changed = (new_port != old_port) or (new_relays != old_relays) + # A port/relay change needs a server restart. So does *raising* the PoW + # target, since the running server is announcing with a nonce that no + # longer meets it. The fee is read live on each request, and lowering the + # target keeps the existing nonce valid, so neither needs a bounce. + transport_changed = (new_port != old_port) or (new_relays != old_relays) \ + or (new_pow > old_pow) if self.plugin.is_running() and transport_changed: self.plugin.stop_server() try: @@ -202,8 +251,21 @@ def on_toggle(self) -> None: self.config.SWAPSERVER_GUI_AUTOSTART = True self.refresh() + # -------------------------------------------------------------- teardown + def clean_up(self) -> None: + """Stop the periodic refresh. Idempotent; safe to call during shutdown.""" + if self._timer is not None: + self._timer.stop() + try: + self._timer.timeout.disconnect(self.refresh) + except TypeError: + pass # already disconnected + self._timer = None + # -------------------------------------------------------------- refresh def refresh(self) -> None: + if self._timer is None: + return # torn down; the wallet/loop may already be gone self.plugin.request_pairs_update() st = self.plugin.status() running = st["running"] @@ -228,12 +290,29 @@ def refresh(self) -> None: if not st["nostr_enabled"]: nostr_txt = _("disabled") + elif running and st["pow_grinding"]: + nostr_txt = _("waiting for proof of work") elif running: nostr_txt = _("announcing to {} relay(s)").format(st["nostr_relay_count"]) else: nostr_txt = _("{} relay(s) configured").format(st["nostr_relay_count"]) self._out_labels["nostr"].setText(nostr_txt) + target = st["pow_target"] + if not st["nostr_enabled"] or not target: + pow_txt = _("not required") + elif st["pow_grinding"]: + # There is no meaningful "percent done": each nonce is an independent + # trial, so we report scanned hashes and the best result so far. + pow_txt = _("computing… best {best}/{target} bits, {hashes} hashes scanned").format( + best=st["pow_best_bits"], target=target, + hashes=f"{st['pow_hashes_done']:,}") + elif st["pow_ready"] and running: + pow_txt = _("ready ({} bits)").format(target) + else: + pow_txt = _("target {} bits").format(target) + self._out_labels["pow"].setText(pow_txt) + pct = st["percentage"] self._out_labels["percentage"].setText("—" if pct is None else "{:.4f} %".format(pct)) self._out_labels["min_amount"].setText(_fmt_sat(self.config, st["min_amount"])) @@ -302,9 +381,14 @@ def _add_tab(self, window: 'ElectrumWindow') -> None: def _remove_tab(self) -> None: if self._tab is None or self._window is None: return + # Stop the refresh timer first: it calls back into the plugin (and hence + # into get_asyncio_loop()) every 4s, and during shutdown the wallet and + # the asyncio loop are torn down underneath it. + self._tab.clean_up() idx = self._window.tabs.indexOf(self._tab) if idx != -1: self._window.tabs.removeTab(idx) + self._tab.deleteLater() self._tab = None self._window = None diff --git a/plugins/swapserver_gui/swapserver_gui.py b/plugins/swapserver_gui/swapserver_gui.py index 268d3b3..d5fd3f2 100644 --- a/plugins/swapserver_gui/swapserver_gui.py +++ b/plugins/swapserver_gui/swapserver_gui.py @@ -17,6 +17,13 @@ # ``ManagedHttpSwapServer`` keeps the ``AppRunner`` so we can shut it down. # * The nostr server (``SwapManager.run_nostr_server``) is a long-running # coroutine that cleans up when cancelled, so for it we just cancel the task. +# * Both ``run_nostr_server`` and ``SwapManager.set_nostr_proof_of_work`` route +# into ``electrum.util.gen_nostr_ann_pow``, which *deadlocks the event loop +# thread* if cancelled while grinding -- and cancelling is exactly what we do +# at shutdown. That is what made Electrum hang on exit. We therefore +# compute the announcement proof-of-work ourselves (``pow.py``) before +# starting the nostr transport, so upstream always finds a good cached nonce +# and never enters that function. See ``pow.py`` for the full analysis. import asyncio import concurrent.futures @@ -29,6 +36,8 @@ from electrum.address_synchronizer import TX_HEIGHT_UNCONFIRMED from electrum.submarine_swaps import NostrTransport +from . import pow as swap_pow + # Importing the bundled swapserver plugin's server module has the useful side # effect of registering the shared config vars (plugins.swapserver.port etc.) # via electrum/plugins/swapserver/__init__.py. Reusing it avoids duplicating @@ -107,6 +116,12 @@ def __init__(self, parent: Any, config: 'SimpleConfig', name: str) -> None: self._nostr_fut: Optional['concurrent.futures.Future'] = None self._publish_now_fut: Optional['concurrent.futures.Future'] = None self._running: bool = False + # Set once the announcement proof-of-work is usable (or once we know we + # cannot compute one, e.g. no nostr keypair). Gates the nostr announce + # path so we never publish an offer that takers would reject. + self._pow_gate: asyncio.Event = asyncio.Event() + self._pow_state: Optional['swap_pow.PowState'] = None + self._pow_grinding: bool = False # ------------------------------------------------------------------ utils @property @@ -141,6 +156,103 @@ def _cancel_fut(fut: Optional['concurrent.futures.Future']) -> None: if fut is not None: fut.cancel() + # ------------------------------------------------------------ proof of work + def _nostr_pubkey(self) -> Optional[bytes]: + """The x-only nostr pubkey the announcement PoW is bound to, if any. + + Upstream hashes ``b'electrum-' + keypair.pubkey[1:]`` (the compressed + pubkey without its prefix byte), so we must use exactly the same bytes. + """ + if self.wallet is None or self.wallet.lnworker is None: + return None + keypair = getattr(self.wallet.lnworker, "nostr_keypair", None) + pubkey = getattr(keypair, "pubkey", None) + if not isinstance(pubkey, (bytes, bytearray)) or len(pubkey) < 2: + return None + return bytes(pubkey[1:]) + + def _save_pow_state(self, state: 'swap_pow.PowState') -> None: + self._pow_state = state + try: + self.config.SWAPSERVER_GUI_POW_STATE = state.dumps() + except Exception: + self.logger.debug("could not persist proof-of-work state", exc_info=True) + + async def ensure_pow_nonce(self) -> bool: + """Make sure ``SWAPSERVER_ANN_POW_NONCE`` satisfies ``SWAPSERVER_POW_TARGET``. + + Returns True when a usable nonce is in place. Always opens + :attr:`_pow_gate` before returning so the announce path is never + blocked forever, even when we cannot compute a proof of work. + + This is cancellable and, unlike upstream's ``gen_nostr_ann_pow``, + cancelling it neither blocks nor wedges the event loop thread. + """ + pubkey = self._nostr_pubkey() + if pubkey is None: + # No nostr keypair (or a stand-in without one): nothing to grind + # against. Let the caller proceed; upstream will report the problem. + self.logger.info("no nostr keypair available; skipping announcement proof-of-work") + self._pow_gate.set() + return False + + target = int(self.config.SWAPSERVER_POW_TARGET or 0) + if swap_pow.pow_bits(pubkey, self.config.SWAPSERVER_ANN_POW_NONCE) >= target: + self.logger.debug("reusing cached nostr announcement proof-of-work") + self._pow_gate.set() + return True + + state = swap_pow.PowState.load( + self.config.SWAPSERVER_GUI_POW_STATE, pubkey=pubkey, target=target) + self._pow_state = state + # A previously-found "best" nonce may already satisfy a lowered target. + if state.best_nonce is not None and state.best_bits >= target: + self.logger.info(f"reusing best-known nonce ({state.best_bits} bits) for target {target}") + self.config.SWAPSERVER_ANN_POW_NONCE = state.best_nonce + self._pow_gate.set() + return True + + est = swap_pow.format_duration(swap_pow.estimate_seconds(target)) + self.logger.info( + f"generating nostr announcement proof-of-work: target={target} bits, " + f"resuming from {state.hashes_done()} hashes already scanned, " + f"estimated {est}") + self._pow_grinding = True + try: + result = await swap_pow.grind( + pubkey=pubkey, + target_bits=target, + state=state, + on_progress=self._save_pow_state, + ) + except asyncio.CancelledError: + # Search cursors were already persisted by pow.grind's cancel path, + # so the next attempt resumes instead of re-scanning. + self.logger.info("announcement proof-of-work cancelled; progress saved") + raise + finally: + self._pow_grinding = False + self._pow_gate.set() + + if not result.found or result.nonce is None: + self.logger.warning("announcement proof-of-work ended without a solution") + return False + self.config.SWAPSERVER_ANN_POW_NONCE = result.nonce + self.logger.info(f"found nostr announcement proof-of-work: {result.bits} bits") + return True + + async def _nostr_startup(self) -> None: + """Compute the PoW ourselves, then hand over to upstream's nostr server. + + Ordering matters: ``run_nostr_server`` begins with + ``set_nostr_proof_of_work``, which grinds inside the un-cancellable + upstream helper. By seeding a good nonce first we guarantee it + short-circuits, so cancelling this task at shutdown is always safe. + """ + await self.ensure_pow_nonce() + assert self._sm is not None + await self._sm.run_nostr_server() + # -------------------------------------------------------------- lifecycle def bind_wallet(self, wallet: 'Abstract_Wallet') -> None: """Associate this plugin instance with a wallet's swap manager.""" @@ -173,13 +285,16 @@ def start_server(self) -> None: port = self.config.SWAPSERVER_PORT relays = (self.config.NOSTR_RELAYS or "").strip() + self._pow_gate = asyncio.Event() # fresh gate for this run if port: server = ManagedHttpSwapServer(self.config, self.wallet) sm.http_server = server self._http_fut = self._spawn(server.run(), "http") if relays: - self._nostr_fut = self._spawn(sm.run_nostr_server(), "nostr") + # _nostr_startup seeds the proof-of-work before starting upstream's + # nostr server; see the note in the module docstring. + self._nostr_fut = self._spawn(self._nostr_startup(), "nostr") self._running = True self.logger.info(f"swap server started (http_port={port or None}, " @@ -227,13 +342,12 @@ async def _one_shot_publish(self) -> None: keypair = getattr(self.wallet.lnworker, "nostr_keypair", None) if keypair is None: return - # Ensure the announcement carries a valid proof-of-work nonce (reuses a - # cached one instantly; only grinds on a brand-new server key). - try: - await sm.set_nostr_proof_of_work() - except Exception: - self.logger.debug("set_nostr_proof_of_work failed before publish_now", - exc_info=True) + # Wait for the announcement proof-of-work instead of computing one here. + # We must NOT call sm.set_nostr_proof_of_work(): it routes into + # electrum.util.gen_nostr_ann_pow, which wedges the event loop thread if + # this task is cancelled mid-grind. _nostr_startup owns the PoW and + # opens the gate; that path is cancel-safe. Waiting here is cancellable. + await self._pow_gate.wait() loop = asyncio.get_running_loop() deadline = loop.time() + self.PUBLISH_NOW_LIQUIDITY_WAIT_SEC while True: @@ -277,7 +391,16 @@ def stop_server(self) -> None: # Stop the HTTP listener (needs an explicit aiohttp runner cleanup). # Schedule it on the loop fire-and-forget; do NOT block the GUI thread. if sm is not None and isinstance(getattr(sm, 'http_server', None), ManagedHttpSwapServer): - self._spawn(sm.http_server.stop(), "http-stop") + http_server = sm.http_server + # Unregister the EventListener synchronously: the cleanup coroutine + # below can be cancelled by the loop shutting down before it runs, + # and a stale registration would keep a stopped wallet's callbacks + # alive in the global registry. + try: + http_server.unregister_callbacks() + except Exception: + self.logger.debug("unregister_callbacks failed", exc_info=True) + self._spawn(http_server.stop(), "http-stop") sm.http_server = None self._cancel_fut(self._http_fut) self._cancel_fut(self._nostr_fut) @@ -288,6 +411,7 @@ def stop_server(self) -> None: if sm is not None: sm.is_server = False self._running = False + self._pow_grinding = False self.logger.info("swap server stopped") def request_pairs_update(self) -> None: @@ -318,6 +442,11 @@ def status(self) -> Dict[str, Any]: ) if sm is not None else False, "nostr_enabled": bool(relays), "nostr_relay_count": len(relays), + "pow_target": int(self.config.SWAPSERVER_POW_TARGET or 0), + "pow_ready": self._pow_gate.is_set() and not self._pow_grinding, + "pow_grinding": self._pow_grinding, + "pow_best_bits": self._pow_state.best_bits if self._pow_state else 0, + "pow_hashes_done": self._pow_state.hashes_done() if self._pow_state else 0, "percentage": None, "min_amount": None, "max_forward": None, diff --git a/tests/test_pow.py b/tests/test_pow.py new file mode 100644 index 0000000..52ca0a9 --- /dev/null +++ b/tests/test_pow.py @@ -0,0 +1,319 @@ +#!/usr/bin/env python3 +"""Unit tests for the cancel-safe, resumable announcement proof-of-work. + +The headline test is :class:`CancelSafetyTests` -- it is the regression test for +the bug where quitting Electrum mid-grind wedged the asyncio loop thread and the +process never exited. See plugins/swapserver_gui/pow.py for the analysis. + +Run with: python3 -m pytest tests/test_pow.py +""" +import asyncio +import json +import os +import sys +import threading +import time +import unittest + +# --- make electrum + the plugin importable --------------------------------- +# pow.py itself has no electrum dependency, but importing it as part of the +# swapserver_gui package runs __init__.py, which registers config vars. +_HERE = os.path.dirname(os.path.abspath(__file__)) +_PROJECT_ROOT = os.path.dirname(os.path.dirname(_HERE)) # /home/user/electrum_swapgui +_ELECTRUM_SRC = os.environ.get("ELECTRUM_SRC", os.path.join(_PROJECT_ROOT, "electrum")) +_PLUGINS_DIR = os.path.join(os.path.dirname(_HERE), "plugins") +for _p in (_ELECTRUM_SRC, _PLUGINS_DIR): + if _p not in sys.path: + sys.path.insert(0, _p) + +from swapserver_gui import pow as swap_pow # noqa: E402 + +PUBKEY = bytes(range(32)) +OTHER_PUBKEY = bytes(range(1, 33)) + +# A target that is found near-instantly, for the "does it actually work" tests. +EASY_TARGET = 8 +# A target that will never be hit during a test, so the grind is guaranteed to +# still be running when we cancel it. +IMPOSSIBLE_TARGET = 96 + + +class _LoopThread: + """A real asyncio loop on a background thread, like Electrum's 'EventLoop'. + + Non-daemon on purpose in the shutdown tests: that is precisely what makes a + wedged loop thread able to hang the whole interpreter. + """ + + def __init__(self, *, daemon: bool = True) -> None: + self.loop = asyncio.new_event_loop() + self.thread = threading.Thread(target=self._run, name="EventLoop", daemon=daemon) + + def _run(self) -> None: + asyncio.set_event_loop(self.loop) + self.loop.run_forever() + + def __enter__(self) -> asyncio.AbstractEventLoop: + self.thread.start() + return self.loop + + def __exit__(self, *exc: object) -> None: + # let in-flight cancellations settle before stopping, so the loop does + # not complain about tasks destroyed while pending + time.sleep(0.2) + self.loop.call_soon_threadsafe(self.loop.stop) + self.thread.join(timeout=10) + if not self.thread.is_alive(): + self.loop.close() + + def loop_is_responsive(self, timeout: float = 10.0) -> bool: + """True if the loop can still execute a trivial coroutine.""" + try: + fut = asyncio.run_coroutine_threadsafe(asyncio.sleep(0), self.loop) + fut.result(timeout=timeout) + return True + except Exception: + return False + + +class PowBitsTests(unittest.TestCase): + def test_matches_upstream_implementation(self): + # guard against drift from electrum.util.get_nostr_ann_pow_amount + try: + from electrum.util import get_nostr_ann_pow_amount + except ImportError: + self.skipTest("electrum not importable") + for nonce in (0, 1, 12345, 2 ** 100 + 7): + self.assertEqual(swap_pow.pow_bits(PUBKEY, nonce), + get_nostr_ann_pow_amount(PUBKEY, nonce)) + + def test_zero_nonce_is_zero_bits(self): + # upstream treats nonce 0 as "no PoW", which is why a fresh server + # always has to grind + self.assertEqual(swap_pow.pow_bits(PUBKEY, 0), 0) + self.assertEqual(swap_pow.pow_bits(PUBKEY, None), 0) + + +class PowStateTests(unittest.TestCase): + def test_round_trip(self): + state = swap_pow.PowState( + pubkey_hex=PUBKEY.hex(), target=30, + offsets=[10, 20, 30], best_nonce=None, best_bits=0) + restored = swap_pow.PowState.load(state.dumps(), pubkey=PUBKEY, target=30) + self.assertEqual(restored.offsets, [10, 20, 30]) + self.assertEqual(restored.hashes_done(), 60) + + def test_best_nonce_round_trips_as_big_int(self): + nonce = swap_pow.lane_start(3) + 987654321 + bits = swap_pow.pow_bits(PUBKEY, nonce) + state = swap_pow.PowState(pubkey_hex=PUBKEY.hex(), target=4, + offsets=[1], best_nonce=nonce, best_bits=bits) + restored = swap_pow.PowState.load(state.dumps(), pubkey=PUBKEY, target=4) + self.assertEqual(restored.best_nonce, nonce) + self.assertEqual(restored.best_bits, bits) + + def test_discarded_on_pubkey_change(self): + # the pubkey is part of the hash preimage, so cursors from another + # wallet's nostr key say nothing about this one + state = swap_pow.PowState(pubkey_hex=PUBKEY.hex(), target=30, offsets=[999]) + restored = swap_pow.PowState.load(state.dumps(), pubkey=OTHER_PUBKEY, target=30) + self.assertEqual(restored.offsets, []) + self.assertEqual(restored.hashes_done(), 0) + + def test_kept_when_target_raised(self): + state = swap_pow.PowState(pubkey_hex=PUBKEY.hex(), target=20, offsets=[500]) + restored = swap_pow.PowState.load(state.dumps(), pubkey=PUBKEY, target=30) + self.assertEqual(restored.offsets, [500]) + + def test_kept_when_target_lowered(self): + # Safe because best_bits is the maximum over everything scanned: if the + # best is below the new target, nothing in the scanned range meets it. + state = swap_pow.PowState(pubkey_hex=PUBKEY.hex(), target=30, offsets=[500], + best_nonce=None, best_bits=0) + restored = swap_pow.PowState.load(state.dumps(), pubkey=PUBKEY, target=20) + self.assertEqual(restored.offsets, [500]) + + def test_corrupt_input_yields_blank_state(self): + for raw in ("", None, "not json", "[]", '{"offsets": [-1]}', + '{"pubkey": "zz", "offsets": "x"}'): + restored = swap_pow.PowState.load(raw, pubkey=PUBKEY, target=30) + self.assertEqual(restored.offsets, []) + self.assertEqual(restored.pubkey_hex, PUBKEY.hex()) + + def test_lying_best_bits_is_recomputed(self): + # a hand-edited config must never make us publish a bad nonce + raw = json.dumps({"pubkey": PUBKEY.hex(), "target": 30, "offsets": [1], + "best_nonce": "12345", "best_bits": 250}) + restored = swap_pow.PowState.load(raw, pubkey=PUBKEY, target=30) + self.assertEqual(restored.best_bits, swap_pow.pow_bits(PUBKEY, 12345)) + self.assertLess(restored.best_bits, 250) + + +class GrindTests(unittest.TestCase): + def test_finds_a_valid_nonce(self): + state = swap_pow.PowState(pubkey_hex=PUBKEY.hex(), target=EASY_TARGET) + result = asyncio.run(swap_pow.grind( + pubkey=PUBKEY, target_bits=EASY_TARGET, state=state, num_workers=2)) + self.assertTrue(result.found) + self.assertIsNotNone(result.nonce) + # the returned nonce must genuinely meet the target + self.assertGreaterEqual(swap_pow.pow_bits(PUBKEY, result.nonce), EASY_TARGET) + self.assertGreaterEqual(result.bits, EASY_TARGET) + + def test_thread_fallback_finds_a_valid_nonce(self): + # num_workers=1 exercises the no-fork path (Windows / Android) + state = swap_pow.PowState(pubkey_hex=PUBKEY.hex(), target=EASY_TARGET) + result = asyncio.run(swap_pow.grind( + pubkey=PUBKEY, target_bits=EASY_TARGET, state=state, num_workers=1)) + self.assertTrue(result.found) + self.assertGreaterEqual(swap_pow.pow_bits(PUBKEY, result.nonce), EASY_TARGET) + + +class CancelSafetyTests(unittest.TestCase): + """Regression tests for the shutdown hang.""" + + def _start_grind(self, loop, state, *, num_workers=2): + return asyncio.run_coroutine_threadsafe( + swap_pow.grind(pubkey=PUBKEY, target_bits=IMPOSSIBLE_TARGET, + state=state, num_workers=num_workers, + poll_interval=0.05, persist_interval=0.1), + loop) + + def test_cancel_leaves_loop_responsive(self): + # THE regression test: upstream's gen_nostr_ann_pow calls + # executor.shutdown(wait=True) from __exit__ on the cancel path, which + # blocks the loop thread forever against workers that never stop. + state = swap_pow.PowState(pubkey_hex=PUBKEY.hex(), target=IMPOSSIBLE_TARGET) + harness = _LoopThread() + with harness as loop: + fut = self._start_grind(loop, state) + time.sleep(1.5) # let the workers actually start grinding + self.assertTrue(fut.cancel()) + self.assertTrue( + harness.loop_is_responsive(), + "event loop thread was wedged by cancelling the proof-of-work") + + def test_cancel_thread_backend_leaves_loop_responsive(self): + # num_workers=1 selects the no-fork fallback (Windows/Android, and CI + # runners with 2 cores). It must be just as cancel-safe. + state = swap_pow.PowState(pubkey_hex=PUBKEY.hex(), target=IMPOSSIBLE_TARGET) + harness = _LoopThread() + with harness as loop: + fut = self._start_grind(loop, state, num_workers=1) + time.sleep(1.5) + self.assertTrue(fut.cancel()) + self.assertTrue(harness.loop_is_responsive(), + "thread backend wedged the event loop on cancel") + + def test_cancelled_workers_are_reaped(self): + state = swap_pow.PowState(pubkey_hex=PUBKEY.hex(), target=IMPOSSIBLE_TARGET) + backend_box = {} + real_start = swap_pow._Backend.start + + def _spy(self_backend, **kwargs): + real_start(self_backend, **kwargs) + backend_box["backend"] = self_backend + + swap_pow._Backend.start = _spy + try: + harness = _LoopThread() + with harness as loop: + fut = self._start_grind(loop, state) + time.sleep(1.5) + backend = backend_box["backend"] + self.assertTrue(backend.any_alive()) + fut.cancel() + harness.loop_is_responsive() + deadline = time.time() + 20 + while backend.any_alive() and time.time() < deadline: + time.sleep(0.1) + self.assertFalse(backend.any_alive(), + "proof-of-work workers survived cancellation") + finally: + swap_pow._Backend.start = real_start + + def test_cancel_does_not_block_the_caller(self): + # stop_server() calls this from the GUI thread during shutdown; it must + # return promptly rather than joining worker processes. + state = swap_pow.PowState(pubkey_hex=PUBKEY.hex(), target=IMPOSSIBLE_TARGET) + harness = _LoopThread() + with harness as loop: + fut = self._start_grind(loop, state) + time.sleep(1.5) + t0 = time.monotonic() + fut.cancel() + self.assertTrue(harness.loop_is_responsive()) + elapsed = time.monotonic() - t0 + self.assertLess(elapsed, 5.0, f"cancellation blocked for {elapsed:.2f}s") + + def test_non_daemon_loop_thread_still_exits(self): + # The real failure mode: Electrum's loop thread is non-daemon, so a + # wedged loop means the interpreter can never exit. + state = swap_pow.PowState(pubkey_hex=PUBKEY.hex(), target=IMPOSSIBLE_TARGET) + harness = _LoopThread(daemon=False) + with harness as loop: + fut = self._start_grind(loop, state) + time.sleep(1.5) + fut.cancel() + self.assertFalse(harness.thread.is_alive(), + "non-daemon EventLoop thread did not exit -> " + "the process would hang at interpreter shutdown") + + def test_progress_is_persisted_on_cancel(self): + saved = [] + state = swap_pow.PowState(pubkey_hex=PUBKEY.hex(), target=IMPOSSIBLE_TARGET) + harness = _LoopThread() + with harness as loop: + fut = asyncio.run_coroutine_threadsafe( + swap_pow.grind(pubkey=PUBKEY, target_bits=IMPOSSIBLE_TARGET, + state=state, num_workers=2, poll_interval=0.05, + persist_interval=0.1, + on_progress=lambda s: saved.append(s.dumps())), + loop) + time.sleep(2.5) + fut.cancel() + harness.loop_is_responsive() + self.assertTrue(saved, "no progress was persisted") + final = swap_pow.PowState.load(saved[-1], pubkey=PUBKEY, target=IMPOSSIBLE_TARGET) + self.assertGreater(final.hashes_done(), 0, + "cancelled grind saved no scanned-nonce cursor") + + def test_resume_continues_past_saved_cursor(self): + # work must accumulate across cancels rather than re-scanning the same + # (deterministically chosen) range, which is what upstream does + state = swap_pow.PowState(pubkey_hex=PUBKEY.hex(), target=IMPOSSIBLE_TARGET, + offsets=[5_000_000]) + harness = _LoopThread() + with harness as loop: + fut = self._start_grind(loop, state, num_workers=1) + time.sleep(2.5) + fut.cancel() + harness.loop_is_responsive() + self.assertGreater(state.offsets[0], 5_000_000, + "resumed grind did not continue past the saved cursor") + + +class EstimateTests(unittest.TestCase): + def test_estimate_doubles_per_bit(self): + a = swap_pow.estimate_seconds(20, num_workers=1, hash_rate=1e6) + b = swap_pow.estimate_seconds(21, num_workers=1, hash_rate=1e6) + self.assertAlmostEqual(b / a, 2.0, places=6) + + def test_estimate_scales_with_workers(self): + one = swap_pow.estimate_seconds(20, num_workers=1, hash_rate=1e6) + four = swap_pow.estimate_seconds(20, num_workers=4, hash_rate=1e6) + self.assertAlmostEqual(one / four, 4.0, places=6) + + def test_zero_target_is_free(self): + self.assertEqual(swap_pow.estimate_seconds(0), 0.0) + + def test_format_duration(self): + self.assertEqual(swap_pow.format_duration(0.2), "< 1s") + self.assertEqual(swap_pow.format_duration(30), "30s") + self.assertIn("min", swap_pow.format_duration(600)) + self.assertIn("hours", swap_pow.format_duration(7200)) + self.assertIn("days", swap_pow.format_duration(400000)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_shutdown_e2e.py b/tests/test_shutdown_e2e.py new file mode 100644 index 0000000..00c967f --- /dev/null +++ b/tests/test_shutdown_e2e.py @@ -0,0 +1,374 @@ +#!/usr/bin/env python3 +"""End-to-end shutdown tests: the wallet must close even mid-proof-of-work. + +These reproduce the reported bug ("Electrum hangs on shutdown and never fully +shuts down") against the real plugin, driving the same sequence Electrum uses: + + 1. ``create_and_start_event_loop`` starts the asyncio loop on a **non-daemon** + thread called 'EventLoop' (electrum/util.py). + 2. On quit, the Qt layer runs the ``close_wallet`` / ``on_close_window`` hooks, + which call ``plugin.stop_server()``. + 3. ``daemon.stop()`` is awaited *on that loop* from the GUI thread with a + blocking ``.result()`` and no timeout (electrum/daemon.py:694). + 4. ``run_electrum:sys_exit`` resolves ``stopping_fut``; the loop thread then + cancels every remaining task and exits. + +If anything blocks the loop thread in step 2-4, step 3 never returns and the +non-daemon thread in step 1 keeps the interpreter alive forever. Before the fix +that is exactly what cancelling the proof-of-work did. + +The 'subprocess' test is the strongest form: it asserts a *whole interpreter* +running this sequence actually exits, which is what the user observes. + +Run with: python3 -m pytest tests/test_shutdown_e2e.py +""" +import asyncio +import os +import subprocess +import sys +import textwrap +import threading +import time +import unittest +from unittest import mock + +# --- make electrum + the plugin importable --------------------------------- +_HERE = os.path.dirname(os.path.abspath(__file__)) +_PROJECT_ROOT = os.path.dirname(os.path.dirname(_HERE)) # /home/user/electrum_swapgui +_ELECTRUM_SRC = os.environ.get("ELECTRUM_SRC", os.path.join(_PROJECT_ROOT, "electrum")) +_PLUGINS_DIR = os.path.join(os.path.dirname(_HERE), "plugins") +for _p in (_ELECTRUM_SRC, _PLUGINS_DIR): + if _p not in sys.path: + sys.path.insert(0, _p) + +from swapserver_gui.swapserver_gui import SwapServerGuiPlugin # noqa: E402 +from swapserver_gui import pow as swap_pow # noqa: E402 + +# High enough that the grind is guaranteed to still be running when we shut down. +IMPOSSIBLE_TARGET = 96 +# A real 33-byte compressed pubkey shape; the plugin uses pubkey[1:]. +PUBKEY_33 = bytes([2]) + bytes(range(32)) + + +class _Keypair: + def __init__(self, pubkey: bytes) -> None: + self.pubkey = pubkey + + +class _Config: + """Minimal SimpleConfig stand-in, including the PoW state var.""" + + def __init__(self, *, port=None, relays="wss://relay.one", pow_target=IMPOSSIBLE_TARGET): + self.SWAPSERVER_PORT = port + self.NOSTR_RELAYS = relays + self.SWAPSERVER_FEE_MILLIONTHS = 5000 + self.SWAPSERVER_POW_TARGET = pow_target + self.SWAPSERVER_ANN_POW_NONCE = 0 + self.SWAPSERVER_GUI_AUTOSTART = False + self.SWAPSERVER_GUI_POW_STATE = '' + + +class _SwapManager: + def __init__(self): + self.is_server = False + self.http_server = None + self.percentage = None + self._min_amount = 20000 + self._max_forward = None + self._max_reverse = None + self.mining_fee = None + self.pow_calls = 0 + self.nostr_started = threading.Event() + + async def run_nostr_server(self): + self.nostr_started.set() + await asyncio.Event().wait() + + async def set_nostr_proof_of_work(self): + self.pow_calls += 1 + + def server_update_pairs(self): + self.percentage = 0.5 + + +def _make_wallet(sm, *, pubkey=PUBKEY_33): + wallet = mock.MagicMock() + wallet.lnworker.swap_manager = sm + wallet.lnworker.nostr_keypair = _Keypair(pubkey) + wallet.has_password.return_value = False + return wallet + + +class _ElectrumLikeLoop: + """Mirrors electrum.util.create_and_start_event_loop closely enough to matter.""" + + def __init__(self): + self.loop = asyncio.new_event_loop() + self.stopping_fut = self.loop.create_future() + # NON-daemon, exactly like Electrum's 'EventLoop' thread. This is the + # property that turns a wedged loop into a process that never exits. + self.thread = threading.Thread(target=self._run, name="EventLoop", daemon=False) + + def _run(self): + asyncio.set_event_loop(self.loop) + try: + self.loop.run_until_complete(self.stopping_fut) + finally: + pending = asyncio.gather(*asyncio.all_tasks(self.loop), return_exceptions=True) + pending.cancel() + try: + self.loop.run_until_complete(pending) + except asyncio.CancelledError: + pass + self.loop.run_until_complete(self.loop.shutdown_asyncgens()) + self.loop.close() + + def __enter__(self): + self.thread.start() + t0 = time.monotonic() + while not self.loop.is_running(): + time.sleep(0.01) + if time.monotonic() - t0 > 5: + raise RuntimeError("loop would not start") + return self.loop + + def stop_and_join(self, timeout=20.0): + """The run_electrum:sys_exit sequence. Returns True if the thread exited.""" + self.loop.call_soon_threadsafe(self.stopping_fut.set_result, 1) + self.thread.join(timeout=timeout) + return not self.thread.is_alive() + + def __exit__(self, *exc): + if self.thread.is_alive(): + self.stop_and_join() + + +class WalletCloseDuringPowTests(unittest.TestCase): + """The wallet must close cleanly while the proof-of-work is still running.""" + + def _running_plugin(self, loop, config, sm): + plugin = SwapServerGuiPlugin(mock.MagicMock(), config, "swapserver_gui") + plugin.bind_wallet(_make_wallet(sm)) + plugin.start_server() + return plugin + + def test_stop_server_mid_pow_leaves_loop_responsive(self): + config = _Config() + sm = _SwapManager() + harness = _ElectrumLikeLoop() + with harness as loop: + with mock.patch("swapserver_gui.swapserver_gui.get_asyncio_loop", return_value=loop): + plugin = self._running_plugin(loop, config, sm) + # wait until the grind is genuinely underway + deadline = time.time() + 20 + while not plugin.status()["pow_grinding"] and time.time() < deadline: + time.sleep(0.05) + self.assertTrue(plugin.status()["pow_grinding"], "PoW never started") + self.assertFalse(sm.nostr_started.is_set(), + "nostr server started before the PoW was ready") + + plugin.stop_server() # the close_wallet / on_close_window hook + + probe = asyncio.run_coroutine_threadsafe(asyncio.sleep(0), loop) + probe.result(timeout=15) # raises if the loop thread is wedged + self.assertFalse(plugin.is_running()) + self.assertFalse(sm.is_server) + + def test_stop_server_mid_pow_does_not_block_gui_thread(self): + # stop_server() runs on the Qt GUI thread; blocking it there is what + # makes Electrum's window freeze instead of closing. + config = _Config() + sm = _SwapManager() + harness = _ElectrumLikeLoop() + with harness as loop: + with mock.patch("swapserver_gui.swapserver_gui.get_asyncio_loop", return_value=loop): + plugin = self._running_plugin(loop, config, sm) + deadline = time.time() + 20 + while not plugin.status()["pow_grinding"] and time.time() < deadline: + time.sleep(0.05) + t0 = time.monotonic() + plugin.stop_server() + elapsed = time.monotonic() - t0 + self.assertLess(elapsed, 2.0, + f"stop_server blocked the GUI thread for {elapsed:.2f}s") + + def test_event_loop_thread_exits_after_close_mid_pow(self): + # The reported symptom: Electrum "never fully shuts down". + config = _Config() + sm = _SwapManager() + harness = _ElectrumLikeLoop() + with harness as loop: + with mock.patch("swapserver_gui.swapserver_gui.get_asyncio_loop", return_value=loop): + plugin = self._running_plugin(loop, config, sm) + deadline = time.time() + 20 + while not plugin.status()["pow_grinding"] and time.time() < deadline: + time.sleep(0.05) + plugin.stop_server() + exited = harness.stop_and_join(timeout=25) + self.assertTrue(exited, + "the non-daemon EventLoop thread never exited: the " + "Electrum process would hang forever on shutdown") + + def test_shutdown_without_stop_server_still_exits(self): + # Defence in depth: even if the plugin hooks never fire (e.g. the window + # is destroyed without close_wallet), the loop's own cancel-everything + # teardown must not wedge on the proof-of-work task. + config = _Config() + sm = _SwapManager() + harness = _ElectrumLikeLoop() + with harness as loop: + with mock.patch("swapserver_gui.swapserver_gui.get_asyncio_loop", return_value=loop): + plugin = self._running_plugin(loop, config, sm) + deadline = time.time() + 20 + while not plugin.status()["pow_grinding"] and time.time() < deadline: + time.sleep(0.05) + self.assertTrue(plugin.status()["pow_grinding"]) + # no stop_server() call at all + exited = harness.stop_and_join(timeout=25) + self.assertTrue(exited, "loop teardown wedged on the proof-of-work task") + + def test_daemon_stop_style_blocking_result_returns(self): + # This is the precise call that hangs in the field: daemon.run_gui()'s + # finally-block does + # asyncio.run_coroutine_threadsafe(self.stop(), self.asyncio_loop).result() + # from the GUI thread, with NO timeout (electrum/daemon.py:694). If the + # loop thread is wedged by the proof-of-work, this never returns and the + # Electrum window just sits there. + config = _Config() + sm = _SwapManager() + harness = _ElectrumLikeLoop() + with harness as loop: + with mock.patch("swapserver_gui.swapserver_gui.get_asyncio_loop", return_value=loop): + plugin = self._running_plugin(loop, config, sm) + deadline = time.time() + 20 + while not plugin.status()["pow_grinding"] and time.time() < deadline: + time.sleep(0.05) + self.assertTrue(plugin.status()["pow_grinding"]) + + plugin.stop_server() # close_wallet hook + + async def _fake_daemon_stop(): + # stands in for stopping wallets / network on the loop + await asyncio.sleep(0.1) + return "stopped" + + fut = asyncio.run_coroutine_threadsafe(_fake_daemon_stop(), loop) + # bounded here only so the test fails instead of hanging forever + self.assertEqual(fut.result(timeout=20), "stopped") + self.assertTrue(harness.stop_and_join(timeout=25)) + + def test_pow_progress_survives_the_shutdown(self): + # Quitting mid-grind must not throw the scanned range away. + config = _Config() + sm = _SwapManager() + harness = _ElectrumLikeLoop() + with harness as loop: + with mock.patch("swapserver_gui.swapserver_gui.get_asyncio_loop", return_value=loop), \ + mock.patch.object(swap_pow, "CHECKPOINT_BLOCK", 20_000): + plugin = self._running_plugin(loop, config, sm) + deadline = time.time() + 25 + while config.SWAPSERVER_GUI_POW_STATE == '' and time.time() < deadline: + time.sleep(0.1) + plugin.stop_server() + harness.stop_and_join(timeout=25) + self.assertNotEqual(config.SWAPSERVER_GUI_POW_STATE, '', + "no proof-of-work progress was persisted") + state = swap_pow.PowState.load(config.SWAPSERVER_GUI_POW_STATE, + pubkey=PUBKEY_33[1:], target=IMPOSSIBLE_TARGET) + self.assertGreater(state.hashes_done(), 0) + + def test_cached_nonce_skips_the_grind_entirely(self): + # The whole point: with a good cached nonce, upstream's + # set_nostr_proof_of_work short-circuits and never enters the deadlocking + # gen_nostr_ann_pow, so the nostr server starts immediately. + config = _Config(pow_target=4) + pubkey = PUBKEY_33[1:] + nonce = next(n for n in range(1, 100000) if swap_pow.pow_bits(pubkey, n) >= 4) + config.SWAPSERVER_ANN_POW_NONCE = nonce + sm = _SwapManager() + harness = _ElectrumLikeLoop() + with harness as loop: + with mock.patch("swapserver_gui.swapserver_gui.get_asyncio_loop", return_value=loop): + plugin = self._running_plugin(loop, config, sm) + self.assertTrue(sm.nostr_started.wait(timeout=10)) + self.assertFalse(plugin.status()["pow_grinding"]) + self.assertTrue(plugin.status()["pow_ready"]) + plugin.stop_server() + self.assertTrue(harness.stop_and_join(timeout=25)) + + +class InterpreterExitTests(unittest.TestCase): + """The end-to-end claim: a whole interpreter doing this actually exits.""" + + SCRIPT = textwrap.dedent( + """ + import asyncio, os, sys, threading, time + from unittest import mock + sys.path.insert(0, {electrum!r}) + sys.path.insert(0, {plugins!r}) + sys.path.insert(0, {tests!r}) + from test_shutdown_e2e import ( + _Config, _SwapManager, _make_wallet, _ElectrumLikeLoop, IMPOSSIBLE_TARGET) + from swapserver_gui.swapserver_gui import SwapServerGuiPlugin + + config = _Config(pow_target=IMPOSSIBLE_TARGET) + sm = _SwapManager() + harness = _ElectrumLikeLoop() + loop = harness.__enter__() + with mock.patch("swapserver_gui.swapserver_gui.get_asyncio_loop", return_value=loop): + plugin = SwapServerGuiPlugin(mock.MagicMock(), config, "swapserver_gui") + plugin.bind_wallet(_make_wallet(sm)) + plugin.start_server() + deadline = time.time() + 20 + while not plugin.status()["pow_grinding"] and time.time() < deadline: + time.sleep(0.05) + assert plugin.status()["pow_grinding"], "PoW never started" + print("GRINDING", flush=True) + plugin.stop_server() # close_wallet hook + harness.loop.call_soon_threadsafe(harness.stopping_fut.set_result, 1) + harness.thread.join(timeout=1) # run_electrum:sys_exit uses timeout=1 + print("EXITING", flush=True) + sys.exit(0) + """ + ) + + def test_process_exits_while_pow_is_running(self): + script = self.SCRIPT.format( + electrum=_ELECTRUM_SRC, plugins=_PLUGINS_DIR, tests=_HERE) + try: + proc = subprocess.run( + [sys.executable, "-c", script], + capture_output=True, text=True, timeout=90, + ) + except subprocess.TimeoutExpired: + self.fail("the interpreter never exited while the proof-of-work was " + "running -- this is the reported shutdown hang") + self.assertIn("GRINDING", proc.stdout) + self.assertIn("EXITING", proc.stdout) + self.assertEqual(proc.returncode, 0, f"stderr:\n{proc.stderr}") + + def test_no_orphan_worker_processes_survive(self): + # A daemonic multiprocessing.Process is killed with its parent, so the + # grind must not leave CPU-burning orphans behind after Electrum exits. + script = self.SCRIPT.format( + electrum=_ELECTRUM_SRC, plugins=_PLUGINS_DIR, tests=_HERE) + before = self._child_count() + proc = subprocess.run([sys.executable, "-c", script], + capture_output=True, text=True, timeout=90) + self.assertEqual(proc.returncode, 0, f"stderr:\n{proc.stderr}") + time.sleep(2) + self.assertLessEqual(self._child_count(), before, + "proof-of-work worker processes outlived the parent") + + @staticmethod + def _child_count(): + try: + out = subprocess.run(["pgrep", "-fc", "swap-pow"], + capture_output=True, text=True, timeout=10) + return int(out.stdout.strip() or 0) + except Exception: + return 0 + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_swapserver_gui.py b/tests/test_swapserver_gui.py index 8b63b98..008b35b 100644 --- a/tests/test_swapserver_gui.py +++ b/tests/test_swapserver_gui.py @@ -77,6 +77,11 @@ def server_update_pairs(self): self._max_reverse = 100000 +async def _never_finishing_pow(self): + """Stand-in for ensure_pow_nonce that never opens the gate (still cancellable).""" + await asyncio.Event().wait() + + def _make_wallet(sm, *, nostr_keypair=None): wallet = mock.MagicMock() wallet.lnworker.swap_manager = sm @@ -100,6 +105,14 @@ def __enter__(self): return self.loop def __exit__(self, *exc): + # Let cancelled tasks unwind before the loop goes away, otherwise + # asyncio logs "Task was destroyed but it is pending". Draining via a + # probe (rather than a fixed sleep) also covers tests that deliberately + # keep the loop busy for a while. + try: + asyncio.run_coroutine_threadsafe(asyncio.sleep(0.1), self.loop).result(timeout=5) + except Exception: + pass self.loop.call_soon_threadsafe(self.loop.stop) self.thread.join(timeout=5) self.loop.close() @@ -301,7 +314,47 @@ def test_start_server_publishes_immediately(self): self._fake_transport_cls(published)): p.start_server() self.assertTrue(published.wait(timeout=5)) - self.assertGreaterEqual(sm.pow_calls, 1) # PoW ensured before announce + # The plugin owns the proof-of-work now (see pow.py): calling + # upstream's set_nostr_proof_of_work from this task would route + # into gen_nostr_ann_pow, which wedges the event loop thread if + # cancelled mid-grind -- the shutdown-hang bug. + self.assertEqual(sm.pow_calls, 0) + p.stop_server() + + def test_publish_never_calls_upstream_pow(self): + # Explicit regression guard for the shutdown hang: no code path reached + # from publish_now may call SwapManager.set_nostr_proof_of_work. + sm = _SwapManager() + sm._advertise_liquidity = True + p = _make_plugin(_Config(relays="wss://relay.one")) + p.bind_wallet(_make_wallet(sm, nostr_keypair=object())) + published = threading.Event() + with _LoopThread() as loop: + with mock.patch("swapserver_gui.swapserver_gui.get_asyncio_loop", return_value=loop), \ + mock.patch("swapserver_gui.swapserver_gui.NostrTransport", + self._fake_transport_cls(published)): + p.start_server() + self.assertTrue(published.wait(timeout=5)) + p.stop_server() + self.assertEqual(sm.pow_calls, 0) + + def test_publish_waits_for_pow_gate(self): + # An announcement made before the PoW is ready carries a nonce that + # takers reject, so publishing must wait for the gate to open. + sm = _SwapManager() + sm._advertise_liquidity = True + p = _make_plugin(_Config(relays="wss://relay.one")) + p.bind_wallet(_make_wallet(sm, nostr_keypair=object())) + published = threading.Event() + with _LoopThread() as loop: + with mock.patch("swapserver_gui.swapserver_gui.get_asyncio_loop", return_value=loop), \ + mock.patch("swapserver_gui.swapserver_gui.NostrTransport", + self._fake_transport_cls(published)), \ + mock.patch.object(SwapServerGuiPlugin, "ensure_pow_nonce", + _never_finishing_pow): + p.start_server() + self.assertFalse(published.wait(timeout=1.5), + "announced before the proof-of-work was ready") p.stop_server() def test_publish_now_waits_for_liquidity(self):