From a31a8df17c9d481e36c00f92a57e3049e0f31d8d Mon Sep 17 00:00:00 2001 From: Logan Cusano Date: Sun, 4 Oct 2026 18:05:26 -0400 Subject: [PATCH] Discord radio channels: wire leases + desired-state streams (server-26#203, #204) c2-core - discord_streams/discord_leases wired in: POST /discord/streams, POST /discord/guilds/{id}/stop, GET /discord/guilds/{id}/overview, GET /discord/streams/{id}, PUT /discord/guilds/{id}/link (admin), GET /discord/node/{node}/tokens/{id} (node key, lease holder only). - Retained nodes/{id}/discord/desired (token ids, never token strings), nodes/+/discord/actual subscribed and handled; per-node publish lock keeps the retained list the newest; failed publishes retried by the reconciler. - Tokens freed only after the node acks logout (or offline past discord_lease_offline_release_seconds) - sweeper and C2 startup no longer release leases. - /nodes/{id}/command discord_join/leave map onto streams; CommandPayload keeps preferred_token_id. /tokens: admin-only list, validate on add against Discord, refresh, flush via transactional force release. - Guild -> org via discord_guilds, unlinked guilds fall back to DISCORD_UNLINKED_GUILD_ORG_ID (default "founding"). - tests/fake_firestore.py: optimistic-transaction fake; 27 lease/race tests. discord bot - /join, /leave, /status, autocomplete scoped to the calling guild; /leave never touches another server; Manage Channels default permission; /join edits its reply with the node's real result; /help copy fixed. Co-Authored-By: Claude Opus 5.5 --- drb-c2-core/app/config.py | 18 + drb-c2-core/app/internal/discord_leases.py | 82 +++- drb-c2-core/app/internal/discord_streams.py | 252 ++++++++--- drb-c2-core/app/internal/mqtt_handler.py | 33 +- drb-c2-core/app/internal/node_sweeper.py | 7 +- drb-c2-core/app/main.py | 33 +- drb-c2-core/app/models.py | 3 + drb-c2-core/app/routers/discord.py | 184 ++++++++ drb-c2-core/app/routers/nodes.py | 41 +- drb-c2-core/app/routers/tokens.py | 140 +++--- drb-c2-core/tests/fake_firestore.py | 172 +++++++ drb-c2-core/tests/test_discord_leases.py | 423 ++++++++++++++++++ drb-c2-core/tests/test_node_sweeper.py | 28 +- drb-server-discord-bot/app/commands/radio.py | 299 ++++++++----- .../app/internal/c2_client.py | 72 +-- 15 files changed, 1428 insertions(+), 359 deletions(-) create mode 100644 drb-c2-core/app/routers/discord.py create mode 100644 drb-c2-core/tests/fake_firestore.py create mode 100644 drb-c2-core/tests/test_discord_leases.py diff --git a/drb-c2-core/app/config.py b/drb-c2-core/app/config.py index cb5695f..a4a546a 100644 --- a/drb-c2-core/app/config.py +++ b/drb-c2-core/app/config.py @@ -210,6 +210,24 @@ class Settings(BaseSettings): # deployment will set this up, and skipping it must be silent. ai_alert_webhook_url: str = "" + # --- Discord radio channels (server-26#203/#204, node-26#25) ----------- + # Voice connections one node may carry at once (≈64 kbps upstream each — + # this is what protects an LTE node). + discord_node_max_streams: int = 4 + # Radio channels one Discord server may have on at once. + discord_guild_max_streams: int = 4 + # How often the reconciler re-places persistent streams / retries publishes. + discord_reconcile_interval_seconds: int = 30 + # A node offline this long has its radio-bot tokens returned to the pool. + # MUST stay above the edge's DISCORD_MQTT_LOSS_STOP_SECONDS (default 300): + # by then the edge has logged every radio bot out by itself, so the same + # token can never end up logged in on two nodes. + discord_lease_offline_release_seconds: int = 600 + # Org a Discord server that was never linked to an org (discord_guilds/{id}) + # acts for. Default keeps today's single-org deployment working unchanged; + # set it to "" to require every server to be linked first. + discord_unlinked_guild_org_id: str = "founding" + class Config: env_file = ".env" diff --git a/drb-c2-core/app/internal/discord_leases.py b/drb-c2-core/app/internal/discord_leases.py index 049e193..1c4c31b 100644 --- a/drb-c2-core/app/internal/discord_leases.py +++ b/drb-c2-core/app/internal/discord_leases.py @@ -318,7 +318,11 @@ async def release_node_leases(node_id: str) -> list[str]: async def force_release_token(token_id: str) -> Optional[dict]: - """Admin override (/tokens/flush, Force release). Returns the old lease.""" + """Admin override (/tokens/flush, delete). Frees the token and detaches + every stream that named it, in one transaction, so no stream doc is left + pointing at a token whose lease no longer lists it. Returns the old lease + (``None`` if the token was not leased or does not exist); the caller + republishes desired state for ``lease["node_id"]`` so the edge logs out.""" db = fstore.db def _txn(txn): @@ -327,7 +331,83 @@ async def force_release_token(token_id: str) -> Optional[dict]: if not snap.exists: return None lease = snap.to_dict().get("lease") + stream_refs = [] + for sid in (lease or {}).get("streams") or []: + sref = db.collection(STREAMS).document(sid) + ssnap = sref.get(transaction=txn) + if ssnap.exists and (ssnap.to_dict() or {}).get("token_id") == token_id: + stream_refs.append(sref) txn.update(ref, _legacy_mirror(None)) + for sref in stream_refs: + txn.set(sref, {"token_id": None, "lease_id": None}, merge=True) return lease return await fstore.run_transaction(_txn) + + +# --------------------------------------------------------------------------- +# Token validation / guild membership (server-26#204 item 5) +# --------------------------------------------------------------------------- + +DISCORD_API = "https://discord.com/api/v10" + + +class TokenRejected(Exception): + """Discord answered 401: the string is not a working bot token.""" + + +async def inspect_token(token: str) -> dict: + """Ask Discord who this token is and which servers it is in. + + Returns ``{bot_user_id, bot_name, application_id, guild_ids, valid, + validated_at, check_error}``. Raises TokenRejected on a 401. Any other + failure (Discord down, rate limit) is reported in ``check_error`` with + ``valid``/``guild_ids`` = None — unknown, not bad: a token is never + rejected because Discord had a bad minute. + """ + import httpx + headers = {"Authorization": f"Bot {token}"} + out: dict = {"valid": None, "guild_ids": None, "check_error": None, + "validated_at": _now().isoformat()} + try: + async with httpx.AsyncClient(timeout=10, base_url=DISCORD_API, headers=headers) as client: + me = await client.get("/users/@me") + if me.status_code == 401: + raise TokenRejected("Discord rejected this bot token (401).") + me.raise_for_status() + user = me.json() + out["bot_user_id"] = str(user.get("id")) + out["bot_name"] = user.get("username") + app = await client.get("/oauth2/applications/@me") + if app.status_code == 200: + out["application_id"] = str(app.json().get("id")) + guilds = await client.get("/users/@me/guilds", params={"limit": 200}) + guilds.raise_for_status() + out["guild_ids"] = [str(g["id"]) for g in guilds.json()] + out["valid"] = True + except TokenRejected: + raise + except Exception as e: # network / 5xx / 429 — unknown, not invalid + out["check_error"] = str(e)[:200] + return out + + +async def refresh_membership(token_ids: list[str]) -> int: + """Re-read guild membership for these tokens. Returns how many changed.""" + changed = 0 + for tok_id in token_ids: + tok = await fstore.doc_get(TOKENS, tok_id) + if not tok or not tok.get("token"): + continue + try: + info = await inspect_token(tok["token"]) + except TokenRejected as e: + await fstore.doc_update(TOKENS, tok_id, {"valid": False, "check_error": str(e), + "validated_at": _now().isoformat()}) + changed += 1 + continue + updates = {k: v for k, v in info.items() if v is not None or k == "check_error"} + if updates.get("guild_ids") != tok.get("guild_ids") or updates.get("valid") != tok.get("valid"): + changed += 1 + await fstore.doc_update(TOKENS, tok_id, updates) + return changed diff --git a/drb-c2-core/app/internal/discord_streams.py b/drb-c2-core/app/internal/discord_streams.py index 7d53485..5353e7f 100644 --- a/drb-c2-core/app/internal/discord_streams.py +++ b/drb-c2-core/app/internal/discord_streams.py @@ -1,5 +1,5 @@ """ -Discord radio channels as desired state (server-26#203 MVP). +Discord radio channels as desired state (server-26#203, #204; node-26#25). One ``discord_streams/{guild_id}:{channel_id}`` doc per Discord voice channel that should carry radio. Fields: @@ -7,14 +7,15 @@ that should carry radio. Fields: org_id, guild_id, channel_id, channel_name, guild_name, system_id, system_name, mode: "persistent" | "adhoc" - persistent — stays connected and reconnects by itself (node reboot, - Discord voice drop, C2 restart) until someone stops it. - adhoc — the Discord bot sets desired=off after ``idle_minutes`` - with no humans in the voice channel. + 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, since, at} — reported by the node, never guessed here - state: pending | connecting | connected | error | stopped | node_offline + state: pending | connecting | connected | error | stopping | + node_offline | paused_locally 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 @@ -23,12 +24,13 @@ 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. +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/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. +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 @@ -42,7 +44,8 @@ from app.internal import discord_leases as leases from app.internal.logger import logger STREAMS = leases.STREAMS -NODE_STATE = "discord_nodes" # discord_nodes/{node_id}: {seq, applied_seq, actual_at} +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") @@ -56,33 +59,56 @@ def _effective_system(node: dict) -> Optional[str]: return node.get("assigned_system_id") -async def _node_stream_counts(org_id: Optional[str]) -> dict[str, int]: +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).""" + link = await fstore.doc_get(GUILDS, str(guild_id)) + if link and link.get("org_id"): + return link["org_id"] + 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"): + 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 -async def candidate_nodes(org_id: str, system_id: str, exclude: Optional[str] = None) -> list[dict]: +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(org_id) + counts = await _node_stream_counts(exclude_stream_id) cap = settings.discord_node_max_streams - out = [] - for n in nodes: - if n.get("node_id") == exclude: - continue - if n.get("status") not in ONLINE or n.get("approval_status") not in (None, "approved"): - continue - if _effective_system(n) != system_id: - continue - if counts.get(n["node_id"], 0) >= cap: - continue - out.append(n) + 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 @@ -91,13 +117,27 @@ async def candidate_nodes(org_id: str, system_id: str, exclude: Optional[str] = # 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 {}).get("seq") or 0) + 1 if snap.exists else 1 + 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 @@ -125,14 +165,23 @@ async def desired_for_node(node_id: str) -> list[dict]: return out -async def publish_desired(node_id: str) -> int: - """Retain the node's full stream list; stamp releasing tokens with the seq.""" +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 - seq = await _bump_seq(node_id) - payload = {"seq": seq, "streams": await desired_for_node(node_id), "issued_at": _now_iso()} - mqtt_handler.publish_discord_desired(node_id, payload) - await leases.stamp_release_seq(node_id, seq) - return seq + 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()} + await leases.stamp_release_seq(node_id, seq) + 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) # --------------------------------------------------------------------------- @@ -149,11 +198,12 @@ class StreamError(Exception): 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. Stop another radio channel or add a bot.", - "limit": "This server already has its maximum number of radio channels. Stop one with /radio stop first.", + "limit": "This server already has its maximum number of radio channels. Stop one with /leave first.", } @@ -175,6 +225,9 @@ async def start_stream( it for a channel that is already on reuses the node and the lease.""" 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) + guild_id, channel_id = str(guild_id), str(channel_id) stream_id = leases.stream_id_for(guild_id, channel_id) existing = await fstore.doc_get(STREAMS, stream_id) or {} @@ -186,9 +239,14 @@ async def start_stream( system = await fstore.doc_get_cached("systems", system_id) or {} if node_id: - candidates = [await fstore.doc_get("nodes", node_id) or {"node_id": 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) + 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: @@ -200,10 +258,12 @@ async def start_stream( 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, - "actual": {"state": "pending", "error": None, "since": now, "at": now}, + "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: @@ -215,19 +275,31 @@ async def start_stream( 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 leases.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: - await publish_desired(node["node_id"]) + 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)") 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} + "token_id": res.token_id, "lease_id": res.lease_id, "reused": res.reused, + "delivered": delivered} last = res - code = last.error if last else "no_node" + code = last.error if last and last.error else "no_node" raise StreamError(code, ERROR_TEXT.get(code, code), invite=last.invite if last else []) @@ -246,22 +318,27 @@ async def stop_stream(stream_id: str, reason: str = "stopped") -> bool: return True -async def stop_guild(guild_id: str, reason: str) -> int: - n = 0 - for s in await fstore.collection_list(STREAMS, guild_id=guild_id): - if s.get("desired") == "on": - await stop_stream(s["stream_id"], reason) - n += 1 - return n +async def stop_guild(guild_id: str, reason: str, channel_id: Optional[str] = None) -> list[str]: + """Stop every live stream in ONE Discord server (optionally one channel). + The /leave fix for server-26#203: never touches another server's streams.""" + stopped = [] + for s in await fstore.collection_list(STREAMS, guild_id=str(guild_id)): + if s.get("desired") != "on": + continue + if channel_id and s.get("channel_id") != str(channel_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) -> int: - n = 0 +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) - n += 1 - return n + stopped.append(s["stream_id"]) + return stopped # --------------------------------------------------------------------------- @@ -275,8 +352,11 @@ 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): @@ -287,16 +367,23 @@ async def handle_actual(node_id: str, payload: dict) -> None: 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 + if st.get("code") == "not_in_guild" and doc.get("token_id"): + recheck.add(doc["token_id"]) prev = doc.get("actual") or {} - since = prev.get("since") if prev.get("state") == state else now - await fstore.doc_update(STREAMS, stream_id, { + if prev.get("state") == state and prev.get("error") == error: + since = prev.get("since") or now + else: + since = now + await fstore.doc_update(STREAMS, str(stream_id), { "actual": {"state": state, "error": error, "since": since, "at": now}, }) + if recheck: + await leases.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)) else None + 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") @@ -309,7 +396,7 @@ async def handle_actual(node_id: str, payload: dict) -> None: 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 re-place its streams.""" + logged them out — see release_node_leases) and unplace its streams.""" orphaned = await leases.release_node_leases(node_id) if not orphaned: return @@ -324,38 +411,57 @@ async def release_offline_node(node_id: str) -> None: 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 "") - online = bool(node) and node.get("status") in ONLINE and _effective_system(node) == s.get("system_id") - if online and s.get("token_id"): - continue if node and s.get("token_id"): - # Node offline (or retuned): hold the lease until the grace passes. - last_seen = node.get("last_seen") - try: - ls = datetime.fromisoformat(last_seen) if isinstance(last_seen, str) else last_seen - if ls and ls.tzinfo is None: - ls = ls.replace(tzinfo=timezone.utc) - except Exception: - ls = None - if node.get("status") in ONLINE or (ls and now - ls < grace): - if node.get("status") not in ONLINE: + 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 - await release_offline_node(node["node_id"]) + 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=s["org_id"], guild_id=s["guild_id"], channel_id=s["channel_id"], + await start_stream(org_id=org_id, guild_id=s["guild_id"], channel_id=s["channel_id"], system_id=s["system_id"], mode="persistent") except StreamError as e: await _mark(s, "error", e.message) + # 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 {} diff --git a/drb-c2-core/app/internal/mqtt_handler.py b/drb-c2-core/app/internal/mqtt_handler.py index a2782df..0dd672a 100644 --- a/drb-c2-core/app/internal/mqtt_handler.py +++ b/drb-c2-core/app/internal/mqtt_handler.py @@ -35,6 +35,8 @@ class MQTTHandler: client.subscribe("nodes/+/checkin", qos=1) client.subscribe("nodes/+/status", qos=1) client.subscribe("nodes/+/metadata", qos=1) + # Edge -> C2 actual state of its Discord radio streams (node-26#25). + client.subscribe("nodes/+/discord/actual", qos=1) # TODO(mqtt-cutover): drop this subscribe once the enrollment/HTTP # credentials flow (routers/enrollment.py) is stable in prod and # node-26 (the one live node) has been migrated. See @@ -61,6 +63,14 @@ class MQTTHandler: async def _dispatch(self, topic: str, payload: dict): parts = topic.split("/") + # nodes/{node_id}/discord/actual — the one four-level node topic. + if len(parts) == 4 and parts[0] == "nodes" and parts[2:] == ["discord", "actual"]: + from app.internal import discord_streams + try: + await discord_streams.handle_actual(parts[1], payload) + except Exception as e: + logger.error(f"MQTT dispatch error [discord/actual] from {parts[1]}: {e}") + return # Expected: nodes/{node_id}/{type} if len(parts) != 3 or parts[0] != "nodes": return @@ -179,11 +189,11 @@ class MQTTHandler: await fstore.doc_update("nodes", node_id, updates) - # NOTE: discord_connected in checkins is informational only — do NOT release the - # token here. The bot watchdog reconnects on transient Discord drops, so a single - # checkin with discord_connected=False during a brief reconnect window would - # incorrectly free the token while the bot is still active. Token release is - # handled exclusively by the discord_leave command and the node offline sweeper. + # NOTE: discord_connected in checkins is informational only — never release a + # token from it. Radio-bot tokens are freed only when the node acks (over + # nodes/{id}/discord/actual) that it logged the bot out, or by the + # reconciler once the node has been offline past the lease grace — see + # app/internal/discord_leases.py. # ------------------------------------------------------------------ # Status update @@ -335,6 +345,19 @@ class MQTTHandler: else: logger.warning(f"MQTT not connected — could not push config to {node_id}") + def publish_discord_desired(self, node_id: str, payload: dict) -> bool: + """Retained full list of the Discord streams this node should carry + (app/internal/discord_streams.py). Carries token IDs, never token + strings — the broker never stores a bot credential.""" + topic = f"nodes/{node_id}/discord/desired" + if self._client and self._connected: + info = self._client.publish(topic, json.dumps(payload), qos=1, retain=True) + if getattr(info, "rc", 0) == mqtt.MQTT_ERR_SUCCESS: + logger.info(f"Discord desired state seq {payload.get('seq')} -> {node_id}") + return True + logger.warning(f"MQTT not connected — Discord desired state for {node_id} not published") + return False + def publish_node_key(self, node_id: str, api_key: str): """Publish the provisioned API key to the node (retained so it survives reconnects). TODO(mqtt-cutover): dead once nodes.py's callers switch to the HTTP diff --git a/drb-c2-core/app/internal/node_sweeper.py b/drb-c2-core/app/internal/node_sweeper.py index 330ece7..a5b9086 100644 --- a/drb-c2-core/app/internal/node_sweeper.py +++ b/drb-c2-core/app/internal/node_sweeper.py @@ -53,8 +53,11 @@ async def _sweep(): node_id = node.get("node_id") await fstore.doc_update("nodes", node_id, {"status": "offline"}) logger.info(f"Node {node_id} marked offline (last seen: {last_seen.isoformat()})") - from app.routers.tokens import release_token - await release_token(node_id) + # Radio-bot tokens are NOT freed here any more (node-26#25): the + # node may still be logged in to Discord on them. The Discord + # reconciler (internal/discord_streams.py) frees them once the node + # has been offline past discord_lease_offline_release_seconds — by + # which point the edge has logged its bots out by itself. continue # Check for expired system overrides (only for fixed nodes with timeout enforced) diff --git a/drb-c2-core/app/main.py b/drb-c2-core/app/main.py index f62b20d..02ed503 100644 --- a/drb-c2-core/app/main.py +++ b/drb-c2-core/app/main.py @@ -10,6 +10,7 @@ from app.internal.summarizer import summarizer_loop from app.internal.vocabulary_learner import vocabulary_induction_loop from app.internal.recorrelation_sweep import recorrelation_loop from app.internal import ai_health +from app.internal import discord_streams from app.config import settings from app.internal.auth import ( require_firebase_token, @@ -18,31 +19,42 @@ from app.internal.auth import ( ) from app.routers import nodes, systems, calls, upload, tokens, incidents, alerts, admin, trips, places, links, users from app.routers import enrollment, media, org, waitlist, telemetry, replay, traffic +from app.routers import discord as discord_router from app.internal import dynsec from app.internal import firestore as fstore -async def _release_orphaned_tokens(): - """Release all in-use tokens on startup — voice connections don't survive server restarts.""" +async def _clear_legacy_token_flags(): + """Tokens written by the pre-lease code carry ``in_use`` with no ``lease``. + + Nothing reads ``in_use`` for decisions any more (discord_leases.py decides + on ``lease``), but the /tokens page shows it, so a legacy "in use" with no + lease behind it would be a permanent false label. A token that HAS a lease + is left alone: voice connections now belong to the edge and survive a C2 + restart, so releasing leases at startup (what this used to do) would hand + a still-logged-in token to a second node. + """ def _find(): from app.internal.firestore import db return [d for d in db.collection("bot_tokens").where("in_use", "==", True).stream()] results = await asyncio.to_thread(_find) + cleared = 0 for doc in results: + if (doc.to_dict() or {}).get("lease"): + continue await fstore.doc_update("bot_tokens", doc.id, { - "in_use": False, - "assigned_node_id": None, - "assigned_at": None, + "lease": None, "in_use": False, "assigned_node_id": None, "assigned_at": None, }) - if results: - logger.info(f"Released {len(results)} orphaned token(s) on startup.") + cleared += 1 + if cleared: + logger.info(f"Cleared legacy in_use flag on {cleared} token(s) with no lease.") @asynccontextmanager async def lifespan(app: FastAPI): logger.info("DRB C2 Core starting.") - await _release_orphaned_tokens() + await _clear_legacy_token_flags() # dynsec bootstrap + reconcile — must happen before mqtt_handler.connect() # so that by the time the app is serving requests, c2-core's own dynsec @@ -65,6 +77,7 @@ async def lifespan(app: FastAPI): summarizer_task = asyncio.create_task(summarizer_loop()) induction_task = asyncio.create_task(vocabulary_induction_loop()) recorrelation_task = asyncio.create_task(recorrelation_loop()) + discord_task = asyncio.create_task(discord_streams.reconcile_loop()) yield # --- app running --- @@ -73,6 +86,7 @@ async def lifespan(app: FastAPI): summarizer_task.cancel() induction_task.cancel() recorrelation_task.cancel() + discord_task.cancel() await mqtt_handler.disconnect() @@ -139,6 +153,9 @@ app.include_router(waitlist.router) # public — no auth, source-IP rate limi # public by necessity — an