Logan CusanoandClaude Opus 5.5
455c55f5c2
- c2-core: POST /discord/link-codes (org owner/admin) mints a single-use,
10-minute code; minting revokes the org's older codes. The bot redeems it
via POST /discord/guilds/{id}/link/redeem (service key) inside one
Firestore transaction, so two servers racing for one code cannot both win
and a server linked to org A refuses org B's code. GET/DELETE
/discord/linked-guilds lists/unlinks the caller's org's servers; unlink
stops that server's streams. Rate-limited (mint per user, redeem per
server and per Discord user); every mint/link/refusal/unlink audited.
- resolve_guild_org: an explicitly unlinked server no longer falls back to
DISCORD_UNLINKED_GUILD_ORG_ID. Linking away from the fallback org stops
the streams it had running for that org.
- Discord bot: /drb link code: (Manage Server, enforced at runtime too);
GatedTree answers every other command in an unlinked server with how to
link (autocomplete returns nothing). With the default fallback org set,
nothing is gated — unchanged behaviour.
- Frontend: "Linked Discord servers" card on Settings -> Organization —
generate code with countdown + copy, polls for the new link, unlink.
- 16 tests: single-use, two-server race, one-server/two-org race, expiry,
revoked/malformed codes, wrong-org takeover, idempotent relink, rate
limits, fallback stop, scoped list, unlink without fallback.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
516 lines
23 KiB
Python
516 lines
23 KiB
Python
"""
|
|
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:
|
|
|
|
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, since, at} — reported by the node, never guessed here
|
|
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
|
|
|
|
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.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):
|
|
super().__init__(message)
|
|
self.code = code
|
|
self.message = message
|
|
self.invite = invite or []
|
|
self.status = status
|
|
|
|
|
|
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 /leave first.",
|
|
}
|
|
|
|
|
|
async def start_stream(
|
|
*,
|
|
org_id: str,
|
|
guild_id: str,
|
|
channel_id: str,
|
|
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,
|
|
) -> dict:
|
|
"""Make ``guild_id:channel_id`` carry ``system_id``. Idempotent: re-running
|
|
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 {}
|
|
|
|
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 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 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:
|
|
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,
|
|
"delivered": delivered}
|
|
last = res
|
|
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 [])
|
|
|
|
|
|
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})")
|
|
return True
|
|
|
|
|
|
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) -> 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"}
|
|
|
|
|
|
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):
|
|
continue
|
|
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
|
|
if st.get("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:
|
|
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)) 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, 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)
|
|
|
|
# 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}")
|