Background transaction monitor.
Polls the wallet backend for new transactions and pushes a WebSocket
notification ({"txid": ..., "txdetails": {...}}) for each, so Jam and other
clients can refresh balances without polling (issue #560). Unlike the reference
implementation -- which only announced direct-send transactions -- this covers
every wallet transaction: external deposits, maker/taker coinjoins, and sends.
A transaction is announced twice at most: once when first seen (in the
mempool) and once when it first confirms (the txdetails then carries a
confirmations count). The direct-send endpoint announces its own
transaction inline and marks it so the monitor does not duplicate the
first-seen notification (it still reports the later confirmation).
Classes
Functions:
run_tx_monitor(state: DaemonState, poll_interval: float = _POLL_INTERVAL_SEC, ready: asyncio.Event | None = None, initialization_task: asyncio.Task[None] | None = None, baseline_existing: bool = True) -> None
async
Poll the backend and broadcast a notification per new/confirmed tx.
Runs until cancelled (on wallet lock). The first pass establishes a silent
baseline of pre-existing transactions so only activity that occurs after
the wallet is loaded is announced.
Source code in jmwalletd/src/jmwalletd/tx_monitor.py
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131 | async def run_tx_monitor(
state: DaemonState,
poll_interval: float = _POLL_INTERVAL_SEC,
ready: asyncio.Event | None = None,
initialization_task: asyncio.Task[None] | None = None,
baseline_existing: bool = True,
) -> None:
"""Poll the backend and broadcast a notification per new/confirmed tx.
Runs until cancelled (on wallet lock). The first pass establishes a silent
baseline of pre-existing transactions so only activity that occurs after
the wallet is loaded is announced.
"""
from jmwalletd.wallet_ops import _get_network
ws = state.wallet_service
if ws is None:
return
backend = ws.backend
network = _get_network()
# Neutrino registers watch addresses during wallet sync. Enumerating before
# that task completes can establish an incomplete baseline and later hide
# transactions discovered by the rescan as if they predated this session.
if initialization_task is not None:
await asyncio.shield(initialization_task)
cursor: str | None = None
confirmed_notified: set[str] = set()
first_pass = baseline_existing
if not first_pass and ready is not None:
ready.set()
while True:
try:
entries, next_cursor = await backend.list_wallet_transactions_since(cursor)
except asyncio.CancelledError:
raise
except Exception as exc: # pragma: no cover - transient RPC issues
logger.debug(f"tx monitor: enumeration failed: {exc}")
await asyncio.sleep(poll_interval)
continue
notification_failed = False
for entry in entries:
txid = entry.txid
confirmations = entry.confirmations
if first_pass:
# Baseline: mark pre-existing transactions as seen without
# announcing them, so we only notify on activity after load.
state._tx_broadcast_notified.add(txid)
if confirmations >= 1:
confirmed_notified.add(txid)
continue
try:
if txid not in state._tx_broadcast_notified:
notified = await _notify(
state,
backend,
network,
txid,
confirmations,
entry.raw,
suppress_if_first_seen=True,
)
if notified:
state._tx_broadcast_notified.add(txid)
if confirmations >= 1:
confirmed_notified.add(txid)
else:
notification_failed = True
elif confirmations >= 1 and txid not in confirmed_notified:
if await _notify(state, backend, network, txid, confirmations, entry.raw):
confirmed_notified.add(txid)
else:
notification_failed = True
except asyncio.CancelledError:
raise
except Exception as exc: # pragma: no cover - defensive
logger.debug(f"tx monitor: failed to notify {txid[:16]}...: {exc}")
notification_failed = True
# A confirmed transaction may only appear once for a given backend
# cursor. Keep the old cursor when notification construction failed so
# the backend returns that transaction again on the next poll.
if not notification_failed:
cursor = next_cursor
# The baseline is only complete once the backend actually enumerated
# (a non-None cursor). Before the wallet is set up/loaded the backend
# returns an empty list and a None cursor; staying in ``first_pass``
# until then avoids announcing the wallet's entire history the moment
# it finishes loading.
if first_pass and cursor is not None:
first_pass = False
if ready is not None:
ready.set()
await asyncio.sleep(poll_interval)
|