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>
This commit is contained in:
co-authored by
Claude Opus 5.5
parent
a31a8df17c
commit
03ba7eacd9
@@ -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"):
|
||||
|
||||
Reference in New Issue
Block a user