install.sh: enrollment token never on the command line (#18) #30
@@ -38,6 +38,10 @@ jobs:
|
||||
run: |
|
||||
echo "REPO_NAME=$(echo ${GITHUB_REPOSITORY} | awk -F'/' '{print $2}')" >> $GITHUB_OUTPUT
|
||||
echo "VERSION=$(git describe --tags --always | sed 's/^v//')" >> $GITHUB_OUTPUT
|
||||
# Source pin (node-26#19): the git tree hash of the build context.
|
||||
# A node pulls src-<tree of its own checkout> — scripts/image-tags.sh
|
||||
# computes the identical value — so its code and image always match.
|
||||
echo "SRC_TAG=src-$(git rev-parse HEAD:drb-edge-node)" >> $GITHUB_OUTPUT
|
||||
|
||||
- uses: docker/build-push-action@v6
|
||||
with:
|
||||
@@ -47,6 +51,7 @@ jobs:
|
||||
push: true
|
||||
tags: |
|
||||
git.vpn.cusano.net/${{ vars.DOCKER_ORG }}/${{ steps.meta.outputs.REPO_NAME }}/${{ env.CONTAINER_NAME }}:${{ steps.meta.outputs.VERSION }}
|
||||
git.vpn.cusano.net/${{ vars.DOCKER_ORG }}/${{ steps.meta.outputs.REPO_NAME }}/${{ env.CONTAINER_NAME }}:${{ steps.meta.outputs.SRC_TAG }}
|
||||
git.vpn.cusano.net/${{ vars.DOCKER_ORG }}/${{ steps.meta.outputs.REPO_NAME }}/${{ env.CONTAINER_NAME }}:latest
|
||||
cache-from: type=registry,ref=git.vpn.cusano.net/${{ vars.DOCKER_ORG }}/${{ steps.meta.outputs.REPO_NAME }}/${{ env.CONTAINER_NAME }}:buildcache
|
||||
cache-to: type=registry,ref=git.vpn.cusano.net/${{ vars.DOCKER_ORG }}/${{ steps.meta.outputs.REPO_NAME }}/${{ env.CONTAINER_NAME }}:buildcache,mode=max
|
||||
|
||||
@@ -38,6 +38,10 @@ jobs:
|
||||
run: |
|
||||
echo "REPO_NAME=$(echo ${GITHUB_REPOSITORY} | awk -F'/' '{print $2}')" >> $GITHUB_OUTPUT
|
||||
echo "VERSION=$(git describe --tags --always | sed 's/^v//')" >> $GITHUB_OUTPUT
|
||||
# Source pin (node-26#19): the git tree hash of the build context.
|
||||
# A node pulls src-<tree of its own checkout> — scripts/image-tags.sh
|
||||
# computes the identical value — so its code and image always match.
|
||||
echo "SRC_TAG=src-$(git rev-parse HEAD:icecast)" >> $GITHUB_OUTPUT
|
||||
|
||||
- uses: docker/build-push-action@v6
|
||||
with:
|
||||
@@ -47,6 +51,7 @@ jobs:
|
||||
push: true
|
||||
tags: |
|
||||
git.vpn.cusano.net/${{ vars.DOCKER_ORG }}/${{ steps.meta.outputs.REPO_NAME }}/${{ env.CONTAINER_NAME }}:${{ steps.meta.outputs.VERSION }}
|
||||
git.vpn.cusano.net/${{ vars.DOCKER_ORG }}/${{ steps.meta.outputs.REPO_NAME }}/${{ env.CONTAINER_NAME }}:${{ steps.meta.outputs.SRC_TAG }}
|
||||
git.vpn.cusano.net/${{ vars.DOCKER_ORG }}/${{ steps.meta.outputs.REPO_NAME }}/${{ env.CONTAINER_NAME }}:latest
|
||||
cache-from: type=registry,ref=git.vpn.cusano.net/${{ vars.DOCKER_ORG }}/${{ steps.meta.outputs.REPO_NAME }}/${{ env.CONTAINER_NAME }}:buildcache
|
||||
cache-to: type=registry,ref=git.vpn.cusano.net/${{ vars.DOCKER_ORG }}/${{ steps.meta.outputs.REPO_NAME }}/${{ env.CONTAINER_NAME }}:buildcache,mode=max
|
||||
|
||||
@@ -38,6 +38,10 @@ jobs:
|
||||
run: |
|
||||
echo "REPO_NAME=$(echo ${GITHUB_REPOSITORY} | awk -F'/' '{print $2}')" >> $GITHUB_OUTPUT
|
||||
echo "VERSION=$(git describe --tags --always | sed 's/^v//')" >> $GITHUB_OUTPUT
|
||||
# Source pin (node-26#19): the git tree hash of the build context.
|
||||
# A node pulls src-<tree of its own checkout> — scripts/image-tags.sh
|
||||
# computes the identical value — so its code and image always match.
|
||||
echo "SRC_TAG=src-$(git rev-parse HEAD:op25-container)" >> $GITHUB_OUTPUT
|
||||
|
||||
- uses: docker/build-push-action@v6
|
||||
with:
|
||||
@@ -47,6 +51,7 @@ jobs:
|
||||
push: true
|
||||
tags: |
|
||||
git.vpn.cusano.net/${{ vars.DOCKER_ORG }}/${{ steps.meta.outputs.REPO_NAME }}/${{ env.CONTAINER_NAME }}:${{ steps.meta.outputs.VERSION }}
|
||||
git.vpn.cusano.net/${{ vars.DOCKER_ORG }}/${{ steps.meta.outputs.REPO_NAME }}/${{ env.CONTAINER_NAME }}:${{ steps.meta.outputs.SRC_TAG }}
|
||||
git.vpn.cusano.net/${{ vars.DOCKER_ORG }}/${{ steps.meta.outputs.REPO_NAME }}/${{ env.CONTAINER_NAME }}:stable
|
||||
# Registry cache — avoids full recompile of GnuRadio/OP25 when only app code changes
|
||||
cache-from: type=registry,ref=git.vpn.cusano.net/${{ vars.DOCKER_ORG }}/${{ steps.meta.outputs.REPO_NAME }}/${{ env.CONTAINER_NAME }}:buildcache
|
||||
|
||||
@@ -38,6 +38,10 @@ jobs:
|
||||
run: |
|
||||
echo "REPO_NAME=$(echo ${GITHUB_REPOSITORY} | awk -F'/' '{print $2}')" >> $GITHUB_OUTPUT
|
||||
echo "VERSION=$(git describe --tags --always | sed 's/^v//')" >> $GITHUB_OUTPUT
|
||||
# Source pin (node-26#19): the git tree hash of the build context.
|
||||
# A node pulls src-<tree of its own checkout> — scripts/image-tags.sh
|
||||
# computes the identical value — so its code and image always match.
|
||||
echo "SRC_TAG=src-$(git rev-parse HEAD:secondary-sdr-container)" >> $GITHUB_OUTPUT
|
||||
|
||||
- uses: docker/build-push-action@v6
|
||||
with:
|
||||
@@ -47,6 +51,7 @@ jobs:
|
||||
push: true
|
||||
tags: |
|
||||
git.vpn.cusano.net/${{ vars.DOCKER_ORG }}/${{ steps.meta.outputs.REPO_NAME }}/${{ env.CONTAINER_NAME }}:${{ steps.meta.outputs.VERSION }}
|
||||
git.vpn.cusano.net/${{ vars.DOCKER_ORG }}/${{ steps.meta.outputs.REPO_NAME }}/${{ env.CONTAINER_NAME }}:${{ steps.meta.outputs.SRC_TAG }}
|
||||
git.vpn.cusano.net/${{ vars.DOCKER_ORG }}/${{ steps.meta.outputs.REPO_NAME }}/${{ env.CONTAINER_NAME }}:latest
|
||||
cache-from: type=registry,ref=git.vpn.cusano.net/${{ vars.DOCKER_ORG }}/${{ steps.meta.outputs.REPO_NAME }}/${{ env.CONTAINER_NAME }}:buildcache
|
||||
cache-to: type=registry,ref=git.vpn.cusano.net/${{ vars.DOCKER_ORG }}/${{ steps.meta.outputs.REPO_NAME }}/${{ env.CONTAINER_NAME }}:buildcache,mode=max
|
||||
|
||||
@@ -0,0 +1,87 @@
|
||||
name: Release images
|
||||
|
||||
# Make sure every node image a release ref needs exists in the registry
|
||||
# (node-26#19). A node on `v1` pulls <image>:src-<tree hash of that image's
|
||||
# directory at v1>. The per-image build workflows only run when their own
|
||||
# directory changes on main, and anything built before #19 has no src- tag at
|
||||
# all, so a release tag can point at trees nobody ever built. This fills the
|
||||
# gaps: for each image, if src-<tree> is already in the registry it does
|
||||
# nothing; otherwise it builds and pushes it (registry cache keeps op25 off a
|
||||
# full recompile when only its app code changed).
|
||||
#
|
||||
# Runs on every v* tag push, including a re-cut `v1`. Also runnable by hand
|
||||
# (workflow_dispatch) on any ref — run it once on the ref `v1` points at
|
||||
# before re-cutting `v1` onto a commit that contains #19, so nodes never see
|
||||
# a moment where their pinned images don't exist yet.
|
||||
#
|
||||
# Pushes ONLY src- tags. :latest / :stable keep tracking main, as before.
|
||||
|
||||
on:
|
||||
workflow_dispatch:
|
||||
push:
|
||||
tags:
|
||||
- "v*"
|
||||
|
||||
jobs:
|
||||
ensure:
|
||||
runs-on: ubuntu-latest
|
||||
permissions:
|
||||
contents: read
|
||||
packages: write
|
||||
strategy:
|
||||
fail-fast: false
|
||||
matrix:
|
||||
include:
|
||||
- name: edge-node
|
||||
context: drb-edge-node
|
||||
- name: icecast
|
||||
context: icecast
|
||||
- name: op25-client
|
||||
context: op25-container
|
||||
- name: secondary-sdr
|
||||
context: secondary-sdr-container
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
|
||||
- uses: docker/setup-qemu-action@v3
|
||||
|
||||
- uses: docker/setup-buildx-action@v3
|
||||
with:
|
||||
config-inline: |
|
||||
[registry."git.vpn.cusano.net"]
|
||||
http = false
|
||||
insecure = false
|
||||
|
||||
- uses: docker/login-action@v3
|
||||
with:
|
||||
registry: git.vpn.cusano.net
|
||||
username: ${{ gitea.actor }}
|
||||
password: ${{ secrets.BUILD_TOKEN }}
|
||||
|
||||
- name: Resolve source tag
|
||||
id: meta
|
||||
run: |
|
||||
REPO_NAME="$(echo ${GITHUB_REPOSITORY} | awk -F'/' '{print $2}')"
|
||||
# Must match scripts/image-tags.sh and the build-*.yml workflows.
|
||||
IMAGE="git.vpn.cusano.net/${{ vars.DOCKER_ORG }}/${REPO_NAME}/${{ matrix.name }}"
|
||||
TAG="src-$(git rev-parse HEAD:${{ matrix.context }})"
|
||||
echo "IMAGE=$IMAGE" >> $GITHUB_OUTPUT
|
||||
echo "TAG=$TAG" >> $GITHUB_OUTPUT
|
||||
if docker buildx imagetools inspect "$IMAGE:$TAG" >/dev/null 2>&1; then
|
||||
echo "EXISTS=true" >> $GITHUB_OUTPUT
|
||||
echo "$IMAGE:$TAG already published — nothing to do"
|
||||
else
|
||||
echo "EXISTS=false" >> $GITHUB_OUTPUT
|
||||
echo "$IMAGE:$TAG missing — building it"
|
||||
fi
|
||||
|
||||
- if: steps.meta.outputs.EXISTS != 'true'
|
||||
uses: docker/build-push-action@v6
|
||||
with:
|
||||
context: ./${{ matrix.context }}
|
||||
file: ./${{ matrix.context }}/Dockerfile
|
||||
# arm64 only, same as the build-*.yml workflows (node-26#28).
|
||||
platforms: linux/arm64
|
||||
push: true
|
||||
tags: ${{ steps.meta.outputs.IMAGE }}:${{ steps.meta.outputs.TAG }}
|
||||
cache-from: type=registry,ref=${{ steps.meta.outputs.IMAGE }}:buildcache
|
||||
@@ -13,6 +13,7 @@ venv/
|
||||
# Runtime-generated configs and recordings
|
||||
configs/
|
||||
recordings/
|
||||
.drb-update-compose.yml
|
||||
|
||||
# Logs
|
||||
*.log
|
||||
|
||||
+8
-4
@@ -1,6 +1,10 @@
|
||||
# Image tags: a provisioned node pins every image to the source tree it has
|
||||
# checked out (DRB_TAG_* = src-<git tree hash>, written into .env by install.sh
|
||||
# and scripts/drb-update.sh via scripts/image-tags.sh — node-26#19). Unset, as
|
||||
# in local dev, they fall back to :latest / :stable, which CI builds from main.
|
||||
services:
|
||||
icecast:
|
||||
image: ${IMAGE_REGISTRY:-git.vpn.cusano.net}/${DOCKER_ORG:-logan}/${DOCKER_REPO:-node-26}/icecast:latest
|
||||
image: ${IMAGE_REGISTRY:-git.vpn.cusano.net}/${DOCKER_ORG:-logan}/${DOCKER_REPO:-node-26}/icecast:${DRB_TAG_ICECAST:-latest}
|
||||
build: ./icecast
|
||||
restart: unless-stopped
|
||||
network_mode: host
|
||||
@@ -16,7 +20,7 @@ services:
|
||||
# rather than being exposed. edge-node still reaches both over localhost
|
||||
# because it shares this host network namespace.
|
||||
op25:
|
||||
image: ${IMAGE_REGISTRY:-git.vpn.cusano.net}/${DOCKER_ORG:-logan}/${DOCKER_REPO:-node-26}/op25-client:stable
|
||||
image: ${IMAGE_REGISTRY:-git.vpn.cusano.net}/${DOCKER_ORG:-logan}/${DOCKER_REPO:-node-26}/op25-client:${DRB_TAG_OP25:-stable}
|
||||
build: ./op25-container
|
||||
restart: unless-stopped
|
||||
privileged: true
|
||||
@@ -38,7 +42,7 @@ services:
|
||||
# node-26#9. Same network/device access as op25 for the same reason: it
|
||||
# needs the raw USB device, not a virtualized one.
|
||||
secondary-sdr:
|
||||
image: ${IMAGE_REGISTRY:-git.vpn.cusano.net}/${DOCKER_ORG:-logan}/${DOCKER_REPO:-node-26}/secondary-sdr:latest
|
||||
image: ${IMAGE_REGISTRY:-git.vpn.cusano.net}/${DOCKER_ORG:-logan}/${DOCKER_REPO:-node-26}/secondary-sdr:${DRB_TAG_SECONDARY_SDR:-latest}
|
||||
build: ./secondary-sdr-container
|
||||
restart: unless-stopped
|
||||
privileged: true
|
||||
@@ -48,7 +52,7 @@ services:
|
||||
- /dev:/dev
|
||||
|
||||
edge-node:
|
||||
image: ${IMAGE_REGISTRY:-git.vpn.cusano.net}/${DOCKER_ORG:-logan}/${DOCKER_REPO:-node-26}/edge-node:latest
|
||||
image: ${IMAGE_REGISTRY:-git.vpn.cusano.net}/${DOCKER_ORG:-logan}/${DOCKER_REPO:-node-26}/edge-node:${DRB_TAG_EDGE_NODE:-latest}
|
||||
build: ./drb-edge-node
|
||||
restart: unless-stopped
|
||||
network_mode: host
|
||||
|
||||
@@ -54,6 +54,13 @@ class Settings(BaseSettings):
|
||||
# Bounded wait for the shared PulseAudio socket before launching FFmpeg.
|
||||
pulse_wait_timeout: float = 30.0
|
||||
|
||||
# Discord radio (node-26#25): with the broker unreachable this long, log
|
||||
# every radio bot out. MUST stay below C2's
|
||||
# DISCORD_LEASE_OFFLINE_RELEASE_SECONDS (default 600), after which C2 hands
|
||||
# this node's bot tokens to other nodes — so a token is never logged in on
|
||||
# two nodes at once.
|
||||
discord_mqtt_loss_stop_seconds: int = 300
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Call segmentation
|
||||
#
|
||||
|
||||
@@ -1,299 +1,484 @@
|
||||
"""
|
||||
Discord radio streams on the edge, driven by desired state (node-26#25,
|
||||
server-26#203/#204).
|
||||
|
||||
C2 retains ``nodes/{id}/discord/desired`` = the FULL list of streams this node
|
||||
should carry: ``{seq, streams: [{stream_id, guild_id, channel_id, token_id,
|
||||
system_name, ...}]}``. This module reconciles toward that list and publishes
|
||||
``nodes/{id}/discord/actual`` = ``{applied_seq, sessions: [token_id],
|
||||
streams: {stream_id: {state, error, code}}}`` after every change.
|
||||
|
||||
Rules this enforces by construction:
|
||||
|
||||
* One gateway session per leased token; a token no stream needs any more is
|
||||
logged OUT (``client.close()``), not just disconnected from voice. C2 frees
|
||||
the token only after an ``actual`` with ``applied_seq`` >= the drop and the
|
||||
token absent from ``sessions`` — so one token is never logged in on two
|
||||
nodes (node-26#25 item 1).
|
||||
* One voice connection per (token, guild); one token can serve many guilds.
|
||||
* The watchdog is a separate reconcile loop that never closes itself: closing
|
||||
a session cannot cancel it (node-26#25 item 2).
|
||||
* No message-content intent, no ``@bot leave/join`` mention controls — voice
|
||||
is controlled only through C2 (node-26#25 item 3).
|
||||
* MQTT-loss fail-safe: if the broker has been unreachable for
|
||||
``discord_mqtt_loss_stop_seconds`` every radio bot is logged out, because C2
|
||||
returns this node's tokens to the pool after its own (longer) grace.
|
||||
|
||||
Bot token strings never travel over MQTT: they are fetched over HTTPS from C2
|
||||
(``GET /discord/node/{node}/tokens/{id}``, node API key), which hands a token
|
||||
only to the node holding its lease, and are kept in memory only.
|
||||
|
||||
Audio: each voice connection plays the PulseAudio monitor (``-f pulse`` on
|
||||
``settings.pulse_source``) — the accuracy path, never Icecast.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from typing import Optional
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
from typing import Awaitable, Callable, Optional
|
||||
|
||||
import discord
|
||||
from discord.ext import commands
|
||||
|
||||
from app.config import settings
|
||||
from app.internal import pulse
|
||||
from app.internal.logger import logger
|
||||
|
||||
BOT_READY_TIMEOUT = 15 # seconds to wait for Discord bot to become ready
|
||||
WATCHDOG_INTERVAL = 30 # seconds between voice-connection health checks
|
||||
REJOIN_DELAY = 5 # seconds to wait before attempting a rejoin
|
||||
STREAM_RETRY_DELAY = 5 # seconds to back off before re-arming the audio source
|
||||
BOT_READY_TIMEOUT = 15 # seconds to wait for a Discord login to become ready
|
||||
WATCHDOG_INTERVAL = 15 # seconds between reconcile passes
|
||||
STREAM_RETRY_DELAY = 5 # seconds to back off before re-arming an audio source
|
||||
REPORT_EVERY_TICKS = 4 # unconditional actual-state report every ~60s
|
||||
_VOICE_TYPES = (discord.ChannelType.voice, discord.ChannelType.stage_voice)
|
||||
|
||||
|
||||
class RadioBot:
|
||||
def __init__(self):
|
||||
self._bot: Optional[commands.Bot] = None
|
||||
self._voice_client: Optional[discord.VoiceClient] = None
|
||||
self._task: Optional[asyncio.Task] = None
|
||||
self._watchdog_task: Optional[asyncio.Task] = None
|
||||
self._ready_event: Optional[asyncio.Event] = None
|
||||
self._current_token: Optional[str] = None
|
||||
@dataclass
|
||||
class _Session:
|
||||
token_id: str
|
||||
client: object
|
||||
task: Optional[asyncio.Task] = None
|
||||
presence: Optional[str] = None
|
||||
|
||||
def alive(self) -> bool:
|
||||
return self.task is not None and not self.task.done() and not self.client.is_closed()
|
||||
|
||||
|
||||
@dataclass
|
||||
class _Voice:
|
||||
stream_id: str
|
||||
token_id: str
|
||||
guild_id: int
|
||||
channel_id: int
|
||||
vc: Optional[object] = None
|
||||
state: str = "connecting"
|
||||
error: Optional[str] = None
|
||||
code: Optional[str] = None
|
||||
|
||||
|
||||
def _default_client_factory():
|
||||
intents = discord.Intents.none()
|
||||
intents.guilds = True
|
||||
intents.voice_states = True
|
||||
return discord.Client(intents=intents)
|
||||
|
||||
|
||||
def _default_audio_factory():
|
||||
# before_options land ahead of -i: ffmpeg -f pulse -i drb_sink.monitor …
|
||||
source = discord.FFmpegPCMAudio(settings.pulse_source, before_options="-f pulse")
|
||||
return discord.PCMVolumeTransformer(source, volume=1.0)
|
||||
|
||||
|
||||
async def _default_token_fetcher(token_id: str) -> Optional[str]:
|
||||
import httpx
|
||||
from app.internal import credentials
|
||||
api_key = credentials.get_api_key()
|
||||
if not settings.c2_url or not api_key:
|
||||
logger.error("Cannot fetch a radio-bot token: C2_URL or the node API key is missing.")
|
||||
return None
|
||||
url = f"{settings.c2_url.rstrip('/')}/discord/node/{settings.node_id}/tokens/{token_id}"
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=10) as client:
|
||||
r = await client.get(url, headers={"Authorization": f"Bearer {api_key}"})
|
||||
if r.status_code != 200:
|
||||
logger.error(f"C2 refused radio-bot token {token_id}: HTTP {r.status_code}")
|
||||
return None
|
||||
return r.json().get("token")
|
||||
except Exception as e:
|
||||
logger.error(f"Fetching radio-bot token {token_id} failed: {e}")
|
||||
return None
|
||||
|
||||
|
||||
async def _default_publisher(payload: dict) -> None:
|
||||
from app.internal.mqtt_manager import mqtt_manager
|
||||
mqtt_manager.publish_discord_actual(payload)
|
||||
|
||||
|
||||
def _default_mqtt_up() -> bool:
|
||||
from app.internal.mqtt_manager import mqtt_manager
|
||||
return mqtt_manager.is_connected
|
||||
|
||||
|
||||
class RadioManager:
|
||||
def __init__(
|
||||
self,
|
||||
client_factory: Callable[[], object] = _default_client_factory,
|
||||
token_fetcher: Callable[[str], Awaitable[Optional[str]]] = _default_token_fetcher,
|
||||
publisher: Callable[[dict], Awaitable[None]] = _default_publisher,
|
||||
audio_factory: Callable[[], object] = _default_audio_factory,
|
||||
mqtt_up: Callable[[], bool] = _default_mqtt_up,
|
||||
ready_timeout: float = BOT_READY_TIMEOUT,
|
||||
):
|
||||
self._client_factory = client_factory
|
||||
self._fetch_token = token_fetcher
|
||||
self._publish = publisher
|
||||
self._audio_factory = audio_factory
|
||||
self._mqtt_up = mqtt_up
|
||||
self._ready_timeout = ready_timeout
|
||||
|
||||
self._desired: dict[str, dict] = {}
|
||||
self._applied_seq: Optional[int] = None
|
||||
self._sessions: dict[str, _Session] = {}
|
||||
self._voices: dict[str, _Voice] = {}
|
||||
self._session_errors: dict[str, str] = {}
|
||||
self._tokens: dict[str, str] = {} # token_id -> string, memory only
|
||||
self._lock = asyncio.Lock()
|
||||
self._loop: Optional[asyncio.AbstractEventLoop] = None
|
||||
self._watchdog: Optional[asyncio.Task] = None
|
||||
self._mqtt_down_since: Optional[float] = None
|
||||
self._failsafe = False
|
||||
self._paused = False
|
||||
self._ticks = 0
|
||||
|
||||
# Remembered so we can rejoin after an unexpected disconnect
|
||||
self._guild_id: Optional[int] = None
|
||||
self._channel_id: Optional[int] = None
|
||||
self._was_streaming: bool = False
|
||||
# ------------------------------------------------------------------
|
||||
# Inputs
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
async def join(self, guild_id: int, channel_id: int, token: str,
|
||||
call_active: bool = False, system_name: str = None) -> bool:
|
||||
# (Re)start the bot if the token changed or the bot isn't running
|
||||
if self._current_token != token or not self._is_bot_running():
|
||||
if not await self._start_bot(token):
|
||||
return False
|
||||
|
||||
guild = self._bot.get_guild(guild_id)
|
||||
if not guild:
|
||||
logger.error(f"Guild {guild_id} not found — bot may not be a member.")
|
||||
return False
|
||||
|
||||
channel = guild.get_channel(channel_id)
|
||||
if not isinstance(channel, discord.VoiceChannel):
|
||||
logger.error(f"Channel {channel_id} is not a voice channel.")
|
||||
return False
|
||||
|
||||
try:
|
||||
if self._voice_client and self._voice_client.is_connected():
|
||||
await self._voice_client.disconnect(force=True)
|
||||
self._voice_client = await channel.connect()
|
||||
# Remember where we are so the watchdog can rejoin if we drop
|
||||
self._guild_id = guild_id
|
||||
self._channel_id = channel_id
|
||||
# Bounded wait for the shared PulseAudio socket. Historically FFmpeg was
|
||||
# launched before the op25 container had created it, failed instantly,
|
||||
# and the bot sat silently connected forever.
|
||||
await pulse.wait_until_ready()
|
||||
self._play_stream()
|
||||
if system_name:
|
||||
await self._bot.change_presence(
|
||||
activity=discord.Activity(
|
||||
type=discord.ActivityType.listening,
|
||||
name=system_name,
|
||||
)
|
||||
)
|
||||
logger.info(f"Joined #{channel.name} in {guild.name}")
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to join voice channel: {e}")
|
||||
return False
|
||||
|
||||
async def leave(self) -> bool:
|
||||
# Clear remembered channel so the watchdog doesn't rejoin
|
||||
self._guild_id = None
|
||||
self._channel_id = None
|
||||
self._was_streaming = False
|
||||
|
||||
if self._voice_client and self._voice_client.is_connected():
|
||||
async def apply_desired(self, payload: dict) -> None:
|
||||
"""Handle a (retained) desired-state message from C2."""
|
||||
if not isinstance(payload, dict):
|
||||
return
|
||||
seq = payload.get("seq")
|
||||
seq = seq if isinstance(seq, int) and not isinstance(seq, bool) else None
|
||||
if seq is not None and self._applied_seq is not None and seq < self._applied_seq:
|
||||
logger.warning(f"Ignoring stale Discord desired state seq {seq} (applied {self._applied_seq})")
|
||||
return
|
||||
wanted = {}
|
||||
for s in payload.get("streams") or []:
|
||||
try:
|
||||
self._stop_stream()
|
||||
await self._voice_client.disconnect(force=True)
|
||||
self._voice_client = None
|
||||
await self._bot.change_presence(activity=None)
|
||||
logger.info("Disconnected from voice channel.")
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to disconnect: {e}")
|
||||
return False
|
||||
wanted[str(s["stream_id"])] = {
|
||||
"stream_id": str(s["stream_id"]), "guild_id": int(s["guild_id"]),
|
||||
"channel_id": int(s["channel_id"]), "token_id": str(s["token_id"]),
|
||||
"system_name": s.get("system_name"),
|
||||
}
|
||||
except (KeyError, TypeError, ValueError):
|
||||
logger.error(f"Malformed Discord desired stream entry ignored: {s!r}")
|
||||
self._ensure_watchdog()
|
||||
self._desired = wanted
|
||||
self._paused = False
|
||||
await self.reconcile()
|
||||
if seq is not None:
|
||||
self._applied_seq = seq
|
||||
await self.publish_actual()
|
||||
|
||||
async def pause_locally(self) -> None:
|
||||
"""Local-dashboard stop: log every bot out until C2 sends a new list."""
|
||||
self._paused = True
|
||||
await self.reconcile()
|
||||
await self.publish_actual()
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Reconcile
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def _target(self) -> dict[str, dict]:
|
||||
return {} if (self._paused or self._failsafe) else self._desired
|
||||
|
||||
async def reconcile(self) -> None:
|
||||
async with self._lock:
|
||||
target = self._target()
|
||||
# 1) voice connections no longer wanted (or re-pointed)
|
||||
for sid, v in list(self._voices.items()):
|
||||
d = target.get(sid)
|
||||
if not d or (d["token_id"], d["guild_id"], d["channel_id"]) != (v.token_id, v.guild_id, v.channel_id):
|
||||
await self._disconnect(v)
|
||||
del self._voices[sid]
|
||||
# 2) gateway sessions no stream needs: LOG OUT, not just leave voice
|
||||
needed = {d["token_id"] for d in target.values()}
|
||||
for tid in list(self._sessions):
|
||||
if tid not in needed:
|
||||
await self._close_session(tid)
|
||||
for tid in list(self._tokens):
|
||||
if tid not in needed:
|
||||
self._tokens.pop(tid, None)
|
||||
# 3) sessions + voice for everything wanted
|
||||
for sid, d in sorted(target.items()):
|
||||
v = self._voices.get(sid)
|
||||
if v is None:
|
||||
v = self._voices[sid] = _Voice(sid, d["token_id"], d["guild_id"], d["channel_id"])
|
||||
sess = await self._ensure_session(d["token_id"])
|
||||
if sess is None:
|
||||
v.state, v.code = "error", "login_failed"
|
||||
v.error = self._session_errors.get(d["token_id"], "The radio bot could not log in to Discord.")
|
||||
continue
|
||||
await self._ensure_voice(sess, v, d)
|
||||
|
||||
async def _ensure_session(self, token_id: str) -> Optional[_Session]:
|
||||
sess = self._sessions.get(token_id)
|
||||
if sess and sess.alive():
|
||||
return sess
|
||||
if sess:
|
||||
await self._close_session(token_id)
|
||||
token = self._tokens.get(token_id) or await self._fetch_token(token_id)
|
||||
if not token:
|
||||
self._session_errors[token_id] = "DRB would not hand this node the radio bot's token."
|
||||
return None
|
||||
client = self._client_factory()
|
||||
sess = _Session(token_id, client)
|
||||
ready = asyncio.Event()
|
||||
|
||||
async def on_ready():
|
||||
ready.set()
|
||||
client.event(on_ready)
|
||||
sess.task = asyncio.create_task(client.start(token))
|
||||
waiter = asyncio.create_task(ready.wait())
|
||||
done, _ = await asyncio.wait({waiter, sess.task}, timeout=self._ready_timeout,
|
||||
return_when=asyncio.FIRST_COMPLETED)
|
||||
if waiter not in done:
|
||||
waiter.cancel()
|
||||
err = sess.task.exception() if sess.task.done() and not sess.task.cancelled() else None
|
||||
msg = f"{type(err).__name__}: {err}" if err else f"not ready after {self._ready_timeout:.0f}s"
|
||||
logger.error(f"Radio bot {token_id} failed to log in: {msg}")
|
||||
self._session_errors[token_id] = f"The radio bot could not log in to Discord ({msg[:120]})."
|
||||
await self._shutdown_client(sess)
|
||||
if isinstance(err, discord.LoginFailure):
|
||||
self._tokens.pop(token_id, None)
|
||||
return None
|
||||
self._tokens[token_id] = token
|
||||
self._sessions[token_id] = sess
|
||||
self._session_errors.pop(token_id, None)
|
||||
logger.info(f"Radio bot {token_id} logged in as {getattr(client, 'user', None)}")
|
||||
return sess
|
||||
|
||||
async def _ensure_voice(self, sess: _Session, v: _Voice, d: dict) -> None:
|
||||
client = sess.client
|
||||
guild = client.get_guild(v.guild_id)
|
||||
if guild is None:
|
||||
v.state, v.code = "error", "not_in_guild"
|
||||
v.error = "This radio bot is not a member of that Discord server — invite it first."
|
||||
return
|
||||
channel = guild.get_channel(v.channel_id)
|
||||
if channel is None or getattr(channel, "type", None) not in _VOICE_TYPES:
|
||||
v.state, v.code = "error", "bad_channel"
|
||||
v.error = "That channel is not a voice channel the radio bot can see."
|
||||
return
|
||||
vc = v.vc
|
||||
if vc is not None and vc.is_connected() and getattr(vc.channel, "id", None) == v.channel_id:
|
||||
if not vc.is_playing():
|
||||
self._play(v)
|
||||
v.state, v.error, v.code = "connected", None, None
|
||||
await self._set_presence(sess, d.get("system_name"))
|
||||
return
|
||||
v.state = "connecting"
|
||||
try:
|
||||
existing = guild.voice_client
|
||||
if existing is not None and existing.is_connected():
|
||||
await existing.move_to(channel)
|
||||
v.vc = existing
|
||||
else:
|
||||
if existing is not None:
|
||||
await existing.disconnect(force=True)
|
||||
v.vc = await channel.connect()
|
||||
except Exception as e:
|
||||
v.vc = None
|
||||
v.state, v.code = "error", "connect_failed"
|
||||
v.error = f"Could not connect to the voice channel ({type(e).__name__}: {str(e)[:120]})."
|
||||
logger.error(f"Stream {v.stream_id}: voice connect failed: {e}")
|
||||
return
|
||||
# Bounded wait for the shared PulseAudio socket (op25 container creates it).
|
||||
if not await pulse.wait_until_ready():
|
||||
v.state, v.code = "error", "no_audio"
|
||||
v.error = "The node's audio path (PulseAudio) is not up."
|
||||
return
|
||||
self._play(v, pulse_checked=True)
|
||||
v.state, v.error, v.code = "connected", None, None
|
||||
await self._set_presence(sess, d.get("system_name"))
|
||||
logger.info(f"Stream {v.stream_id} connected (bot {v.token_id})")
|
||||
|
||||
async def _set_presence(self, sess: _Session, name: Optional[str]) -> None:
|
||||
if not name or sess.presence == name:
|
||||
return
|
||||
try:
|
||||
await sess.client.change_presence(
|
||||
activity=discord.Activity(type=discord.ActivityType.listening, name=name))
|
||||
sess.presence = name
|
||||
except Exception as e:
|
||||
logger.warning(f"Presence update failed: {e}")
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Audio
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def _play(self, v: _Voice, pulse_checked: bool = False) -> None:
|
||||
vc = v.vc
|
||||
if vc is None or not vc.is_connected():
|
||||
return
|
||||
if not pulse_checked and not pulse.is_ready():
|
||||
logger.error(f"PulseAudio socket {pulse.socket_path()} missing — retrying audio in {STREAM_RETRY_DELAY}s.")
|
||||
self._schedule_replay(v.stream_id)
|
||||
return
|
||||
try:
|
||||
vc.play(self._audio_factory(), after=lambda err, sid=v.stream_id: self._on_audio_end(sid, err))
|
||||
except Exception as e:
|
||||
logger.error(f"Stream {v.stream_id}: could not start audio: {e}")
|
||||
self._schedule_replay(v.stream_id)
|
||||
|
||||
def _on_audio_end(self, stream_id: str, error) -> None:
|
||||
"""discord.py calls this from its player thread."""
|
||||
if error:
|
||||
logger.error(f"Stream {stream_id} audio ended with error: {error}")
|
||||
self._schedule_replay(stream_id, STREAM_RETRY_DELAY if error else 0.5)
|
||||
|
||||
def _schedule_replay(self, stream_id: str, delay: float = STREAM_RETRY_DELAY) -> None:
|
||||
loop = self._loop
|
||||
if loop is None or loop.is_closed():
|
||||
return
|
||||
|
||||
async def _later():
|
||||
await asyncio.sleep(delay)
|
||||
v = self._voices.get(stream_id)
|
||||
if v and v.vc and v.vc.is_connected() and not v.vc.is_playing():
|
||||
self._play(v)
|
||||
|
||||
loop.call_soon_threadsafe(lambda: asyncio.ensure_future(_later()))
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Teardown
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
async def _disconnect(self, v: _Voice) -> None:
|
||||
vc, v.vc = v.vc, None
|
||||
if vc is None:
|
||||
return
|
||||
try:
|
||||
if vc.is_playing():
|
||||
vc.stop()
|
||||
await vc.disconnect(force=True)
|
||||
except Exception as e:
|
||||
logger.warning(f"Stream {v.stream_id}: voice disconnect error: {e}")
|
||||
|
||||
async def _close_session(self, token_id: str) -> None:
|
||||
sess = self._sessions.pop(token_id, None)
|
||||
if sess is None:
|
||||
return
|
||||
for v in self._voices.values():
|
||||
if v.token_id == token_id:
|
||||
await self._disconnect(v)
|
||||
await self._shutdown_client(sess)
|
||||
logger.info(f"Radio bot {token_id} logged out")
|
||||
|
||||
async def _shutdown_client(self, sess: _Session) -> None:
|
||||
try:
|
||||
await sess.client.close()
|
||||
except Exception as e:
|
||||
logger.warning(f"Radio bot {sess.token_id} close error: {e}")
|
||||
if sess.task and not sess.task.done():
|
||||
sess.task.cancel()
|
||||
try:
|
||||
await sess.task
|
||||
except BaseException:
|
||||
pass
|
||||
|
||||
async def stop(self) -> None:
|
||||
"""Node shutdown: log every radio bot out."""
|
||||
if self._watchdog:
|
||||
self._watchdog.cancel()
|
||||
self._watchdog = None
|
||||
async with self._lock:
|
||||
for v in list(self._voices.values()):
|
||||
await self._disconnect(v)
|
||||
self._voices.clear()
|
||||
for tid in list(self._sessions):
|
||||
await self._close_session(tid)
|
||||
self._tokens.clear()
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Watchdog + MQTT-loss fail-safe
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def _ensure_watchdog(self) -> None:
|
||||
self._loop = asyncio.get_running_loop()
|
||||
if self._watchdog is None or self._watchdog.done():
|
||||
self._watchdog = asyncio.create_task(self._watchdog_loop())
|
||||
|
||||
async def _watchdog_loop(self) -> None:
|
||||
while True:
|
||||
await asyncio.sleep(WATCHDOG_INTERVAL)
|
||||
try:
|
||||
await self.tick()
|
||||
except asyncio.CancelledError:
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"Discord watchdog error: {e}")
|
||||
|
||||
async def tick(self, now: Optional[float] = None) -> None:
|
||||
"""One watchdog pass: fail-safe check, then reconcile + report if needed."""
|
||||
now = time.monotonic() if now is None else now
|
||||
before = self.snapshot()
|
||||
if self._mqtt_up():
|
||||
self._mqtt_down_since = None
|
||||
if self._failsafe:
|
||||
logger.info("MQTT back — resuming Discord radio streams.")
|
||||
self._failsafe = False
|
||||
else:
|
||||
if self._mqtt_down_since is None:
|
||||
self._mqtt_down_since = now
|
||||
elif (not self._failsafe and self._sessions
|
||||
and now - self._mqtt_down_since >= settings.discord_mqtt_loss_stop_seconds):
|
||||
logger.warning("MQTT unreachable past the fail-safe — logging every radio bot out "
|
||||
"(C2 will hand these tokens to other nodes).")
|
||||
self._failsafe = True
|
||||
await self.reconcile()
|
||||
self._ticks += 1
|
||||
# On change, and every few passes regardless: a periodic report is what
|
||||
# lets C2 settle a release whose first ack was lost.
|
||||
if self.snapshot() != before or self._ticks % REPORT_EVERY_TICKS == 0:
|
||||
await self.publish_actual()
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Reporting
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def snapshot(self) -> dict:
|
||||
streams = {}
|
||||
for sid in self._desired:
|
||||
v = self._voices.get(sid)
|
||||
if self._paused:
|
||||
streams[sid] = {"state": "paused_locally", "error": None, "code": None}
|
||||
elif v is None:
|
||||
streams[sid] = {"state": "connecting", "error": None, "code": None}
|
||||
else:
|
||||
streams[sid] = {"state": v.state, "error": v.error, "code": v.code}
|
||||
return {
|
||||
"applied_seq": self._applied_seq,
|
||||
"sessions": sorted(tid for tid, s in self._sessions.items() if s.alive()),
|
||||
"streams": streams,
|
||||
}
|
||||
|
||||
async def publish_actual(self) -> None:
|
||||
try:
|
||||
await self._publish(self.snapshot())
|
||||
except Exception as e:
|
||||
logger.warning(f"Publishing Discord actual state failed: {e}")
|
||||
|
||||
# metadata_watcher hooks — audio plays continuously, nothing to do per call.
|
||||
def start_stream(self):
|
||||
"""Called when an OP25 call starts — stream plays continuously, nothing to do."""
|
||||
pass
|
||||
|
||||
def stop_stream(self):
|
||||
"""Called when an OP25 call ends — stream plays continuously, nothing to do."""
|
||||
pass
|
||||
|
||||
async def stop(self):
|
||||
self._guild_id = None
|
||||
self._channel_id = None
|
||||
self._was_streaming = False
|
||||
|
||||
if self._watchdog_task:
|
||||
self._watchdog_task.cancel()
|
||||
self._watchdog_task = None
|
||||
|
||||
await self.leave()
|
||||
|
||||
if self._task:
|
||||
self._task.cancel()
|
||||
if self._bot:
|
||||
await self._bot.close()
|
||||
self._bot = None
|
||||
self._task = None
|
||||
self._current_token = None
|
||||
self._ready_event = None
|
||||
|
||||
def _play_stream(self):
|
||||
"""
|
||||
Feed Discord voice straight from the PulseAudio monitor.
|
||||
|
||||
Icecast is NOT used here: it lags ~1 s at connect and drifts to 100 s+,
|
||||
which is unusable for live listening. Icecast remains the frontend/mobile
|
||||
listening path only.
|
||||
"""
|
||||
if not self._voice_client:
|
||||
return
|
||||
|
||||
if not pulse.is_ready():
|
||||
logger.error(
|
||||
f"PulseAudio socket {pulse.socket_path()} missing — "
|
||||
f"cannot start Discord audio; retrying in {STREAM_RETRY_DELAY}s."
|
||||
)
|
||||
self._schedule_restart()
|
||||
return
|
||||
|
||||
# before_options land ahead of -i, so this becomes:
|
||||
# ffmpeg -f pulse -i drb_sink.monitor …
|
||||
source = discord.FFmpegPCMAudio(
|
||||
settings.pulse_source,
|
||||
before_options="-f pulse",
|
||||
)
|
||||
self._voice_client.play(
|
||||
discord.PCMVolumeTransformer(source, volume=1.0),
|
||||
after=self._on_stream_end,
|
||||
)
|
||||
|
||||
def _schedule_restart(self, delay: float = STREAM_RETRY_DELAY):
|
||||
"""Re-arm the audio source after a delay — safe to call from any thread."""
|
||||
if not self._loop:
|
||||
return
|
||||
|
||||
async def _delayed_restart():
|
||||
await asyncio.sleep(delay)
|
||||
vc = self._voice_client
|
||||
if vc and vc.is_connected() and not vc.is_playing():
|
||||
self._play_stream()
|
||||
|
||||
self._loop.call_soon_threadsafe(lambda: asyncio.ensure_future(_delayed_restart()))
|
||||
|
||||
def _on_stream_end(self, error):
|
||||
if error:
|
||||
logger.error(f"Stream ended with error: {error}")
|
||||
vc = self._voice_client
|
||||
if not (self._loop and vc and vc.is_connected() and not vc.is_playing()):
|
||||
return
|
||||
if error:
|
||||
# Back off before retrying — prevents a tight loop when PulseAudio is
|
||||
# unavailable (FFmpeg exits immediately in that case).
|
||||
self._schedule_restart()
|
||||
else:
|
||||
self._loop.call_soon_threadsafe(self._play_stream)
|
||||
|
||||
def _stop_stream(self):
|
||||
if self._voice_client and self._voice_client.is_playing():
|
||||
self._voice_client.stop()
|
||||
|
||||
async def _watchdog_loop(self):
|
||||
"""Periodically verify the voice connection is alive; rejoin silently if not."""
|
||||
await asyncio.sleep(WATCHDOG_INTERVAL) # give initial join time to settle
|
||||
while True:
|
||||
try:
|
||||
await asyncio.sleep(WATCHDOG_INTERVAL)
|
||||
|
||||
# Only act if we're supposed to be in a channel
|
||||
if not self._guild_id or not self._channel_id:
|
||||
continue
|
||||
|
||||
connected = self._voice_client is not None and self._voice_client.is_connected()
|
||||
if not connected:
|
||||
logger.warning("Watchdog: voice connection lost — attempting rejoin.")
|
||||
await asyncio.sleep(REJOIN_DELAY)
|
||||
rejoined = await self.join(
|
||||
self._guild_id,
|
||||
self._channel_id,
|
||||
self._current_token,
|
||||
)
|
||||
if rejoined:
|
||||
logger.info("Watchdog: successfully rejoined voice channel.")
|
||||
else:
|
||||
logger.error("Watchdog: rejoin failed — will retry next cycle.")
|
||||
|
||||
except asyncio.CancelledError:
|
||||
return
|
||||
except Exception as e:
|
||||
logger.error(f"Watchdog error: {e}")
|
||||
|
||||
async def _start_bot(self, token: str) -> bool:
|
||||
await self.stop() # clean up any previous instance
|
||||
|
||||
intents = discord.Intents.default()
|
||||
intents.voice_states = True
|
||||
intents.message_content = True
|
||||
intents.messages = True
|
||||
self._bot = commands.Bot(command_prefix="!", intents=intents)
|
||||
self._ready_event = asyncio.Event()
|
||||
self._current_token = token
|
||||
self._loop = asyncio.get_event_loop()
|
||||
|
||||
@self._bot.event
|
||||
async def on_ready():
|
||||
logger.info(f"Discord bot ready: {self._bot.user} ({self._bot.user.id})")
|
||||
self._ready_event.set()
|
||||
|
||||
@self._bot.event
|
||||
async def on_voice_state_update(
|
||||
member: discord.Member,
|
||||
before: discord.VoiceState,
|
||||
after: discord.VoiceState,
|
||||
):
|
||||
"""Detect when our own bot gets disconnected from a voice channel."""
|
||||
if self._bot.user and member.id != self._bot.user.id:
|
||||
return
|
||||
if before.channel is not None and after.channel is None:
|
||||
# Bot was disconnected (kicked or server drop).
|
||||
# Don't null _voice_client here — Discord.py reconnects voice
|
||||
# transparently and would leave _voice_client stale as None.
|
||||
# The watchdog detects real disconnects via is_connected().
|
||||
logger.warning("Bot was disconnected from voice channel — watchdog will rejoin if needed.")
|
||||
|
||||
@self._bot.event
|
||||
async def on_message(message: discord.Message):
|
||||
if message.author.bot:
|
||||
return
|
||||
if self._bot.user not in message.mentions:
|
||||
return
|
||||
content = message.content.lower()
|
||||
if "leave" in content:
|
||||
await self.leave()
|
||||
try:
|
||||
await message.reply("Disconnected.")
|
||||
except Exception:
|
||||
pass
|
||||
elif "joinme" in content or "join" in content:
|
||||
member = message.guild.get_member(message.author.id) if message.guild else None
|
||||
vc = member.voice.channel if member and member.voice else None
|
||||
if not vc:
|
||||
try:
|
||||
await message.reply("You're not in a voice channel.")
|
||||
except Exception:
|
||||
pass
|
||||
return
|
||||
try:
|
||||
if self._voice_client and self._voice_client.is_connected():
|
||||
await self._voice_client.move_to(vc)
|
||||
else:
|
||||
self._voice_client = await vc.connect()
|
||||
self._channel_id = vc.id
|
||||
# A fresh connect() has no audio source attached yet.
|
||||
if not self._voice_client.is_playing():
|
||||
self._play_stream()
|
||||
await message.reply(f"Joined {vc.name}.")
|
||||
except Exception as e:
|
||||
logger.error(f"joinme failed: {e}")
|
||||
|
||||
self._task = asyncio.create_task(self._bot.start(token))
|
||||
|
||||
try:
|
||||
await asyncio.wait_for(self._ready_event.wait(), timeout=BOT_READY_TIMEOUT)
|
||||
self._watchdog_task = asyncio.create_task(self._watchdog_loop())
|
||||
return True
|
||||
except asyncio.TimeoutError:
|
||||
logger.error("Timed out waiting for Discord bot to become ready.")
|
||||
await self.stop()
|
||||
return False
|
||||
|
||||
def _is_bot_running(self) -> bool:
|
||||
return (
|
||||
self._bot is not None
|
||||
and self._task is not None
|
||||
and not self._task.done()
|
||||
)
|
||||
|
||||
@property
|
||||
def is_connected(self) -> bool:
|
||||
return self._voice_client is not None and self._voice_client.is_connected()
|
||||
return any(v.vc is not None and v.vc.is_connected() for v in self._voices.values())
|
||||
|
||||
|
||||
radio_bot = RadioBot()
|
||||
radio_bot = RadioManager()
|
||||
|
||||
@@ -11,6 +11,7 @@ from app.internal import credentials, key_pickup
|
||||
CommandCallback = Callable[[Dict[str, Any]], Awaitable[None]]
|
||||
ConfigCallback = Callable[[Dict[str, Any]], Awaitable[None]]
|
||||
ApiKeyCallback = Callable[[Dict[str, Any]], Awaitable[None]]
|
||||
DesiredCallback = Callable[[Dict[str, Any]], Awaitable[None]]
|
||||
|
||||
# CONNACK codes meaning "your credentials are wrong", as paho reports them:
|
||||
# MQTT 3.1.1 return codes 4/5, and the MQTT 5 reason codes paho maps them to
|
||||
@@ -46,6 +47,7 @@ class MQTTManager:
|
||||
self.on_command: Optional[CommandCallback] = None
|
||||
self.on_config_push: Optional[ConfigCallback] = None
|
||||
self.on_api_key: Optional[ApiKeyCallback] = None
|
||||
self.on_discord_desired: Optional[DesiredCallback] = None
|
||||
|
||||
self._offline_buffer = deque(maxlen=settings.offline_call_buffer_size)
|
||||
|
||||
@@ -55,6 +57,10 @@ class MQTTManager:
|
||||
self._t_metadata = f"nodes/{nid}/metadata"
|
||||
self._t_commands = f"nodes/{nid}/commands"
|
||||
self._t_config = f"nodes/{nid}/config"
|
||||
# Discord radio desired state (retained by C2) and our actual state —
|
||||
# node-26#25. Token IDs only; the strings come over HTTPS.
|
||||
self._t_discord_desired = f"nodes/{nid}/discord/desired"
|
||||
self._t_discord_actual = f"nodes/{nid}/discord/actual"
|
||||
# TODO(mqtt-cutover): dead once enrollment lands client-side. This
|
||||
# was the pre-dynsec key-delivery path (server retain-publishes the
|
||||
# api_key here after admin approval; node asks for redelivery via
|
||||
@@ -131,6 +137,7 @@ class MQTTManager:
|
||||
self._connected = True
|
||||
client.subscribe(self._t_commands, qos=1)
|
||||
client.subscribe(self._t_config, qos=1)
|
||||
client.subscribe(self._t_discord_desired, qos=1)
|
||||
client.subscribe(self._t_api_key, qos=2) # TODO(mqtt-cutover): see _t_api_key comment above
|
||||
client.subscribe(self._t_discovery, qos=0)
|
||||
logger.info("MQTT connected.")
|
||||
@@ -159,6 +166,8 @@ class MQTTManager:
|
||||
asyncio.run_coroutine_threadsafe(self.on_command(payload), self._loop)
|
||||
elif msg.topic == self._t_config and self.on_config_push:
|
||||
asyncio.run_coroutine_threadsafe(self.on_config_push(payload), self._loop)
|
||||
elif msg.topic == self._t_discord_desired and self.on_discord_desired:
|
||||
asyncio.run_coroutine_threadsafe(self.on_discord_desired(payload), self._loop)
|
||||
elif msg.topic == self._t_api_key and self.on_api_key:
|
||||
asyncio.run_coroutine_threadsafe(self.on_api_key(payload), self._loop)
|
||||
elif msg.topic == self._t_discovery:
|
||||
@@ -285,6 +294,12 @@ class MQTTManager:
|
||||
else:
|
||||
self._publish(self._t_metadata, payload, qos=1)
|
||||
|
||||
def publish_discord_actual(self, payload: dict) -> None:
|
||||
"""What the Discord radio manager is actually doing. Not buffered while
|
||||
offline: a fresh report is sent after every change and periodically,
|
||||
and only the newest one means anything."""
|
||||
self._publish(self._t_discord_actual, payload, qos=1)
|
||||
|
||||
async def _flush_offline_buffer(self):
|
||||
if not self._offline_buffer:
|
||||
return
|
||||
|
||||
@@ -144,20 +144,11 @@ async def on_command(payload: dict):
|
||||
action = payload.get("action")
|
||||
logger.info(f"Command received: {action}")
|
||||
|
||||
if action == "discord_join":
|
||||
token = payload.get("token")
|
||||
if not token:
|
||||
logger.error("discord_join command missing token — ignoring.")
|
||||
return
|
||||
await radio_bot.join(
|
||||
guild_id=int(payload["guild_id"]),
|
||||
channel_id=int(payload["channel_id"]),
|
||||
token=token,
|
||||
call_active=metadata_watcher.is_active,
|
||||
system_name=payload.get("system_name"),
|
||||
)
|
||||
elif action == "discord_leave":
|
||||
await radio_bot.leave()
|
||||
if action in ("discord_join", "discord_leave"):
|
||||
# Pre-desired-state C2. Discord streams now arrive on
|
||||
# nodes/{id}/discord/desired (node-26#25); a raw-token join is refused
|
||||
# so a bot token is never taken from an MQTT command again.
|
||||
logger.warning(f"Ignoring legacy {action} command — Discord radio is driven by desired state now.")
|
||||
elif action in ("set_sdr_config", "set_secondary_priority"):
|
||||
from app.internal.sdr_settings import set_sdr_settings
|
||||
try:
|
||||
@@ -300,8 +291,9 @@ async def lifespan(app: FastAPI):
|
||||
mqtt_manager.on_command = on_command
|
||||
mqtt_manager.on_config_push = on_config_push
|
||||
mqtt_manager.on_api_key = on_api_key
|
||||
mqtt_manager.on_discord_desired = radio_bot.apply_desired
|
||||
|
||||
# Start services (radio_bot starts on-demand when a discord_join command arrives)
|
||||
# Start services (radio_bot starts when C2's retained discord/desired arrives)
|
||||
await mqtt_manager.connect()
|
||||
await metadata_watcher.start()
|
||||
await call_recorder.start() # persistent PulseAudio ring buffer
|
||||
|
||||
@@ -400,15 +400,25 @@ async def set_sdr(body: SdrSettingsBody):
|
||||
return {"ok": True, "running": running}
|
||||
|
||||
|
||||
@router.get("/discord")
|
||||
async def discord_status():
|
||||
"""What the node's radio bots are doing (desired state from C2, actual here)."""
|
||||
return radio_bot.snapshot()
|
||||
|
||||
|
||||
@router.post("/discord/join")
|
||||
async def discord_join(guild_id: int, channel_id: int):
|
||||
ok = await radio_bot.join(guild_id, channel_id)
|
||||
if not ok:
|
||||
raise HTTPException(500, "Failed to join voice channel")
|
||||
return {"ok": True}
|
||||
async def discord_join():
|
||||
# node-26#25 item 4: this used to call join() without a token and 500 every
|
||||
# time. A node cannot lease a radio bot by itself — C2 owns the pool.
|
||||
raise HTTPException(
|
||||
409,
|
||||
"Discord radio channels are started from Discord (/join) or the DRB web app; "
|
||||
"the node cannot pick a radio bot on its own.",
|
||||
)
|
||||
|
||||
|
||||
@router.post("/discord/leave")
|
||||
async def discord_leave():
|
||||
await radio_bot.leave()
|
||||
"""Local stop: log every radio bot out until DRB sends a new channel list."""
|
||||
await radio_bot.pause_locally()
|
||||
return {"ok": True}
|
||||
|
||||
@@ -0,0 +1,260 @@
|
||||
"""
|
||||
Edge Discord radio manager (node-26#25): desired-state reconcile, logout on
|
||||
release, watchdog that survives a dead session, MQTT-loss fail-safe.
|
||||
|
||||
Discord itself is faked: FakeClient logs "in" by firing on_ready, guilds and
|
||||
channels are plain objects, voice connections record play/disconnect.
|
||||
"""
|
||||
import asyncio
|
||||
|
||||
import discord
|
||||
import pytest
|
||||
|
||||
from app.internal import discord_radio
|
||||
from app.internal.discord_radio import RadioManager
|
||||
|
||||
|
||||
class FakeVC:
|
||||
def __init__(self, channel):
|
||||
self.channel = channel
|
||||
self.connected = True
|
||||
self.playing = False
|
||||
self.plays = 0
|
||||
|
||||
def is_connected(self):
|
||||
return self.connected
|
||||
|
||||
def is_playing(self):
|
||||
return self.playing
|
||||
|
||||
def play(self, source, after=None):
|
||||
self.playing = True
|
||||
self.plays += 1
|
||||
|
||||
def stop(self):
|
||||
self.playing = False
|
||||
|
||||
async def disconnect(self, force=False):
|
||||
self.connected = False
|
||||
self.playing = False
|
||||
self.channel.guild.voice_client = None
|
||||
|
||||
async def move_to(self, channel):
|
||||
self.channel = channel
|
||||
channel.guild.voice_client = self
|
||||
|
||||
|
||||
class FakeChannel:
|
||||
def __init__(self, guild, cid, kind=discord.ChannelType.voice):
|
||||
self.guild, self.id, self.type = guild, cid, kind
|
||||
|
||||
async def connect(self):
|
||||
vc = FakeVC(self)
|
||||
self.guild.voice_client = vc
|
||||
return vc
|
||||
|
||||
|
||||
class FakeGuild:
|
||||
def __init__(self, gid, channel_ids):
|
||||
self.id = gid
|
||||
self.voice_client = None
|
||||
self.channels = {c: FakeChannel(self, c) for c in channel_ids}
|
||||
|
||||
def get_channel(self, cid):
|
||||
return self.channels.get(cid)
|
||||
|
||||
|
||||
class FakeClient:
|
||||
instances: list = []
|
||||
|
||||
def __init__(self, guilds, fail_login=False):
|
||||
self.guilds = guilds
|
||||
self.fail_login = fail_login
|
||||
self.closed = False
|
||||
self.handlers = {}
|
||||
self.token = None
|
||||
self.user = "radio-bot"
|
||||
FakeClient.instances.append(self)
|
||||
|
||||
def event(self, coro):
|
||||
self.handlers[coro.__name__] = coro
|
||||
return coro
|
||||
|
||||
async def start(self, token):
|
||||
self.token = token
|
||||
if self.fail_login:
|
||||
raise discord.LoginFailure("Improper token has been passed.")
|
||||
await self.handlers["on_ready"]()
|
||||
while not self.closed:
|
||||
await asyncio.sleep(0.01)
|
||||
|
||||
async def close(self):
|
||||
self.closed = True
|
||||
|
||||
def is_closed(self):
|
||||
return self.closed
|
||||
|
||||
def get_guild(self, gid):
|
||||
return self.guilds.get(gid)
|
||||
|
||||
async def change_presence(self, activity=None):
|
||||
self.presence = activity
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def env(monkeypatch):
|
||||
FakeClient.instances = []
|
||||
guilds = {1: FakeGuild(1, [11, 12]), 2: FakeGuild(2, [21])}
|
||||
published = []
|
||||
fetched = []
|
||||
state = {"mqtt": True, "fail_login": False}
|
||||
|
||||
async def fetch(token_id):
|
||||
fetched.append(token_id)
|
||||
return f"secret-{token_id}"
|
||||
|
||||
async def publish(payload):
|
||||
published.append(payload)
|
||||
|
||||
async def pulse_ready(timeout=None):
|
||||
return True
|
||||
monkeypatch.setattr(discord_radio.pulse, "wait_until_ready", pulse_ready)
|
||||
monkeypatch.setattr(discord_radio.pulse, "is_ready", lambda: True)
|
||||
|
||||
mgr = RadioManager(
|
||||
client_factory=lambda: FakeClient(guilds, fail_login=state["fail_login"]),
|
||||
token_fetcher=fetch, publisher=publish, audio_factory=lambda: object(),
|
||||
mqtt_up=lambda: state["mqtt"], ready_timeout=1,
|
||||
)
|
||||
yield type("Env", (), {"mgr": mgr, "guilds": guilds, "published": published,
|
||||
"fetched": fetched, "state": state})
|
||||
# Leave no background tasks behind.
|
||||
if mgr._watchdog:
|
||||
mgr._watchdog.cancel()
|
||||
|
||||
|
||||
def _desired(seq, *streams):
|
||||
return {"seq": seq, "streams": [
|
||||
{"stream_id": f"{g}:{c}", "guild_id": str(g), "channel_id": str(c), "token_id": t, "system_name": "County"}
|
||||
for g, c, t in streams
|
||||
]}
|
||||
|
||||
|
||||
async def test_apply_connects_and_reports(env):
|
||||
await env.mgr.apply_desired(_desired(1, (1, 11, "t1")))
|
||||
actual = env.published[-1]
|
||||
assert actual["applied_seq"] == 1
|
||||
assert actual["sessions"] == ["t1"]
|
||||
assert actual["streams"] == {"1:11": {"state": "connected", "error": None, "code": None}}
|
||||
assert env.mgr.is_connected
|
||||
vc = env.guilds[1].voice_client
|
||||
assert vc.plays == 1 and vc.channel.id == 11
|
||||
|
||||
|
||||
async def test_one_session_serves_two_guilds(env):
|
||||
await env.mgr.apply_desired(_desired(1, (1, 11, "t1"), (2, 21, "t1")))
|
||||
assert len(FakeClient.instances) == 1
|
||||
assert env.published[-1]["sessions"] == ["t1"]
|
||||
assert env.guilds[1].voice_client.is_connected() and env.guilds[2].voice_client.is_connected()
|
||||
|
||||
|
||||
async def test_dropping_the_last_stream_logs_the_bot_out(env):
|
||||
"""node-26#25 item 1: leave() used to keep the gateway session alive."""
|
||||
await env.mgr.apply_desired(_desired(1, (1, 11, "t1"), (2, 21, "t1")))
|
||||
client = FakeClient.instances[0]
|
||||
await env.mgr.apply_desired(_desired(2, (2, 21, "t1")))
|
||||
assert not client.closed # still needed for guild 2
|
||||
assert env.guilds[1].voice_client is None
|
||||
await env.mgr.apply_desired(_desired(3))
|
||||
assert client.closed # logged OUT, not just out of voice
|
||||
assert env.published[-1] == {"applied_seq": 3, "sessions": [], "streams": {}}
|
||||
|
||||
|
||||
async def test_stale_desired_is_ignored(env):
|
||||
await env.mgr.apply_desired(_desired(5, (1, 11, "t1")))
|
||||
await env.mgr.apply_desired(_desired(4))
|
||||
assert env.mgr.snapshot()["applied_seq"] == 5 and env.mgr.is_connected
|
||||
|
||||
|
||||
async def test_dead_session_is_recovered_by_the_watchdog(env):
|
||||
"""node-26#25 item 2: the old watchdog cancelled itself on rejoin."""
|
||||
await env.mgr.apply_desired(_desired(1, (1, 11, "t1")))
|
||||
first = FakeClient.instances[0]
|
||||
first.closed = True # bot task dies
|
||||
await asyncio.sleep(0.05)
|
||||
env.guilds[1].voice_client.connected = False
|
||||
await env.mgr.tick()
|
||||
assert len(FakeClient.instances) == 2 and not FakeClient.instances[1].closed
|
||||
assert env.mgr.snapshot()["streams"]["1:11"]["state"] == "connected"
|
||||
# The watchdog task itself is still alive after the recovery.
|
||||
assert env.mgr._watchdog is not None and not env.mgr._watchdog.done()
|
||||
|
||||
|
||||
async def test_dropped_voice_is_reconnected(env):
|
||||
await env.mgr.apply_desired(_desired(1, (1, 11, "t1")))
|
||||
env.guilds[1].voice_client.connected = False
|
||||
env.guilds[1].voice_client = None
|
||||
await env.mgr.tick()
|
||||
assert env.guilds[1].voice_client.is_connected()
|
||||
|
||||
|
||||
async def test_bot_not_in_guild_is_reported_not_swallowed(env):
|
||||
await env.mgr.apply_desired(_desired(1, (9, 91, "t1")))
|
||||
st = env.published[-1]["streams"]["9:91"]
|
||||
assert st["state"] == "error" and st["code"] == "not_in_guild"
|
||||
|
||||
|
||||
async def test_login_failure_is_reported(env):
|
||||
env.state["fail_login"] = True
|
||||
await env.mgr.apply_desired(_desired(1, (1, 11, "t1")))
|
||||
st = env.published[-1]["streams"]["1:11"]
|
||||
assert st["state"] == "error" and st["code"] == "login_failed"
|
||||
assert env.published[-1]["sessions"] == []
|
||||
|
||||
|
||||
async def test_mqtt_loss_failsafe_logs_out_then_resumes(env, monkeypatch):
|
||||
monkeypatch.setattr(discord_radio.settings, "discord_mqtt_loss_stop_seconds", 300)
|
||||
await env.mgr.apply_desired(_desired(1, (1, 11, "t1")))
|
||||
client = FakeClient.instances[0]
|
||||
env.state["mqtt"] = False
|
||||
await env.mgr.tick(now=1000.0)
|
||||
assert not client.closed # inside the window
|
||||
await env.mgr.tick(now=1299.0)
|
||||
assert not client.closed
|
||||
await env.mgr.tick(now=1300.0)
|
||||
assert client.closed and env.mgr.snapshot()["sessions"] == []
|
||||
# Token strings are not kept across a fail-safe logout: re-login refetches
|
||||
# from C2, which refuses if the lease has moved on.
|
||||
assert env.mgr._tokens == {}
|
||||
env.state["mqtt"] = True
|
||||
before = len(env.fetched)
|
||||
await env.mgr.tick(now=1400.0)
|
||||
assert len(env.fetched) == before + 1
|
||||
assert env.mgr.snapshot()["streams"]["1:11"]["state"] == "connected"
|
||||
|
||||
|
||||
async def test_token_fetch_refused_reports_error(env):
|
||||
async def refuse(token_id):
|
||||
return None
|
||||
env.mgr._fetch_token = refuse
|
||||
await env.mgr.apply_desired(_desired(1, (1, 11, "t1")))
|
||||
st = env.published[-1]["streams"]["1:11"]
|
||||
assert st["state"] == "error" and FakeClient.instances == []
|
||||
|
||||
|
||||
async def test_local_pause_logs_out_until_next_desired(env):
|
||||
await env.mgr.apply_desired(_desired(1, (1, 11, "t1")))
|
||||
await env.mgr.pause_locally()
|
||||
assert FakeClient.instances[0].closed
|
||||
assert env.published[-1]["streams"]["1:11"]["state"] == "paused_locally"
|
||||
await env.mgr.apply_desired(_desired(2, (1, 11, "t1")))
|
||||
assert env.mgr.is_connected
|
||||
|
||||
|
||||
async def test_no_message_content_intent_on_real_client():
|
||||
client = discord_radio._default_client_factory()
|
||||
try:
|
||||
assert client.intents.message_content is False
|
||||
assert client.intents.voice_states and client.intents.guilds
|
||||
finally:
|
||||
await client.close()
|
||||
@@ -24,21 +24,37 @@ pytestmark = pytest.mark.skipif(
|
||||
)
|
||||
|
||||
REPO = "https://git.example.net/logan/node-26.git"
|
||||
OLD, NEW = "a" * 40, "b" * 40
|
||||
OLD, NEW, TREE = "a" * 40, "b" * 40, "c" * 40
|
||||
REPO_ROOT = SCRIPT.parents[1]
|
||||
|
||||
STUB_GIT = r'''#!/usr/bin/env bash
|
||||
echo "git $*" >> "$STUB/calls"
|
||||
while [ "$1" = -c ]; do shift 2; done
|
||||
while [ "$1" = -c ] || [ "$1" = -C ]; do shift 2; done
|
||||
case "$1" in
|
||||
rev-parse) if [ "$2" = HEAD ]; then cat "$STUB/head"; else echo "$STUB_NEW"; fi ;;
|
||||
rev-parse)
|
||||
if [ "$2" = HEAD ]; then cat "$STUB/head"
|
||||
elif [ "$2" = --verify ]; then echo "$STUB_TREE" # tree hash of <commit>:<dir>
|
||||
else echo "$STUB_NEW"; fi ;;
|
||||
remote) echo "$STUB_ORIGIN" ;;
|
||||
fetch) exit 0 ;;
|
||||
show-ref) [ "$4" = "refs/tags/v1" ] ;;
|
||||
cat-file) [ -z "${STUB_NO_PIN:-}" ] ;; # target has scripts/image-tags.sh
|
||||
show) case "$2" in
|
||||
*:scripts/image-tags.sh) cat "$STUB_REPO/scripts/image-tags.sh" ;;
|
||||
*:docker-compose.yml) cat "$STUB_REPO/docker-compose.yml" ;;
|
||||
esac ;;
|
||||
checkout) echo "${@: -1}" > "$STUB/head" ;;
|
||||
*) exit 0 ;;
|
||||
esac
|
||||
'''
|
||||
STUB_DOCKER = '#!/usr/bin/env bash\necho "docker $*" >> "$STUB/calls"\n'
|
||||
STUB_DOCKER = r'''#!/usr/bin/env bash
|
||||
echo "docker $*" >> "$STUB/calls"
|
||||
case "$*" in
|
||||
*"config --images"*) echo "reg.example/edge-node:$STUB_PULLED_TAG" ;;
|
||||
"image inspect"*) [ -z "${STUB_MISSING_IMAGE:-}" ] || exit 1 ;;
|
||||
esac
|
||||
exit 0
|
||||
'''
|
||||
STUB_CURL = r'''#!/usr/bin/env bash
|
||||
echo "curl $*" >> "$STUB/calls"
|
||||
for a in "$@"; do case "$a" in @*) cat "${a#@}" >> "$STUB/headers" ;; esac; done
|
||||
@@ -79,6 +95,7 @@ def env(tmp_path):
|
||||
e.update({
|
||||
"PATH": f"{bindir}:{os.environ['PATH']}",
|
||||
"STUB": str(stub), "STUB_NEW": NEW, "STUB_ORIGIN": REPO,
|
||||
"STUB_TREE": TREE, "STUB_REPO": str(REPO_ROOT), "STUB_PULLED_TAG": f"src-{TREE}",
|
||||
"DRB_UPDATER_CONF": str(conf),
|
||||
"DRB_UPDATER_STATE": str(tmp_path / "state"),
|
||||
"DRB_UPDATER_LOCK": str(tmp_path / "lock"),
|
||||
@@ -94,6 +111,10 @@ def _calls(env):
|
||||
return (env["stub"] / "calls").read_text()
|
||||
|
||||
|
||||
def _pulled(calls: str) -> bool:
|
||||
return any(line.startswith("docker compose ") and line.endswith(" pull") for line in calls.splitlines())
|
||||
|
||||
|
||||
def test_no_request_means_checkin_only(env):
|
||||
r = _run(env)
|
||||
assert r.returncode == 0, r.stderr
|
||||
@@ -108,7 +129,7 @@ def test_c2_request_updates_once_and_reports(env):
|
||||
r = _run(env)
|
||||
assert r.returncode == 0, r.stderr + r.stdout
|
||||
calls = _calls(env)
|
||||
assert "docker compose pull" in calls
|
||||
assert _pulled(calls)
|
||||
assert "docker compose up -d --no-build --force-recreate --remove-orphans" in calls
|
||||
assert (env["stub"] / "head").read_text().strip() == NEW
|
||||
assert '"applied_request_at":"2026-10-04T12:00:00+00:00"' in calls
|
||||
@@ -126,7 +147,7 @@ def test_mqtt_sentinel_triggers_update_and_is_consumed(env):
|
||||
assert r.returncode == 0, r.stderr
|
||||
assert not sentinel.exists()
|
||||
calls = _calls(env)
|
||||
assert "docker compose pull" in calls
|
||||
assert _pulled(calls)
|
||||
assert "evil" not in calls and "rm -rf" not in calls
|
||||
|
||||
|
||||
@@ -153,7 +174,7 @@ def test_refuses_when_env_registry_differs(env):
|
||||
(env["inst"] / "configs" / "update-requested").write_text("")
|
||||
r = _run(env)
|
||||
assert r.returncode != 0
|
||||
assert "docker compose pull" not in _calls(env)
|
||||
assert not _pulled(_calls(env))
|
||||
|
||||
|
||||
def test_non_https_repo_in_conf_is_rejected(env):
|
||||
@@ -177,3 +198,42 @@ def test_plain_http_c2_skips_checkin(env):
|
||||
r = _run(env)
|
||||
assert r.returncode == 0
|
||||
assert "curl" not in _calls(env)
|
||||
|
||||
|
||||
# ── node-26#19: images pinned to the target's source tree ─────────────────
|
||||
|
||||
def test_pins_images_and_pulls_before_checkout(env):
|
||||
(env["inst"] / "configs" / "update-requested").write_text("")
|
||||
r = _run(env)
|
||||
assert r.returncode == 0, r.stderr + r.stdout
|
||||
lines = _calls(env).splitlines()
|
||||
pull = next(i for i, ln in enumerate(lines) if ln.startswith("docker compose ") and ln.endswith(" pull"))
|
||||
checkout = next(i for i, ln in enumerate(lines) if ln.startswith("git") and " checkout " in ln)
|
||||
assert pull < checkout, "images must be pulled before the new code is checked out"
|
||||
dotenv = (env["inst"] / ".env").read_text()
|
||||
for var in ("DRB_TAG_EDGE_NODE", "DRB_TAG_OP25", "DRB_TAG_ICECAST", "DRB_TAG_SECONDARY_SDR"):
|
||||
assert f"{var}=src-{TREE}\n" in dotenv
|
||||
assert dotenv.startswith("NODE_ID=node-7\n") # existing lines kept
|
||||
assert not (env["inst"] / ".drb-update-compose.yml").exists()
|
||||
|
||||
|
||||
def test_missing_pinned_image_changes_nothing(env):
|
||||
env["env"]["STUB_MISSING_IMAGE"] = "1"
|
||||
(env["inst"] / "configs" / "update-requested").write_text("")
|
||||
r = _run(env)
|
||||
assert r.returncode != 0
|
||||
assert (env["stub"] / "head").read_text().strip() == OLD
|
||||
calls = _calls(env)
|
||||
assert " checkout " not in calls
|
||||
assert "up -d" not in calls
|
||||
assert "DRB_TAG_" not in (env["inst"] / ".env").read_text()
|
||||
|
||||
|
||||
def test_target_without_pinning_still_updates(env):
|
||||
env["env"]["STUB_NO_PIN"] = "1"
|
||||
(env["inst"] / "configs" / "update-requested").write_text("")
|
||||
r = _run(env)
|
||||
assert r.returncode == 0, r.stderr + r.stdout
|
||||
assert (env["stub"] / "head").read_text().strip() == NEW
|
||||
assert _pulled(_calls(env))
|
||||
assert "DRB_TAG_" not in (env["inst"] / ".env").read_text()
|
||||
|
||||
+30
-8
@@ -15,7 +15,7 @@
|
||||
# --track-main for the owner's own always-latest nodes)
|
||||
#
|
||||
# Hosting, the pinned ref, and setup.sh's retirement are settled (owner
|
||||
# decisions, 2026-09-06). One standing hazard remains — D3 below.
|
||||
# decisions, 2026-09-06). D3 below is mitigated by image pinning (node-26#19).
|
||||
#
|
||||
# What it does, in order:
|
||||
# 1. Preflight (root, arch, apt, SDR present)
|
||||
@@ -53,15 +53,18 @@
|
||||
# the mint panel / node-26#4). DEFAULT_REF below is `v1`. `--track-main`
|
||||
# stays as an opt-in for the owner's own nodes.
|
||||
#
|
||||
# STANDING HAZARD
|
||||
# MITIGATED HAZARD
|
||||
#
|
||||
# D3 SOURCE-OVERLAY. docker-compose.yml bind-mounts ./drb-edge-node/app and
|
||||
# ./op25-container/app OVER the image's /app. So even in prebuilt mode
|
||||
# the running Python is the CLONED REF's code against the pulled image's
|
||||
# dependencies. `v1` == the commit CI built the current :latest/:stable
|
||||
# from, so today they match — but the moment `v1` and the image tags
|
||||
# diverge this silently mixes them. Fix: drop those two mounts from the
|
||||
# prod compose path, or always retag images from the same ref as `v1`.
|
||||
# ./op25-container/app OVER the image's /app, so the running Python is
|
||||
# the CLONED REF's code against the pulled image's dependencies. Those
|
||||
# used to be :latest/:stable, built from main — silently a different
|
||||
# commit from `v1`. Since node-26#19 every image is pinned to
|
||||
# src-<git tree hash of its own directory at the checked-out ref>
|
||||
# (scripts/image-tags.sh, written into .env as DRB_TAG_*), which CI tags
|
||||
# identically. The mounts now overlay the same code the image was built
|
||||
# from. A pinned image that was never built fails the pull loudly; the
|
||||
# "Release images" workflow builds any a release tag is missing.
|
||||
# ---------------------------------------------------------------------------
|
||||
set -euo pipefail
|
||||
|
||||
@@ -369,6 +372,20 @@ EOF
|
||||
ok ".env written for '$NODE_ID'"
|
||||
fi
|
||||
|
||||
# Pin every image to the source just checked out (node-26#19, D3). Written
|
||||
# into .env — fresh or preserved — so later manual `docker compose` runs on
|
||||
# this node use the same pins.
|
||||
# A --ref older than #19 has no helper; its compose file hardcodes its tags.
|
||||
if [ -f "$INSTALL_DIR/scripts/image-tags.sh" ]; then
|
||||
# shellcheck source=scripts/image-tags.sh
|
||||
. "$INSTALL_DIR/scripts/image-tags.sh"
|
||||
drb_pin_image_tags "$INSTALL_DIR/.env" "$INSTALL_DIR" "$RESOLVED_SHA" || die "could not write image pins to .env"
|
||||
ok "images pinned to the source at ${RESOLVED_SHA:0:12} (DRB_TAG_* in .env)"
|
||||
else
|
||||
warn "$REF predates image pinning (node-26#19) — using its compose file's default image tags"
|
||||
drb_images_present() { return 0; }
|
||||
fi
|
||||
|
||||
# ── 5. Enrollment ───────────────────────────────────────────────────────────
|
||||
# Client half of Server/drb-c2-core/app/routers/enrollment.py. It does NOT
|
||||
# exist in the edge-node app today (mqtt_manager.py:73-85 says so explicitly),
|
||||
@@ -606,6 +623,11 @@ if [ "$DO_START" = 1 ]; then
|
||||
requirement or a transient network error: re-run with DRB_REGISTRY_USER /
|
||||
DRB_REGISTRY_PASS set, or with --build to compile on the Pi (~1h for op25)."
|
||||
fi
|
||||
# compose pull only WARNS for a missing image that also has a build:
|
||||
# section, so check the pinned images really arrived.
|
||||
drb_images_present || die "the images pinned to $REF (${RESOLVED_SHA:0:12}) are not in
|
||||
$REGISTRY yet. CI may still be building them — wait and re-run; the owner can
|
||||
run the 'Release images' workflow on that ref; or re-run with --build."
|
||||
docker compose up --no-build -d
|
||||
fi
|
||||
# ALWAYS recreate edge-node (node-26#14): `up -d` leaves an already-running
|
||||
|
||||
+60
-8
@@ -17,7 +17,9 @@
|
||||
# * Node credentials go to curl via a 0600 header file, never argv.
|
||||
# * C2 must be https.
|
||||
# What this does NOT defend against: a compromised git host or registry —
|
||||
# refs and images are not signed (see the issue filed with node-26#15).
|
||||
# refs and images are not signed (node-26#17).
|
||||
# * Images are pinned to the checked-out source (node-26#19): each one is
|
||||
# pulled as src-<git tree hash of its build directory>, never :latest.
|
||||
#
|
||||
# The whole body is inside main(), called on the last line: bash parses the
|
||||
# function completely before running it, so `git checkout` replacing this
|
||||
@@ -146,6 +148,47 @@ do_update() {
|
||||
fi
|
||||
local old new
|
||||
old="$(git rev-parse HEAD)"; new="$(git rev-parse "$target^{commit}")"
|
||||
|
||||
local env_registry
|
||||
env_registry="$(grep -E '^IMAGE_REGISTRY=' .env | head -1 | cut -d= -f2- | tr -d '"' || true)"
|
||||
[ "${env_registry:-git.vpn.cusano.net}" = "$REGISTRY" ] \
|
||||
|| { err ".env IMAGE_REGISTRY '$env_registry' != configured '$REGISTRY' — refusing"; return 1; }
|
||||
|
||||
# Image pins for the TARGET commit (node-26#19): each image is tagged with
|
||||
# the git tree hash of its own source directory, so the images always match
|
||||
# the code being checked out. The helper is read from the target commit so
|
||||
# its service list matches the target's compose file.
|
||||
# Read into a variable, not `source <(...)`: a failed `git show` must leave
|
||||
# pin=0 rather than "succeed" with no functions defined.
|
||||
local pin=0 helper
|
||||
if git cat-file -e "$new:scripts/image-tags.sh" 2>/dev/null; then
|
||||
helper="$(git show "$new:scripts/image-tags.sh")" && [ -n "$helper" ] && eval "$helper" \
|
||||
&& declare -F drb_image_tags drb_images_present drb_pin_image_tags >/dev/null && pin=1
|
||||
fi
|
||||
[ "$pin" = 1 ] || log "target ${new:0:12} predates image pinning — using its compose file's default tags"
|
||||
|
||||
if [ "$BUILD" != 1 ]; then
|
||||
# Pull BEFORE checkout. The app code is bind-mounted from the checkout
|
||||
# (install.sh D3) and edge-node runs uvicorn --reload, so checking out first
|
||||
# would run the new code on the old images for as long as the pull takes —
|
||||
# or for good, if the pinned images don't exist. Pulling against the
|
||||
# target's own compose file means a failed pull changes nothing. It is
|
||||
# written next to the real one so its relative paths (.env, env_file,
|
||||
# build contexts) resolve exactly the same way.
|
||||
local tfile="$INSTALL_DIR/.drb-update-compose.yml" pulled=0
|
||||
git show "$new:docker-compose.yml" > "$tfile" || { rm -f "$tfile"; return 1; }
|
||||
local -a tc=(-f "$tfile" --project-directory "$INSTALL_DIR")
|
||||
(
|
||||
if [ "$pin" = 1 ]; then
|
||||
while IFS= read -r l; do export "$l"; done < <(drb_image_tags "$INSTALL_DIR" "$new")
|
||||
fi
|
||||
docker compose "${tc[@]}" pull || exit 1
|
||||
if [ "$pin" = 1 ]; then drb_images_present "${tc[@]}" || exit 1; fi
|
||||
) && pulled=1
|
||||
rm -f "$tfile"
|
||||
[ "$pulled" = 1 ] || { err "images for ${new:0:12} are not available — nothing changed (CI may still be building them; the 'Release images' workflow publishes any that are missing)"; return 1; }
|
||||
fi
|
||||
|
||||
if [ "$old" != "$new" ]; then
|
||||
git -c advice.detachedHead=false checkout --force --detach "$new" || return 1
|
||||
log "checked out $REF: ${old:0:12} -> ${new:0:12}"
|
||||
@@ -153,19 +196,28 @@ do_update() {
|
||||
bash scripts/install-updater.sh --refresh || log "unit refresh failed (continuing with the old units)"
|
||||
fi
|
||||
|
||||
local env_registry
|
||||
env_registry="$(grep -E '^IMAGE_REGISTRY=' .env | head -1 | cut -d= -f2- | tr -d '"' || true)"
|
||||
[ "${env_registry:-git.vpn.cusano.net}" = "$REGISTRY" ] \
|
||||
|| { err ".env IMAGE_REGISTRY '$env_registry' != configured '$REGISTRY' — refusing"; return 1; }
|
||||
if [ "$pin" = 1 ]; then
|
||||
drb_pin_image_tags "$INSTALL_DIR/.env" "$INSTALL_DIR" "$new" || { err "could not write image pins to .env"; return 1; }
|
||||
fi
|
||||
|
||||
if [ "$BUILD" = 1 ]; then
|
||||
docker compose build || return 1
|
||||
else
|
||||
docker compose pull || return 1
|
||||
docker compose build || return 1 # tags the local builds with the same pins
|
||||
fi
|
||||
# --force-recreate: app code is bind-mounted from the checkout (install.sh
|
||||
# D3), so a code-only change also needs every container restarted.
|
||||
docker compose up -d --no-build --force-recreate --remove-orphans || return 1
|
||||
|
||||
# Drop pinned images this node no longer runs. `image prune` only removes
|
||||
# dangling images, and every previous src- tag is still tagged. `image rm`
|
||||
# without -f refuses anything a container still uses.
|
||||
if [ "$pin" = 1 ]; then
|
||||
local keep ref
|
||||
keep="$(drb_image_tags "$INSTALL_DIR" "$new" | cut -d= -f2)"
|
||||
docker images --format '{{.Repository}}:{{.Tag}}' | grep -E ':src-[0-9a-f]{40}$' \
|
||||
| while IFS= read -r ref; do
|
||||
grep -qxF "${ref##*:}" <<< "$keep" || docker image rm "$ref" >/dev/null 2>&1 || true
|
||||
done || true
|
||||
fi
|
||||
docker image prune -f >/dev/null 2>&1 || true
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,65 @@
|
||||
# shellcheck shell=bash
|
||||
#
|
||||
# Pin each node image to the exact source it was built from (node-26#19).
|
||||
# Sourced by install.sh and scripts/drb-update.sh; not run directly.
|
||||
#
|
||||
# The tag is the git TREE hash of the image's build context, `src-<tree>`.
|
||||
# CI computes the same value (`git rev-parse HEAD:<dir>` in the build
|
||||
# workflows), so a node asks for the image built from the very directory it
|
||||
# has checked out. That is exact even though each image is only rebuilt when
|
||||
# its own directory changes: an unchanged directory keeps the same tree hash,
|
||||
# and so the same image, across commits. :latest / :stable still exist for
|
||||
# local dev; a provisioned node never uses them.
|
||||
#
|
||||
# docker-compose.yml reads these as ${DRB_TAG_*:-latest}. The values live in
|
||||
# .env so every manual `docker compose ...` on the node sees the same pins.
|
||||
|
||||
# VAR:build-context-dir, one per compose service with an image.
|
||||
DRB_IMAGE_SOURCES="DRB_TAG_ICECAST:icecast DRB_TAG_OP25:op25-container \
|
||||
DRB_TAG_SECONDARY_SDR:secondary-sdr-container DRB_TAG_EDGE_NODE:drb-edge-node"
|
||||
|
||||
# drb_image_tags REPO_DIR COMMIT -> "VAR=src-<tree>" lines. A directory that
|
||||
# does not exist at COMMIT is skipped (its compose file won't reference it).
|
||||
drb_image_tags() {
|
||||
local _p _var _dir _tree
|
||||
for _p in $DRB_IMAGE_SOURCES; do
|
||||
_var="${_p%%:*}"; _dir="${_p#*:}"
|
||||
_tree="$(git -C "$1" rev-parse --verify -q "$2:$_dir" 2>/dev/null)" || continue
|
||||
printf '%s=src-%s\n' "$_var" "$_tree"
|
||||
done
|
||||
}
|
||||
|
||||
# drb_env_set ENV_FILE KEY VALUE — replace KEY's line in place, or append it.
|
||||
# Rewrites through the same inode so .env keeps its 0600 mode and owner.
|
||||
drb_env_set() {
|
||||
local _tmp; _tmp="$(mktemp)"
|
||||
awk -v k="$2" -v v="$3" '
|
||||
index($0, k "=") == 1 { if (!done) print k "=" v; done = 1; next }
|
||||
{ print }
|
||||
END { if (!done) print k "=" v }' "$1" > "$_tmp" && cat "$_tmp" > "$1"
|
||||
local _rc=$?; rm -f "$_tmp"; return $_rc
|
||||
}
|
||||
|
||||
# drb_pin_image_tags ENV_FILE REPO_DIR COMMIT — write the pins into .env and
|
||||
# export them, so compose sees them whether or not it re-reads .env.
|
||||
drb_pin_image_tags() {
|
||||
local _line
|
||||
while IFS= read -r _line; do
|
||||
drb_env_set "$1" "${_line%%=*}" "${_line#*=}" || return 1
|
||||
export "$_line"
|
||||
done < <(drb_image_tags "$2" "$3")
|
||||
}
|
||||
|
||||
# drb_images_present COMPOSE_ARGS... — 0 if every image the compose project
|
||||
# names is in the local image store. `compose pull` only WARNS when a service
|
||||
# that also has a build: section fails to pull, so its exit code alone does
|
||||
# not prove the pinned images exist.
|
||||
drb_images_present() {
|
||||
local _imgs _img _missing=0
|
||||
_imgs="$(docker compose "$@" config --images)" || return 1
|
||||
while IFS= read -r _img; do
|
||||
[ -n "$_img" ] || continue
|
||||
docker image inspect "$_img" >/dev/null 2>&1 || { printf 'missing image: %s\n' "$_img" >&2; _missing=1; }
|
||||
done <<< "$_imgs"
|
||||
return $_missing
|
||||
}
|
||||
Reference in New Issue
Block a user