""" Discord radio channels as desired state (server-26#203, #204; node-26#25). One ``discord_streams/{stream_id}`` doc per radio stream. A Discord voice channel may carry SEVERAL streams (server-26#213 follow-up, owner decision 2026-10-05): each has its own radio bot and plays its own system, so a channel can hold e.g. County P25 on one bot and City Fire on another. ``stream_id`` is ``{guild_id}:{channel_id}:{suffix}`` (``discord_leases.new_stream_id``); docs from before are ``{guild_id}:{channel_id}`` and are read, stopped and removed exactly the same way. Adding a stream never reuses another stream's doc except the same system in the same channel (idempotent re-add / restart). Fields: org_id, guild_id, channel_id, channel_name, guild_name, system_id, system_name, mode: "persistent" | "adhoc" persistent — stays connected and is re-placed by the reconciler (node reboot, node lost, C2 restart) until someone stops it. adhoc — stopped instead of re-placed when its node goes away. (Idle auto-dismiss is not built yet — see the PR.) desired: "on" | "off" node_id, token_id, lease_id — placement (written with the lease, atomically) actual: {state, error, code, since, at} — reported by the node, never guessed here state: pending | connecting | connected | error | stopping | node_offline | paused_locally code: the node's machine-readable reason on error (login_failed, connect_failed, not_in_guild, bad_channel, no_audio) or None tg_filter: None — phase-2 seam (per-channel talkgroup gate); always None now now_playing_message_id, created_by_discord_id, created_at, updated_at Flow: ``start_stream`` picks a node + leases a token -> ``publish_desired`` retains ``nodes/{node}/discord/desired`` (full list for that node, with a monotonically increasing ``seq``) -> the edge reconciles and publishes ``nodes/{node}/discord/actual`` -> ``handle_actual`` updates ``actual`` and settles token releases. ``reconcile_loop`` re-places persistent streams whose node went away and retries desired-state publishes that never reached the broker. The desired payload carries token **ids**, never token strings: the edge fetches the string over HTTPS (GET /discord/node/{node_id}/tokens/{id}) with its own node API key, and C2 hands it out only to the node holding the lease. A retained MQTT message therefore never stores a bot credential on the broker. """ from __future__ import annotations import asyncio from datetime import datetime, timezone, timedelta from typing import Optional from app.config import settings from app.internal import firestore as fstore from app.internal import discord_leases as leases from app.internal import discord_rest from app.internal.logger import logger STREAMS = leases.STREAMS GUILDS = "discord_guilds" # discord_guilds/{guild_id}: {org_id, linked_by, linked_at} NODE_STATE = "discord_nodes" # discord_nodes/{node_id}: {seq, published, applied_seq, actual_at} ONLINE = ("online", "recording") def _now_iso() -> str: return datetime.now(timezone.utc).isoformat() def _effective_system(node: dict) -> Optional[str]: if node.get("is_overridden") and node.get("override_system_id"): return node["override_system_id"] return node.get("assigned_system_id") def _parse_ts(value) -> Optional[datetime]: try: ts = datetime.fromisoformat(value) if isinstance(value, str) else value except (TypeError, ValueError): return None if isinstance(ts, datetime) and ts.tzinfo is None: ts = ts.replace(tzinfo=timezone.utc) return ts if isinstance(ts, datetime) else None # --------------------------------------------------------------------------- # Guild -> org # --------------------------------------------------------------------------- async def resolve_guild_org(guild_id: str) -> Optional[str]: """The org a Discord server acts for: its ``discord_guilds`` link, else ``settings.discord_unlinked_guild_org_id`` (None when that is empty). A server an org explicitly unlinked (``unlinked_at`` set, server-26#213) gets no fallback: otherwise unlinking org B's server would quietly hand it to the founding org.""" link = await fstore.doc_get(GUILDS, str(guild_id)) if link and link.get("org_id"): return link["org_id"] if link and link.get("unlinked_at"): return None return settings.discord_unlinked_guild_org_id or None async def _node_stream_counts(exclude_stream_id: Optional[str] = None) -> dict[str, int]: streams = await fstore.collection_list(STREAMS, desired="on") counts: dict[str, int] = {} for s in streams: if s.get("node_id") and s.get("stream_id") != exclude_stream_id: counts[s["node_id"]] = counts.get(s["node_id"], 0) + 1 return counts def _node_usable(n: dict, org_id: str, system_id: str) -> bool: return ( n.get("org_id") == org_id and n.get("status") in ONLINE and n.get("approval_status") in (None, "approved") and _effective_system(n) == system_id ) async def candidate_nodes(org_id: str, system_id: str, exclude_stream_id: Optional[str] = None) -> list[dict]: """Online, approved nodes of this org currently listening to ``system_id``, best first: nodes already carrying streams (one gateway session can serve several guilds) but under the per-node cap, then the rest.""" nodes = await fstore.collection_list("nodes", org_id=org_id) counts = await _node_stream_counts(exclude_stream_id) cap = settings.discord_node_max_streams out = [n for n in nodes if n.get("node_id") and _node_usable(n, org_id, system_id) and counts.get(n["node_id"], 0) < cap] out.sort(key=lambda n: (-counts.get(n["node_id"], 0), n.get("name") or n["node_id"])) return out # --------------------------------------------------------------------------- # Desired state -> node # --------------------------------------------------------------------------- # One c2-core process publishes desired state. Serialising bump->build->publish # per node is what keeps the RETAINED message the newest one: two unserialised # publishes could land seq 7 then seq 6, and a reconnecting edge would then be # handed the stale list. _publish_locks: dict[str, asyncio.Lock] = {} def _lock_for(node_id: str) -> asyncio.Lock: lock = _publish_locks.get(node_id) if lock is None: lock = _publish_locks[node_id] = asyncio.Lock() return lock async def _bump_seq(node_id: str) -> int: db = fstore.db def _txn(txn): ref = db.collection(NODE_STATE).document(node_id) snap = ref.get(transaction=txn) seq = int(((snap.to_dict() or {}) if snap.exists else {}).get("seq") or 0) + 1 txn.set(ref, {"node_id": node_id, "seq": seq}, merge=True) return seq return await fstore.run_transaction(_txn) async def desired_for_node(node_id: str) -> list[dict]: streams = await fstore.collection_list(STREAMS, node_id=node_id) out = [] for s in streams: if s.get("desired") != "on" or not s.get("token_id"): continue out.append({ "stream_id": s["stream_id"], "guild_id": s["guild_id"], "channel_id": s["channel_id"], "token_id": s["token_id"], "lease_id": s.get("lease_id"), "system_name": s.get("system_name"), "mode": s.get("mode", "adhoc"), # Phase-2 seam: the edge ignores this today (no TG gate yet). "tg_filter": s.get("tg_filter"), }) out.sort(key=lambda d: d["stream_id"]) return out async def publish_desired(node_id: str) -> bool: """Retain the node's full stream list under a fresh ``seq``. Releasing tokens are stamped with that seq BEFORE the publish, so the earliest ack that can free them is one for a list that already lacks them. Returns whether the broker took the message; a miss is recorded and the reconciler retries it (nothing is lost — the streams stay desired).""" from app.internal.mqtt_handler import mqtt_handler async with _lock_for(node_id): seq = await _bump_seq(node_id) # 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: logger.warning(f"Discord desired state for {node_id} (seq {seq}) not delivered — reconciler will retry") return bool(ok) # --------------------------------------------------------------------------- # Start / stop # --------------------------------------------------------------------------- class StreamError(Exception): def __init__(self, code: str, message: str, invite: Optional[list] = None, status: int = 409, extra: Optional[dict] = None): super().__init__(message) self.code = code self.message = message self.invite = invite or [] self.status = status self.extra = extra or {} ERROR_TEXT = { "no_org": "This Discord server is not linked to a DRB organization yet.", "no_node": "No online node in your organization is listening to that system right now.", "pool_empty": "No radio bots have been added to DRB yet. A platform admin adds them under Radio bots.", "no_bot_in_guild": "No free radio bot is a member of this server. Invite one with the link below, then try again.", "pool_exhausted": ("Every radio bot that can join this server is already busy — a bot can be in only one voice " "channel per server, and a bot carrying radio on another node can't be shared. Add another " "radio bot to DRB and invite it to this server, or stop a radio channel here."), "limit": "This server already has its maximum number of radio channels. Stop one with /leave first.", "preferred_unavailable": "That radio bot can't be used here (it was removed, failed validation, or belongs to " "another organization). Pick another bot, or let DRB choose any free one.", "not_found": "That radio channel no longer exists.", } # One c2-core process. Serialising start_stream per guild makes "is this # system already playing in this channel?" + "is the server at its cap?" + # create a real check-then-write: two concurrent adds of the same system to # one channel land on ONE stream, and two adds can't both take the last slot. _guild_locks: dict[str, asyncio.Lock] = {} def _guild_lock(guild_id: str) -> asyncio.Lock: lock = _guild_locks.get(guild_id) if lock is None: lock = _guild_locks[guild_id] = asyncio.Lock() return lock def _stream_label(s: dict) -> str: where = f"#{s['channel_name']}" if s.get("channel_name") else f"channel {s.get('channel_id')}" return f"{s.get('system_name') or s.get('system_id') or 'radio'} in {where}" async def find_channel_stream(guild_id: str, channel_id: str, system_id: str) -> Optional[dict]: """The stream already carrying ``system_id`` in this voice channel: a live one first, else the most recently updated stopped one (restarted instead of piling up a new doc per /join + /leave). None if there is neither.""" same = [s for s in await fstore.collection_list(STREAMS, guild_id=str(guild_id)) if str(s.get("channel_id")) == str(channel_id) and s.get("system_id") == system_id and s.get("stream_id")] live = sorted((s for s in same if s.get("desired") == "on"), key=lambda s: s.get("created_at") or "") if live: return live[0] same.sort(key=lambda s: s.get("updated_at") or "", reverse=True) return same[0] if same else None async def _lease_error(res: leases.LeaseResult) -> "StreamError": code = res.error or "no_node" d = res.detail or {} bot = d.get("token_name") or "That radio bot" if code == "preferred_busy": busy = (await fstore.doc_get(STREAMS, d["busy_stream_id"]) if d.get("busy_stream_id") else None) or {} what = _stream_label(busy) if busy else "another radio channel" return StreamError(code, f"{bot} is already playing {what} in this server. A bot can be in only one voice " "channel per server — pick another bot, or let DRB choose any free one.", extra={"busy_stream_id": d.get("busy_stream_id")}) if code == "preferred_other_node": node = (await fstore.doc_get("nodes", d["node_id"]) if d.get("node_id") else None) or {} return StreamError(code, f"{bot} is carrying radio on another node ({node.get('name') or d.get('node_id')}) " "that can't serve this channel. Pick another bot, or let DRB choose any free one.") return StreamError(code, ERROR_TEXT.get(code, code), invite=res.invite) async def start_stream( *, org_id: str, guild_id: Optional[str] = None, channel_id: Optional[str] = None, system_id: str, mode: str = "adhoc", channel_name: Optional[str] = None, guild_name: Optional[str] = None, created_by: Optional[str] = None, node_id: Optional[str] = None, preferred_token_id: Optional[str] = None, max_streams: Optional[int] = None, stream_id: Optional[str] = None, ) -> dict: """Put ``system_id`` on a voice channel. * ``stream_id`` given — restart / edit THAT stream (the reconciler, the admin page's Start button). Its guild and channel come from the doc. * otherwise — ADD a stream to ``guild_id:channel_id``. If this system is already on in that channel the existing stream is returned (idempotent re-add, ``already_on``); a stopped one with this system is restarted; anything else gets a brand-new stream id, so another system's stream in the same channel is never touched. ``preferred_token_id`` is strict: a bot already in this server, on another node, or unusable is refused with a reason, never silently swapped. """ if mode not in ("persistent", "adhoc"): raise StreamError("bad_mode", "mode must be persistent or adhoc", status=400) if not org_id: raise StreamError("no_org", ERROR_TEXT["no_org"], status=403) existing: dict = {} if stream_id: existing = await fstore.doc_get(STREAMS, stream_id) or {} if not existing: raise StreamError("not_found", ERROR_TEXT["not_found"], status=404) doc_guild, doc_channel = str(existing.get("guild_id")), str(existing.get("channel_id")) if (guild_id and str(guild_id) != doc_guild) or (channel_id and str(channel_id) != doc_channel): raise StreamError("bad_stream", "That radio channel belongs to a different server or voice channel.", status=400) guild_id, channel_id = doc_guild, doc_channel if not guild_id or not channel_id: raise StreamError("bad_stream", "guild_id and channel_id are required.", status=400) guild_id, channel_id = str(guild_id), str(channel_id) async with _guild_lock(guild_id): return await _start_locked( org_id=org_id, guild_id=guild_id, channel_id=channel_id, system_id=system_id, mode=mode, channel_name=channel_name, guild_name=guild_name, created_by=created_by, node_id=node_id, preferred_token_id=preferred_token_id, max_streams=max_streams, stream_id=stream_id, existing=existing, ) async def _start_locked(*, org_id: str, guild_id: str, channel_id: str, system_id: str, mode: str, channel_name: Optional[str], guild_name: Optional[str], created_by: Optional[str], node_id: Optional[str], preferred_token_id: Optional[str], max_streams: Optional[int], stream_id: Optional[str], existing: dict) -> dict: already_on = False if stream_id: existing = await fstore.doc_get(STREAMS, stream_id) or existing # re-read under the lock else: existing = await find_channel_stream(guild_id, channel_id, system_id) or {} if existing.get("desired") == "on": already_on = True if preferred_token_id and existing.get("token_id") and existing["token_id"] != preferred_token_id: tok = await fstore.doc_get(leases.TOKENS, existing["token_id"]) or {} raise StreamError("duplicate", f"{_stream_label(existing)} is already playing on radio bot " f"{tok.get('name') or existing['token_id']}. Remove that one " "first, or pick another system.", extra={"busy_stream_id": existing["stream_id"]}) # A re-add is not an edit: keep the running stream's label and mode. channel_name, mode = None, existing.get("mode") or mode stream_id = existing.get("stream_id") or leases.new_stream_id(guild_id, channel_id) if max_streams is not None and existing.get("desired") != "on": live = [s for s in await fstore.collection_list(STREAMS, guild_id=guild_id) if s.get("desired") == "on"] if len(live) >= max_streams: raise StreamError("limit", ERROR_TEXT["limit"]) system = await fstore.doc_get_cached("systems", system_id) or {} if node_id: node = await fstore.doc_get("nodes", node_id) # A pinned node still has to be this org's and approved; it may be # listening to anything (the web "Join Discord" button pins a node). if not node or node.get("org_id") != org_id or node.get("approval_status") not in (None, "approved"): raise StreamError("no_node", ERROR_TEXT["no_node"]) candidates = [node] else: candidates = await candidate_nodes(org_id, system_id, exclude_stream_id=stream_id) # Keep the current node when it is still a valid choice (reuse the lease). cur = existing.get("node_id") if cur and existing.get("system_id") == system_id: candidates.sort(key=lambda n: 0 if n["node_id"] == cur else 1) if preferred_token_id: # A chosen bot can only run on the node it is logged in on. tok = await fstore.doc_get(leases.TOKENS, preferred_token_id) or {} held_on = (tok.get("lease") or {}).get("node_id") if held_on: candidates.sort(key=lambda n: 0 if n["node_id"] == held_on else 1) if not candidates: raise StreamError("no_node", ERROR_TEXT["no_node"]) now = _now_iso() fields = { "stream_id": stream_id, "org_id": org_id, "guild_id": guild_id, "channel_id": channel_id, "system_id": system_id, "system_name": system.get("name") or system_id, "mode": mode, "desired": "on", "tg_filter": None, "stop_reason": None, "updated_at": now, } prev_actual = existing.get("actual") or {} if existing.get("desired") != "on" or prev_actual.get("state") not in ("connected", "connecting"): fields["actual"] = {"state": "pending", "error": None, "since": now, "at": now} if channel_name: fields["channel_name"] = channel_name if guild_name: fields["guild_name"] = guild_name if not existing: fields["created_at"] = now fields["created_by_discord_id"] = created_by fields["now_playing_message_id"] = None last: Optional[leases.LeaseResult] = None prev_node = existing.get("node_id") refreshed = False for node in candidates: res = await leases.lease_for_stream(stream_id, node_id=node["node_id"], org_id=org_id, preferred_token_id=preferred_token_id, stream_fields=fields) if not res.ok and res.error == "no_bot_in_guild" and not refreshed: # Membership is cached on the token doc; someone may have just # invited a bot. Re-read it from Discord once, then retry. refreshed = True tokens = await fstore.collection_list(leases.TOKENS) stale = [t["token_id"] for t in tokens if t.get("token_id") and t.get("guild_ids") is not None and guild_id not in t["guild_ids"]] if await discord_rest.refresh_membership(stale): res = await leases.lease_for_stream(stream_id, node_id=node["node_id"], org_id=org_id, preferred_token_id=preferred_token_id, stream_fields=fields) if res.ok: delivered = await publish_desired(node["node_id"]) if prev_node and prev_node != node["node_id"]: await publish_desired(prev_node) logger.info(f"Discord stream {stream_id} -> node {node['node_id']} token {res.token_id}" f" ({'reused' if res.reused else 'new'} lease{', already on' if already_on else ''})") return {**existing, **fields, "node_id": node["node_id"], "node_name": node.get("name"), "token_id": res.token_id, "lease_id": res.lease_id, "reused": res.reused, "delivered": delivered, "already_on": already_on} if res.error in ("preferred_busy", "preferred_unavailable"): raise await _lease_error(res) # true on every node — no point trying the rest last = res if last is None: raise StreamError("no_node", ERROR_TEXT["no_node"]) raise await _lease_error(last) async def stop_stream(stream_id: str, reason: str = "stopped") -> bool: existing = await fstore.doc_get(STREAMS, stream_id) if not existing: return False now = _now_iso() node_id = await leases.detach_stream(stream_id, { "desired": "off", "updated_at": now, "stop_reason": reason, "actual": {**(existing.get("actual") or {}), "state": "stopping", "at": now}, }) if node_id: await publish_desired(node_id) logger.info(f"Discord stream {stream_id} stopped ({reason}) node={node_id or existing.get('node_id')}" f" guild={existing.get('guild_id')} channel={existing.get('channel_id')}") return True async def delete_stream(stream_id: str, reason: str = "removed") -> Optional[dict]: """Permanently remove a radio channel (server-26#213 follow-up). Order matters (server-26#216 lease semantics): the stream is first stopped through the normal desired-state path — detached from its token (token -> ``releasing``) and the node's list republished without it — and only then is the doc deleted. The token returns to the pool exactly as for /leave, when the node acks a list without it and no longer has a session on it (``settle_releases`` reads the token lease, never the stream doc). The delete itself is a transaction that refuses if the stream was started again in between (desired on, or a token re-leased onto it): deleting it then would strand that token's lease on a stream that no longer exists. Returns the doc as it was before removal, or None if it did not exist. """ existing = await fstore.doc_get(STREAMS, stream_id) if not existing: return None if existing.get("desired") == "on" or existing.get("token_id"): await stop_stream(stream_id, reason) if not await leases.delete_stopped_stream(stream_id): raise StreamError("restarted", "That radio channel was started again while it was being removed. " "Stop it and try again.") logger.info(f"Discord stream {stream_id} removed ({reason}) node={existing.get('node_id')}" f" guild={existing.get('guild_id')} channel={existing.get('channel_id')}" f" org={existing.get('org_id')}") return existing async def remove_guild(guild_id: str, by: str) -> dict: """Forget a Discord server: remove every radio channel in it (stop first, then delete — ``delete_stream``) and tombstone its ``discord_guilds`` doc. The doc is NOT deleted: with no doc, ``resolve_guild_org`` would hand the server to ``DISCORD_UNLINKED_GUILD_ORG_ID`` the next time anyone typed a command there. It is unlinked the way PR #218's unlink does it (``org_id`` None + ``unlinked_at``, so no fallback) and stamped ``removed_at`` so the admin page stops listing it. Linking it again (``/drb link`` or the admin link) clears ``removed_at``.""" guild_id = str(guild_id) link = await fstore.doc_get(GUILDS, guild_id) or {} removed = [] for s in await fstore.collection_list(STREAMS, guild_id=guild_id): try: if s.get("stream_id") and await delete_stream(s["stream_id"], "server_removed"): removed.append(s["stream_id"]) except StreamError as e: # A channel restarted mid-removal: stop here, leave the server # linked, and tell the caller which channels are already gone so # the partial removal is still audited. e.removed = removed raise now = _now_iso() await fstore.doc_set(GUILDS, guild_id, { "guild_id": guild_id, "org_id": None, "unlinked_at": now, "unlinked_by": by, "removed_at": now, "removed_by": by, }, merge=True) logger.info(f"Discord guild {guild_id} removed by {by}: {len(removed)} radio channel(s) deleted," f" was org={link.get('org_id')!r}") return {"guild_id": guild_id, "previous_org_id": link.get("org_id"), "streams_removed": removed, "guild_name": link.get("guild_name")} async def stop_guild(guild_id: str, reason: str, channel_id: Optional[str] = None, stream_id: Optional[str] = None) -> list[str]: """Stop live streams in ONE Discord server: all of them, every stream in one voice channel, or one stream by id. The /leave fix for server-26#203: never touches another server's streams (a ``stream_id`` from another server simply matches nothing).""" stopped = [] for s in await fstore.collection_list(STREAMS, guild_id=str(guild_id)): if s.get("desired") != "on": continue if channel_id and str(s.get("channel_id")) != str(channel_id): continue if stream_id and s.get("stream_id") != stream_id: continue await stop_stream(s["stream_id"], reason) stopped.append(s["stream_id"]) return stopped async def stop_node(node_id: str, reason: str) -> list[str]: stopped = [] for s in await fstore.collection_list(STREAMS, node_id=node_id): if s.get("desired") == "on": await stop_stream(s["stream_id"], reason) stopped.append(s["stream_id"]) return stopped # --------------------------------------------------------------------------- # Actual state <- node # --------------------------------------------------------------------------- _VALID_STATES = {"connecting", "connected", "error", "stopped", "paused_locally"} def _log_transition(node_id: str, doc: dict, prev_state: Optional[str], state: str, error: Optional[str], code: Optional[str]) -> None: """One log line per node-reported state CHANGE (never per heartbeat — the edge re-reports every ~60s). Errors are warnings so they surface in the c2-core log without turning debug on; a radio bot that cannot log in gets its own line naming the token, since that is the one failure that takes every channel on the token down at once.""" where = (f"stream={doc.get('stream_id')} node={node_id} guild={doc.get('guild_id')}" f" channel={doc.get('channel_id')} token={doc.get('token_id')}") if state == "error": if code == "login_failed": logger.warning(f"Radio bot {doc.get('token_id')} failed to log in on node {node_id}: {error} ({where})") else: logger.warning(f"Discord stream {prev_state or 'pending'} -> error [{code or 'unknown'}]:" f" {error} ({where})") else: logger.info(f"Discord stream {prev_state or 'pending'} -> {state} ({where})" + (f": {error}" if error else "")) async def handle_actual(node_id: str, payload: dict) -> None: """``nodes/{node_id}/discord/actual``. A node may only report on streams currently placed on it — anything else is ignored, so a compromised node cannot rewrite another node's (or org's) stream state.""" if not isinstance(payload, dict): return reported = payload.get("streams") or {} now = _now_iso() recheck: set = set() if isinstance(reported, dict): for stream_id, st in list(reported.items())[:50]: if not isinstance(st, dict) or "/" in str(stream_id): continue # "/" would be a Firestore sub-path, never a stream id doc = await fstore.doc_get(STREAMS, str(stream_id)) if not doc or doc.get("node_id") != node_id or doc.get("desired") != "on": continue state = st.get("state") if st.get("state") in _VALID_STATES else "error" error = st.get("error") error = str(error)[:300] if error else None code = st.get("code") code = str(code)[:40] if code else None if code == "not_in_guild" and doc.get("token_id"): recheck.add(doc["token_id"]) prev = doc.get("actual") or {} if prev.get("state") == state and prev.get("error") == error and prev.get("code") == code: since = prev.get("since") or now else: since = now _log_transition(node_id, doc, prev.get("state"), state, error, code) await fstore.doc_update(STREAMS, str(stream_id), { "actual": {"state": state, "error": error, "code": code, "since": since, "at": now}, }) if recheck: await discord_rest.refresh_membership(sorted(recheck)) if "sessions" in payload: sessions = {str(t) for t in (payload.get("sessions") or [])} applied = payload.get("applied_seq") applied = int(applied) if isinstance(applied, (int, float)) and not isinstance(applied, bool) else None freed = await leases.settle_releases(node_id, sessions, applied) for tok in freed: logger.info(f"Radio bot token {tok} released — node {node_id} confirmed it logged out") await fstore.doc_set(NODE_STATE, node_id, {"applied_seq": applied, "actual_at": now}) # --------------------------------------------------------------------------- # Self-healing: re-place persistent streams whose node is gone # --------------------------------------------------------------------------- async def release_offline_node(node_id: str) -> None: """Node offline past the lease grace: free its tokens (the edge already logged them out — see release_node_leases) and unplace its streams.""" orphaned = await leases.release_node_leases(node_id) if not orphaned: return now = _now_iso() for sid in orphaned: await fstore.doc_set(STREAMS, sid, { "token_id": None, "lease_id": None, "actual": {"state": "node_offline", "error": "The node carrying this channel went offline.", "since": now, "at": now}, }) await publish_desired(node_id) # retained: the node sees an accurate (empty) list when it returns logger.warning(f"Node {node_id} offline past lease grace — released {len(orphaned)} radio stream(s)") async def _unplace(s: dict, why: str) -> None: """Take a stream off its node (token detached, node republished) but leave it desired, so the reconciler can place it somewhere valid.""" now = _now_iso() old = await leases.detach_stream(s["stream_id"], { "node_id": None, "actual": {"state": "error", "error": why, "since": now, "at": now}, }) if old: await publish_desired(old) async def reconcile_once() -> None: nodes = {n["node_id"]: n for n in await fstore.collection_list("nodes") if n.get("node_id")} grace = timedelta(seconds=settings.discord_lease_offline_release_seconds) now = datetime.now(timezone.utc) released: set = set() for s in await fstore.collection_list(STREAMS, desired="on"): node = nodes.get(s.get("node_id") or "") if node and s.get("token_id"): if node.get("status") in ONLINE: if _effective_system(node) == s.get("system_id"): continue # healthy placement # Node retuned to another system under a live stream. if s.get("mode") != "persistent": await stop_stream(s["stream_id"], "node_retuned") continue await _unplace(s, "The node carrying this channel switched systems; finding another.") else: ls = _parse_ts(node.get("last_seen")) if ls and now - ls < grace: await _mark(s, "node_offline", "The node carrying this channel is offline; waiting for it.") continue # hold the lease through the grace window if node["node_id"] not in released: await release_offline_node(node["node_id"]) released.add(node["node_id"]) if s.get("mode") != "persistent": await stop_stream(s["stream_id"], "node_lost") continue org_id = s.get("org_id") try: await start_stream(org_id=org_id, stream_id=s["stream_id"], system_id=s["system_id"], mode="persistent") 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"): await publish_desired(st["node_id"]) async def _mark(s: dict, state: str, error: str) -> None: prev = s.get("actual") or {} if prev.get("state") == state and prev.get("error") == error: return now = _now_iso() await fstore.doc_set(STREAMS, s["stream_id"], {"actual": {"state": state, "error": error, "since": now, "at": now}}) async def reconcile_loop() -> None: logger.info("Discord stream reconciler started.") while True: await asyncio.sleep(settings.discord_reconcile_interval_seconds) try: await reconcile_once() except Exception as e: logger.error(f"Discord reconcile error: {e}")