From 03ba7eacd9180b86a5c569abae68570a687580b2 Mon Sep 17 00:00:00 2001 From: Logan Cusano Date: Sun, 4 Oct 2026 18:16:31 -0400 Subject: [PATCH] Discord leases: free releasing tokens on gone nodes; stamp before building desired list - reconcile_once only walked desired=on streams, so a token left "releasing" by /leave while its node was offline (or a deleted node) never returned to the pool until that node came back (server-26#204 leak shape). Sweep leased tokens whose node is offline/deleted past the lease grace. - publish_desired stamped release_seq AFTER building the list, so a stop landing between the two got a seq whose list still held the stream; an ack for it could free a token the node was still told to use. Stamp first. Three regression tests; each fails on the previous commit. Co-Authored-By: Claude Opus 5.5 --- drb-c2-core/app/internal/discord_streams.py | 29 ++++++++++- drb-c2-core/tests/test_discord_leases.py | 54 +++++++++++++++++++++ 2 files changed, 82 insertions(+), 1 deletion(-) diff --git a/drb-c2-core/app/internal/discord_streams.py b/drb-c2-core/app/internal/discord_streams.py index 5353e7f..9c49ee4 100644 --- a/drb-c2-core/app/internal/discord_streams.py +++ b/drb-c2-core/app/internal/discord_streams.py @@ -175,8 +175,12 @@ async def publish_desired(node_id: str) -> bool: from app.internal.mqtt_handler import mqtt_handler async with _lock_for(node_id): seq = await _bump_seq(node_id) - payload = {"seq": seq, "streams": await desired_for_node(node_id), "issued_at": _now_iso()} + # Stamp BEFORE building the list: a stream detached between the two + # (stop_stream does not hold this lock) must not get this seq, because + # the list built here could still contain it — an ack for it would then + # free a token the node was still told to use. await leases.stamp_release_seq(node_id, seq) + payload = {"seq": seq, "streams": await desired_for_node(node_id), "issued_at": _now_iso()} ok = mqtt_handler.publish_discord_desired(node_id, payload) await fstore.doc_set(NODE_STATE, node_id, {"published": bool(ok), "published_at": _now_iso()}) if not ok: @@ -457,6 +461,29 @@ async def reconcile_once() -> None: except StreamError as e: await _mark(s, "error", e.message) + # Leases with NO live stream left (``releasing``) on a node that is gone: + # /leave while the node was down, or the node was deleted. The pass above + # only walks desired=on streams, so without this the token would stay out + # of the pool until that node came back — forever for a dead node. + for tok in await fstore.collection_list(leases.TOKENS): + nid = (tok.get("lease") or {}).get("node_id") + if not nid or nid in released: + continue + node = nodes.get(nid) + if node and node.get("status") in ONLINE: + continue + if node is not None: + heard = _parse_ts(node.get("last_seen")) + else: + # Deleted node (its MQTT client is revoked, so the edge fail-safe + # logs out). Last heard = its last actual report, else lease start. + st = await fstore.doc_get(NODE_STATE, nid) or {} + heard = _parse_ts(st.get("actual_at")) or _parse_ts((tok.get("lease") or {}).get("since")) + if heard and now - heard < grace: + continue + await release_offline_node(nid) + released.add(nid) + # Desired-state publishes the broker never took (MQTT was down). for st in await fstore.collection_list(NODE_STATE, published=False): if st.get("node_id"): diff --git a/drb-c2-core/tests/test_discord_leases.py b/drb-c2-core/tests/test_discord_leases.py index 971157f..ec4092f 100644 --- a/drb-c2-core/tests/test_discord_leases.py +++ b/drb-c2-core/tests/test_discord_leases.py @@ -358,6 +358,60 @@ async def test_persistent_stream_moves_after_grace_adhoc_stops(fs, monkeypatch): assert all((_lease(fs, t) or {}).get("node_id") != "n1" for t in ("t1", "t2")) +async def test_leave_while_node_offline_frees_token_after_grace(fs, monkeypatch): + """A stream stopped while its node was down leaves no desired=on stream + behind; the token must still return to the pool once the grace passes.""" + monkeypatch.setattr(streams.settings, "discord_lease_offline_release_seconds", 600) + _system(fs); _node(fs, "n1"); _token(fs, "t1") + await _start() + _node(fs, "n1", status="offline", age=120) + await streams.stop_stream("g1:c1") + await streams.reconcile_once() + assert _lease(fs, "t1")["releasing"] is True # inside the grace: held + _node(fs, "n1", status="offline", age=900) + await streams.reconcile_once() + assert _lease(fs, "t1") is None + + +async def test_deleted_node_lease_freed_after_grace_from_last_report(fs, monkeypatch): + monkeypatch.setattr(streams.settings, "discord_lease_offline_release_seconds", 600) + _system(fs); _node(fs, "n1"); _token(fs, "t1") + await _start() + await streams.stop_stream("g1:c1") + fs.put("discord_nodes", "n1", {**(fs.get("discord_nodes", "n1") or {}), + "actual_at": datetime.now(timezone.utc).isoformat()}) + await fs.doc_delete("nodes", "n1") + await streams.reconcile_once() + assert _lease(fs, "t1") is not None # heard from recently: held + old = (datetime.now(timezone.utc) - timedelta(seconds=900)).isoformat() + fs.put("discord_nodes", "n1", {**fs.get("discord_nodes", "n1"), "actual_at": old}) + await streams.reconcile_once() + assert _lease(fs, "t1") is None + + +async def test_release_seq_never_names_a_list_that_still_holds_the_stream(fs, monkeypatch): + """A stop landing between building the desired list and stamping must not + be stamped with that list's seq (the list still contains the stream).""" + _system(fs); _node(fs, "n1"); _token(fs, "t1") + await _start() + real = streams.desired_for_node + fired = [] + + async def _racy(node_id): + out = await real(node_id) + if not fired: + fired.append(1) + await leases.detach_stream("g1:c1", {"desired": "off"}) + return out + monkeypatch.setattr(streams, "desired_for_node", _racy) + await streams.publish_desired("n1") + racy_seq = fs.published[-1][1]["seq"] + assert any(d["stream_id"] == "g1:c1" for d in fs.published[-1][1]["streams"]) + rseq = _lease(fs, "t1")["release_seq"] + assert rseq is None or rseq > racy_seq + assert await leases.settle_releases("n1", set(), racy_seq) == [] + + async def test_retuned_node_gives_up_a_persistent_stream(fs): _system(fs); _system(fs, "sys-2"); _node(fs, "n1"); _token(fs, "t1") await _start(mode="persistent")