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 <noreply@anthropic.com>

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"):

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")