From a456b502dbf16ec3fc99c2574bde892bc1ae9306 Mon Sep 17 00:00:00 2001 From: Logan Cusano Date: Tue, 27 Oct 2026 20:06:55 -0400 Subject: [PATCH 1/6] discord_radio: retained desired redelivery no longer undoes a local pause (node-26#34) A local pause records the applied seq; only a strictly newer seq from C2 lifts it. The retained message the broker redelivers on reconnect carries the same seq, so it now leaves the bots logged out. Co-Authored-By: Claude Opus 5.5 --- drb-edge-node/app/internal/discord_radio.py | 18 ++++++++++++++++-- drb-edge-node/tests/test_discord_radio.py | 12 ++++++++++++ 2 files changed, 28 insertions(+), 2 deletions(-) diff --git a/drb-edge-node/app/internal/discord_radio.py b/drb-edge-node/app/internal/discord_radio.py index c005ab5..d9cb3eb 100644 --- a/drb-edge-node/app/internal/discord_radio.py +++ b/drb-edge-node/app/internal/discord_radio.py @@ -183,6 +183,10 @@ class RadioManager: self._mqtt_down_since: Optional[float] = None self._failsafe = False self._paused = False + # The applied seq at the moment of a local pause (node-26#34). The + # broker redelivers the retained desired message on every reconnect + # with the SAME seq; only a strictly newer seq from C2 lifts the pause. + self._paused_at_seq: Optional[int] = None self._ticks = 0 # ------------------------------------------------------------------ @@ -210,15 +214,25 @@ class RadioManager: logger.error(f"Malformed Discord desired stream entry ignored: {s!r}") self._ensure_watchdog() self._desired = wanted - self._paused = False + if self._paused and self._redelivery_of_paused_state(seq): + logger.info(f"Discord desired seq {seq} is a redelivery — staying paused locally.") + else: + self._paused = False + self._paused_at_seq = None await self.reconcile() if seq is not None: self._applied_seq = seq await self.publish_actual() + def _redelivery_of_paused_state(self, seq: Optional[int]) -> bool: + """True when `seq` is not newer than what was applied at pause time.""" + return seq is not None and self._paused_at_seq is not None and seq <= self._paused_at_seq + async def pause_locally(self) -> None: - """Local-dashboard stop: log every bot out until C2 sends a new list.""" + """Local-dashboard stop: log every bot out until C2 sends a NEW list + (a seq strictly greater than the one applied now — node-26#34).""" self._paused = True + self._paused_at_seq = self._applied_seq await self.reconcile() await self.publish_actual() diff --git a/drb-edge-node/tests/test_discord_radio.py b/drb-edge-node/tests/test_discord_radio.py index 30e2856..5c3c2e4 100644 --- a/drb-edge-node/tests/test_discord_radio.py +++ b/drb-edge-node/tests/test_discord_radio.py @@ -263,6 +263,18 @@ async def test_local_pause_logs_out_until_next_desired(env): assert env.mgr.is_connected +async def test_retained_redelivery_does_not_undo_local_pause(env): + """node-26#34: the broker redelivers the retained desired message (same + seq) on every reconnect; that must not log the bots back in.""" + await env.mgr.apply_desired(_desired(4, (1, 11, "t1"))) + await env.mgr.pause_locally() + await env.mgr.apply_desired(_desired(4, (1, 11, "t1"))) + assert not env.mgr.is_connected + assert env.published[-1]["streams"]["1:11"]["state"] == "paused_locally" + await env.mgr.apply_desired(_desired(5, (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: From 1b4fcefbf20a45bfa8b62c8766788ff6463581e3 Mon Sep 17 00:00:00 2001 From: Logan Cusano Date: Tue, 27 Oct 2026 20:07:47 -0400 Subject: [PATCH 2/6] mqtt_manager: expose last connect error; health tells refused from unreachable (node-26#21) mqtt_manager.last_error = {kind, detail, epoch}: 'refused' from a non-zero CONNACK, 'unreachable' from the initial connect loop or paho's on_connect_fail, 'disconnected' otherwise. Cleared on connect. The /api/health MQTT check now gives a key fix for a refusal and a network fix for an unreachable broker instead of guessing from whether a key is on disk. Co-Authored-By: Claude Opus 5.5 --- drb-edge-node/app/internal/health.py | 17 ++++++++++++++ drb-edge-node/app/internal/mqtt_manager.py | 26 +++++++++++++++++++++ drb-edge-node/tests/test_local_dashboard.py | 11 +++++++++ drb-edge-node/tests/test_mqtt_manager.py | 25 ++++++++++++++++++++ 4 files changed, 79 insertions(+) diff --git a/drb-edge-node/app/internal/health.py b/drb-edge-node/app/internal/health.py index 1268a67..66ea358 100644 --- a/drb-edge-node/app/internal/health.py +++ b/drb-edge-node/app/internal/health.py @@ -154,6 +154,22 @@ def build_checks(f: Dict[str, Any]) -> List[Dict[str, str]]: f"Not connected to {broker} — expected until the node is approved (the broker only admits approved nodes).", "Fixes itself once the node is approved and has its API key (see above).", )) + elif (f.get("mqtt_last_error") or {}).get("kind") == "refused": + err = f["mqtt_last_error"] + checks.append(_check( + "mqtt", "Live link (MQTT)", ERROR, + f"{broker} refused this node's login ({err.get('detail')}).", + "The API key on file is not the one the broker expects — every re-approval issues a new one. " + "Re-run install.sh to fetch the current key, or ask a DRB admin to re-approve the node.", + )) + elif (f.get("mqtt_last_error") or {}).get("kind") == "unreachable": + err = f["mqtt_last_error"] + checks.append(_check( + "mqtt", "Live link (MQTT)", ERROR, + f"Can't reach {broker} ({err.get('detail')}).", + f"Check this site's internet connection, DNS, and that outbound port {f['mqtt_port']} isn't " + "blocked by a firewall. The key is not the problem.", + )) else: checks.append(_check( "mqtt", "Live link (MQTT)", ERROR, @@ -333,6 +349,7 @@ async def gather_facts() -> Dict[str, Any]: "c2_url": settings.c2_url, "has_api_key": bool(credentials.get_api_key()), "mqtt_connected": mqtt_manager.is_connected, + "mqtt_last_error": mqtt_manager.last_error, "mqtt_broker": settings.mqtt_broker, "mqtt_port": settings.mqtt_port, "sdr_count": sdr_count, diff --git a/drb-edge-node/app/internal/mqtt_manager.py b/drb-edge-node/app/internal/mqtt_manager.py index 09eb12a..52f29bd 100644 --- a/drb-edge-node/app/internal/mqtt_manager.py +++ b/drb-edge-node/app/internal/mqtt_manager.py @@ -1,6 +1,7 @@ import asyncio import json import shutil +import time from datetime import datetime, timezone from collections import deque from typing import Optional, Callable, Awaitable, Dict, Any @@ -60,6 +61,12 @@ class MQTTManager: # install.sh writing the file). Recovery compares against this. self._client_key: Optional[str] = None self._recovery_task: Optional[asyncio.Task] = None + # Why the link is down, for the dashboard (node-26#21): "refused" (the + # broker answered and said no), "unreachable" (TCP/TLS never got that + # far) or "disconnected" (an established session dropped). Cleared on + # every successful connect. Set from paho's thread as well as the event + # loop — a single attribute rebind, so no lock. + self._last_error: Optional[Dict[str, Any]] = None self.on_command: Optional[CommandCallback] = None self.on_config_push: Optional[ConfigCallback] = None @@ -146,12 +153,21 @@ class MQTTManager: client.reconnect_delay_set(min_delay=2, max_delay=60) client.on_connect = self._on_connect client.on_disconnect = self._on_disconnect + client.on_connect_fail = self._on_connect_fail client.on_message = self._on_message return client + def _set_error(self, kind: str, detail: str) -> None: + self._last_error = {"kind": kind, "detail": detail, "epoch": time.time()} + + def _on_connect_fail(self, client, userdata): + # paho's own reconnect (after loop_start) could not open the socket. + self._set_error("unreachable", f"cannot reach {settings.mqtt_broker}:{settings.mqtt_port}") + def _on_connect(self, client, userdata, flags, reason_code, properties): if reason_code == 0: self._connected = True + self._last_error = None client.subscribe(self._t_commands, qos=1) client.subscribe(self._t_config, qos=1) client.subscribe(self._t_discord_desired, qos=1) @@ -164,6 +180,7 @@ class MQTTManager: asyncio.run_coroutine_threadsafe(self._flush_offline_buffer(), self._loop) else: logger.error(f"MQTT connect refused: {reason_code}") + self._set_error("refused", str(reason_code)) if _is_auth_failure(reason_code) and self._loop is not None: # paho thread -> event loop. A refused login never fixes itself # by retrying the same password, so go get the right one. @@ -172,6 +189,9 @@ class MQTTManager: def _on_disconnect(self, client, userdata, disconnect_flags, reason_code, properties): self._connected = False logger.warning(f"MQTT disconnected: {reason_code}") + # A refusal is followed by a disconnect; keep the more specific reason. + if self._last_error is None: + self._set_error("disconnected", str(reason_code)) def _on_message(self, client, userdata, msg): try: @@ -273,6 +293,7 @@ class MQTTManager: return except Exception as e: logger.warning(f"MQTT connect failed ({e}) — retrying in {delay}s") + self._set_error("unreachable", f"cannot reach {settings.mqtt_broker}:{settings.mqtt_port}: {e}") await asyncio.sleep(delay) delay = min(delay * 2, 60) @@ -408,5 +429,10 @@ class MQTTManager: def is_connected(self) -> bool: return self._connected + @property + def last_error(self) -> Optional[Dict[str, Any]]: + """{kind, detail, epoch} for the most recent failure, None while connected.""" + return None if self._connected else self._last_error + mqtt_manager = MQTTManager() diff --git a/drb-edge-node/tests/test_local_dashboard.py b/drb-edge-node/tests/test_local_dashboard.py index 954fb23..e0ffe69 100644 --- a/drb-edge-node/tests/test_local_dashboard.py +++ b/drb-edge-node/tests/test_local_dashboard.py @@ -100,6 +100,17 @@ def test_key_on_disk_but_mqtt_down_says_restart(): assert "restart the edge-node container" in c["mqtt"]["action"] +def test_mqtt_refused_vs_unreachable_are_told_apart(): + """node-26#21: 'not authorised' and 'can't reach the broker' need different fixes.""" + refused = _by_id(health.build_checks(_facts( + mqtt_connected=False, mqtt_last_error={"kind": "refused", "detail": "Not authorized", "epoch": 1.0}))) + assert refused["mqtt"]["state"] == "error" + assert "refused" in refused["mqtt"]["summary"] and "install.sh" in refused["mqtt"]["action"] + unreachable = _by_id(health.build_checks(_facts( + mqtt_connected=False, mqtt_last_error={"kind": "unreachable", "detail": "timed out", "epoch": 1.0}))) + assert "Can't reach" in unreachable["mqtt"]["summary"] and "firewall" in unreachable["mqtt"]["action"] + + def test_standalone_node_is_not_an_error(): c = _by_id(health.build_checks(_facts(c2_url=None, has_api_key=False, mqtt_connected=False))) assert c["c2"]["state"] == "off" diff --git a/drb-edge-node/tests/test_mqtt_manager.py b/drb-edge-node/tests/test_mqtt_manager.py index 96686d6..31414e6 100644 --- a/drb-edge-node/tests/test_mqtt_manager.py +++ b/drb-edge-node/tests/test_mqtt_manager.py @@ -153,3 +153,28 @@ def test_disk_report_nulls_when_path_unreadable(tmp_path): # node-26#41: nulls (not omitted keys) so C2 never keeps a stale reading. from app.internal.mqtt_manager import disk_report assert disk_report(str(tmp_path / "missing")) == {"disk_free_bytes": None, "disk_total_bytes": None} + + +def test_last_error_records_refusal_and_clears_on_connect(monkeypatch): + """node-26#21: the refusal reason survives the disconnect that follows it.""" + from unittest.mock import MagicMock + mqtt_manager._last_error = None + mqtt_manager._loop = None + mqtt_manager._on_connect(MagicMock(), None, None, 135, None) + mqtt_manager._on_disconnect(MagicMock(), None, None, 135, None) + assert mqtt_manager.last_error["kind"] == "refused" + monkeypatch.setattr(mqtt_manager, "_flush_offline_buffer", MagicMock()) + monkeypatch.setattr(mqtt_manager, "_publish_checkin", MagicMock()) + monkeypatch.setattr(mqtt_manager, "_maybe_request_key", MagicMock()) + monkeypatch.setattr("asyncio.run_coroutine_threadsafe", lambda *a, **k: None) + mqtt_manager._on_connect(MagicMock(), None, None, 0, None) + assert mqtt_manager.last_error is None + mqtt_manager._connected = False + + +def test_last_error_unreachable_from_connect_fail(): + mqtt_manager._last_error = None + mqtt_manager._connected = False + mqtt_manager._on_connect_fail(None, None) + assert mqtt_manager.last_error["kind"] == "unreachable" + mqtt_manager._last_error = None From 6b884d54333263e9754a034b105b197efb5ba7fc Mon Sep 17 00:00:00 2001 From: Logan Cusano Date: Tue, 27 Oct 2026 20:14:36 -0400 Subject: [PATCH 3/6] call_recorder: spool failed uploads to disk and retry instead of deleting (node-26#44) upload_recording deleted the file in `finally` even when the POST failed, so any C2 outage, 5xx or timeout lost the call's audio for good. - Delete only after a 2xx that carries a URL (C2 answers 200 {url: null} when its GCS write fails). - Retryable failures (network, timeout, 5xx, 401/408/429, 2xx-no-url) move the file plus a JSON sidecar (call_id, talkgroup, system_id, audio epochs) into /spool. A background task retries with exponential backoff and drains the spool on startup. Other 4xx are deleted with an ERROR naming the status. - Bounded: UPLOAD_SPOOL_MAX_BYTES (512 MiB) and UPLOAD_SPOOL_MAX_AGE_HOURS (72), oldest evicted first, since a full host disk stops recording (#41). - Crash-safe: the sidecar is fsynced and written before the audio moves in; a lost or corrupt sidecar is rebuilt from the filename instead of deleting the audio. - The inline attempt uses UPLOAD_INLINE_TIMEOUT_SECONDS (30), so a black-holed C2 no longer stalls the segmenter for 120 s per call. - call_end carries audio_upload=spooled|rejected|lost; a new API key wakes the spool; /api/status and the checkin report spool depth and the last upload outcome (toward #40). Reviewed by drb-audio-review. Its two server-side findings (late uploads correlated against now; no /upload idempotency) are filed separately. Co-Authored-By: Claude Opus 5.5 --- .env.example | 9 + drb-edge-node/app/config.py | 17 + drb-edge-node/app/internal/call_recorder.py | 94 +++-- drb-edge-node/app/internal/mqtt_manager.py | 8 + drb-edge-node/app/internal/upload_spool.py | 416 ++++++++++++++++++++ drb-edge-node/app/main.py | 18 +- drb-edge-node/app/routers/api.py | 3 + drb-edge-node/tests/test_upload_spool.py | 230 +++++++++++ 8 files changed, 752 insertions(+), 43 deletions(-) create mode 100644 drb-edge-node/app/internal/upload_spool.py create mode 100644 drb-edge-node/tests/test_upload_spool.py diff --git a/.env.example b/.env.example index a46faf3..3261700 100644 --- a/.env.example +++ b/.env.example @@ -99,6 +99,15 @@ TRIM_SILENCE_THRESHOLD_DB=-40 # Seconds of audio kept either side of detected speech. TRIM_SILENCE_GUARD_SECONDS=0.25 +# Upload spool: a recording whose upload to C2 fails is kept on disk and +# retried, never deleted (default dir: /spool). Bounded so a long +# C2 outage cannot fill the host disk; the oldest recordings go first. +# UPLOAD_SPOOL_DIR=/recordings/spool +UPLOAD_SPOOL_MAX_BYTES=536870912 +UPLOAD_SPOOL_MAX_AGE_HOURS=72 +UPLOAD_RETRY_MIN_SECONDS=15 +UPLOAD_RETRY_MAX_SECONDS=600 + # OP25 container (usually no need to change) OP25_API_URL=http://localhost:8001 OP25_TERMINAL_URL=http://localhost:8081 diff --git a/drb-edge-node/app/config.py b/drb-edge-node/app/config.py index 63a53c3..1db01b5 100644 --- a/drb-edge-node/app/config.py +++ b/drb-edge-node/app/config.py @@ -153,6 +153,23 @@ class Settings(BaseSettings): # Offline call buffer — how many call_end events to keep while disconnected offline_call_buffer_size: int = 35 + # Upload spool (node-26#44) — recordings whose upload to C2 failed are kept + # here and retried instead of deleted. Defaults to /spool, + # i.e. on the host disk via the /recordings bind mount, so it survives a + # restart. Bounded both ways because a full host disk stops recording + # outright (node-26#41): FLAC is ~1.3 MB per minute of audio, so 512 MiB is + # roughly 6 h of continuous traffic. Oldest entries are evicted first. + upload_spool_dir: Optional[str] = None + upload_spool_max_bytes: int = 512 * 1024 * 1024 + upload_spool_max_age_hours: float = 72.0 + # Exponential backoff between retries of one spooled recording. + upload_retry_min_seconds: float = 15.0 + upload_retry_max_seconds: float = 600.0 + # Timeout for the FIRST, inline upload attempt. metadata_watcher awaits the + # call close, so a black-holed C2 used to stall segmentation for the full + # 120 s; a failure here just lands in the spool. + upload_inline_timeout_seconds: float = 30.0 + # ------------------------------------------------------------------ # Local dashboard / API authentication # diff --git a/drb-edge-node/app/internal/call_recorder.py b/drb-edge-node/app/internal/call_recorder.py index f6eb98f..ff80551 100644 --- a/drb-edge-node/app/internal/call_recorder.py +++ b/drb-edge-node/app/internal/call_recorder.py @@ -66,10 +66,10 @@ from datetime import datetime, timezone from pathlib import Path from typing import List, Optional, Tuple -import httpx from app.config import settings -from app.internal import audio_trim, credentials, pcm, pulse +from app.internal import audio_trim, pcm, pulse, upload_spool from app.internal.logger import logger +from app.internal.upload_spool import UploadMeta, UploadResult, UploadSpool # Safety cap on a single recording; mirrors MAX_SEGMENT_SECONDS in metadata_watcher. MAX_RECORDING_SECONDS = 600 @@ -320,6 +320,17 @@ class CallRecorder: # why a capture process exited (see _classify_capture_exit). self._last_stderr_lines: deque[str] = deque(maxlen=10) + # Failed uploads wait here for a retry instead of being deleted + # (node-26#44). See internal/upload_spool.py. + self.spool = UploadSpool( + Path(settings.upload_spool_dir or (Path(settings.recordings_path) / "spool")), + uploader=self._post, + max_bytes=settings.upload_spool_max_bytes, + max_age_seconds=settings.upload_spool_max_age_hours * 3600, + backoff_min=settings.upload_retry_min_seconds, + backoff_max=settings.upload_retry_max_seconds, + ) + # ------------------------------------------------------------------ # Lifecycle # ------------------------------------------------------------------ @@ -328,8 +339,11 @@ class CallRecorder: """Start the persistent capture. Call once from app lifespan.""" self._stream_task = asyncio.create_task(self._capture_loop()) logger.info("PulseAudio ring-buffer starting.") + if settings.c2_url: + self.spool.start() async def stop(self) -> None: + await self.spool.stop() if self._stream_task: self._stream_task.cancel() try: @@ -751,6 +765,17 @@ class CallRecorder: # Upload (unchanged interface) # ------------------------------------------------------------------ + async def _post(self, file_path: Path, meta: UploadMeta) -> UploadResult: + return await upload_spool.post_recording(file_path, meta, AUDIO_MIME) + + async def _post_inline(self, file_path: Path, meta: UploadMeta) -> UploadResult: + # metadata_watcher awaits on_call_end, so this attempt blocks the poll + # loop. With the spool behind it, it can afford a short timeout; the + # background retries get the full UPLOAD_TIMEOUT_SECONDS. + return await upload_spool.post_recording( + file_path, meta, AUDIO_MIME, timeout=settings.upload_inline_timeout_seconds, + ) + async def upload_recording( self, file_path: Path, @@ -761,50 +786,37 @@ class CallRecorder: audio_start_epoch: Optional[float] = None, audio_end_epoch: Optional[float] = None, ) -> Optional[str]: + """ + Upload once, inline. Returns the storage URL on success. + + The file is deleted ONLY after a 2xx with a URL (node-26#44). A + retryable failure moves it into the spool, which keeps retrying in the + background; C2's /upload writes `audio_gcs_uri` onto the call doc by + call_id whenever the upload lands, so a late upload still attaches. + """ if not settings.c2_url: logger.info("No C2_URL configured — skipping upload.") + file_path.unlink(missing_ok=True) return None - upload_url = f"{settings.c2_url}/upload" - api_key = credentials.get_api_key() - headers = {"Authorization": f"Bearer {api_key}"} if api_key else {} - - form: dict = {"call_id": call_id, "node_id": settings.node_id} - if talkgroup_id is not None: - form["talkgroup_id"] = str(talkgroup_id) - if talkgroup_name: - form["talkgroup_name"] = talkgroup_name - if system_id: - form["system_id"] = system_id - # Where this audio really sits on the wall clock once silence is trimmed. - # C2 does not declare these Form fields yet, so FastAPI ignores them — - # they cost nothing and are here for when playback/correlation want them. - if audio_start_epoch is not None: - form["audio_start_epoch"] = f"{audio_start_epoch:.3f}" - if audio_end_epoch is not None: - form["audio_end_epoch"] = f"{audio_end_epoch:.3f}" - - try: - async with httpx.AsyncClient(timeout=120) as client: - with open(file_path, "rb") as f: - r = await client.post( - upload_url, - files={"file": (file_path.name, f, AUDIO_MIME)}, - data=form, - headers=headers, - ) - r.raise_for_status() - audio_url = r.json().get("url") - logger.info(f"Upload complete: {audio_url}") - return audio_url - except Exception as e: - logger.error(f"Upload failed: {e}") + meta = UploadMeta( + call_id=call_id, talkgroup_id=talkgroup_id, talkgroup_name=talkgroup_name, + system_id=system_id, audio_start_epoch=audio_start_epoch, audio_end_epoch=audio_end_epoch, + ) + result = await self._post_inline(file_path, meta) + self.spool.last_result = {"call_id": call_id, "kind": result.kind, "status": result.status, + "detail": result.detail, "epoch": time.time()} + if result.kind == upload_spool.OK: + logger.info(f"Upload complete: {result.url}") + file_path.unlink(missing_ok=True) + return result.url + if result.kind == upload_spool.REJECT: + logger.error(f"C2 rejected the upload of call {call_id} ({result.detail}) — not retryable, deleting it.") + file_path.unlink(missing_ok=True) return None - finally: - try: - file_path.unlink() - except Exception: - pass + if self.spool.add(file_path, meta, result) is None: + self.spool.last_result = {**self.spool.last_result, "kind": "lost"} + return None # ------------------------------------------------------------------ # State diff --git a/drb-edge-node/app/internal/mqtt_manager.py b/drb-edge-node/app/internal/mqtt_manager.py index 52f29bd..576e993 100644 --- a/drb-edge-node/app/internal/mqtt_manager.py +++ b/drb-edge-node/app/internal/mqtt_manager.py @@ -411,6 +411,14 @@ class MQTTManager: if count is not None: payload["sdr_count"] = count payload.update(disk_report(settings.recordings_path)) + # Upload spool depth + last upload outcome (node-26#44, toward #40). + from app.internal.call_recorder import call_recorder + spool = call_recorder.spool.stats() + payload["upload_spool_files"] = spool["files"] + payload["upload_spool_bytes"] = spool["bytes"] + last = spool["last_result"] or {} + payload["last_upload_status"] = last.get("status") + payload["last_upload_kind"] = last.get("kind") self._publish(self._t_checkin, payload, qos=1) def _publish(self, topic: str, payload: dict, qos: int = 0, retain: bool = False): diff --git a/drb-edge-node/app/internal/upload_spool.py b/drb-edge-node/app/internal/upload_spool.py new file mode 100644 index 0000000..ddc2015 --- /dev/null +++ b/drb-edge-node/app/internal/upload_spool.py @@ -0,0 +1,416 @@ +""" +On-disk spool for recordings whose upload to C2 failed (node-26#44). + +Before this, `upload_recording` deleted the file in a `finally` whether or not +the POST succeeded, so a C2 outage, a network blip, a 5xx or a timeout lost the +call's audio for good — and with it the transcript and the incident link. + +Now a recording is deleted only after a 2xx that carries a storage URL. A +failure that can succeed later is MOVED into this spool with a JSON sidecar +holding everything needed to re-POST it, and a background task retries with +exponential backoff. The spool is drained on startup, so a reboot or container +restart does not lose it either. + +Failure classes (see `classify`): + + retry network error, timeout, 5xx, 2xx with no URL (C2 accepted the + request but storage failed), 401 (a reissued key is on its way + over MQTT/key pickup), 408, 429 + reject any other 4xx — 400 empty file, 403 not approved, 413 too large. + Retrying the same bytes cannot change the answer, so the file is + deleted with an ERROR log naming the status. + +BOUNDED on purpose: /recordings is a bind mount of the host disk, and a full +disk already stopped node-002 recording for 17 days (node-26#41). Entries older +than `upload_spool_max_age_hours` are dropped, and past +`upload_spool_max_bytes` the OLDEST entries go first — the newest audio is the +most likely to still matter to a live incident. + +File layout, one pair per call: + + /.flac + /.flac.json sidecar (written FIRST, atomically) + +The sidecar is written before the audio is moved in, and deleted after the +audio is removed, so a crash at any point leaves at worst a sidecar with no +audio — which the next scan cleans up — and never audio that cannot be +re-uploaded. +""" +import asyncio +import json +import os +import shutil +import time +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Any, Awaitable, Callable, Dict, List, Optional, Tuple + +import httpx + +from app.config import settings +from app.internal import credentials +from app.internal.logger import logger + +SIDECAR_SUFFIX = ".json" +UPLOAD_TIMEOUT_SECONDS = 120.0 +# Statuses in the 4xx range that are worth retrying unchanged. +RETRYABLE_4XX = {401, 408, 429} + +OK, RETRY, REJECT = "ok", "retry", "reject" + + +@dataclass +class UploadMeta: + """Everything /upload needs besides the file itself.""" + + call_id: str + talkgroup_id: Optional[int] = None + talkgroup_name: Optional[str] = None + system_id: Optional[str] = None + audio_start_epoch: Optional[float] = None + audio_end_epoch: Optional[float] = None + + +@dataclass +class UploadResult: + kind: str # OK | RETRY | REJECT + url: Optional[str] = None + status: Optional[int] = None # HTTP status, None for a network error + detail: str = "" + + +def classify(status: int, url: Optional[str]) -> str: + if 200 <= status < 300: + # C2 answers 200 {"url": null} when its GCS write failed: nothing was + # stored, so this is a server-side failure, not a success. + return OK if url else RETRY + if status in RETRYABLE_4XX or status >= 500: + return RETRY + if 400 <= status < 500: + return REJECT + return RETRY + + +def _form(meta: UploadMeta) -> Dict[str, str]: + form: Dict[str, str] = {"call_id": meta.call_id, "node_id": settings.node_id} + if meta.talkgroup_id is not None: + form["talkgroup_id"] = str(meta.talkgroup_id) + if meta.talkgroup_name: + form["talkgroup_name"] = meta.talkgroup_name + if meta.system_id: + form["system_id"] = meta.system_id + # Where this audio really sits on the wall clock once silence is trimmed. + # C2 does not declare these Form fields yet, so FastAPI ignores them — + # they cost nothing and are here for when playback/correlation want them. + if meta.audio_start_epoch is not None: + form["audio_start_epoch"] = f"{meta.audio_start_epoch:.3f}" + if meta.audio_end_epoch is not None: + form["audio_end_epoch"] = f"{meta.audio_end_epoch:.3f}" + return form + + +async def post_recording( + file_path: Path, meta: UploadMeta, mime: str, timeout: float = UPLOAD_TIMEOUT_SECONDS, +) -> UploadResult: + """One POST to C2 /upload. Never raises; never touches the file on disk.""" + upload_url = f"{settings.c2_url}/upload" + api_key = credentials.get_api_key() + headers = {"Authorization": f"Bearer {api_key}"} if api_key else {} + try: + async with httpx.AsyncClient(timeout=timeout) as client: + with open(file_path, "rb") as f: + r = await client.post( + upload_url, + files={"file": (file_path.name, f, mime)}, + data=_form(meta), + headers=headers, + ) + except Exception as e: + return UploadResult(RETRY, detail=f"{type(e).__name__}: {e}") + + url: Optional[str] = None + if 200 <= r.status_code < 300: + try: + url = r.json().get("url") + except Exception: + url = None + kind = classify(r.status_code, url) + detail = f"HTTP {r.status_code}" + if kind != OK: + detail += f": {r.text[:200]}" + return UploadResult(kind, url=url, status=r.status_code, detail=detail) + + +def _fsync_dir(directory: Path) -> None: + try: + fd = os.open(directory, os.O_RDONLY) + except OSError: + return + try: + os.fsync(fd) + except OSError: + pass + finally: + os.close(fd) + + +Uploader = Callable[[Path, UploadMeta], Awaitable[UploadResult]] + + +class UploadSpool: + def __init__( + self, + directory: Path, + uploader: Uploader, + max_bytes: int, + max_age_seconds: float, + backoff_min: float, + backoff_max: float, + poll_seconds: float = 5.0, + clock: Callable[[], float] = time.time, + ): + self.directory = Path(directory) + self._upload = uploader + self.max_bytes = max_bytes + self.max_age_seconds = max_age_seconds + self.backoff_min = backoff_min + self.backoff_max = backoff_max + self.poll_seconds = poll_seconds + self._clock = clock + self._task: Optional[asyncio.Task] = None + self._wake = asyncio.Event() + self.last_result: Optional[Dict[str, Any]] = None + + # ------------------------------------------------------------------ + # Entries + # ------------------------------------------------------------------ + + def _backoff(self, attempts: int) -> float: + return min(self.backoff_min * (2 ** max(0, attempts - 1)), self.backoff_max) + + @staticmethod + def _write_sidecar(path: Path, data: Dict[str, Any]) -> None: + # fsync file AND directory: on a Pi that loses power, an un-synced + # rename can come back as a zero-length sidecar after the audio move + # was already journaled. + tmp = path.with_name(path.name + ".tmp") + with open(tmp, "w") as f: + f.write(json.dumps(data)) + f.flush() + os.fsync(f.fileno()) + os.replace(tmp, path) + _fsync_dir(path.parent) + + def _recover_sidecar(self, audio: Path, sidecar: Path, why: str) -> Optional[Dict[str, Any]]: + """Rebuild minimal metadata from the recording's own name + (`YYYYMMDD_HHMMSS_.flac`) rather than delete good audio. C2 + already has the talkgroup from call_start; only call_id is required.""" + parts = audio.name.split("_", 2) + call_id = parts[2].split(".", 1)[0] if len(parts) == 3 else "" + if not call_id: + logger.error(f"Spooled audio {audio.name}: {why}, and no call_id in its name — dropping it.") + self._remove(audio, sidecar) + return None + logger.error(f"Spooled audio {audio.name}: {why} — rebuilt metadata from the filename (call {call_id}).") + try: + spooled_at = audio.stat().st_mtime + except OSError: + spooled_at = self._clock() + data = {**asdict(UploadMeta(call_id=call_id)), "spooled_at": spooled_at, "attempts": 1, + "next_attempt_at": self._clock(), "last_error": why, "last_status": None} + self._write_sidecar(sidecar, data) + return data + + def add(self, file_path: Path, meta: UploadMeta, result: UploadResult) -> Optional[Path]: + """Move a recording whose upload failed into the spool. Returns its new path.""" + try: + self.directory.mkdir(parents=True, exist_ok=True) + dest = self.directory / file_path.name + sidecar = dest.with_name(dest.name + SIDECAR_SUFFIX) + now = self._clock() + self._write_sidecar(sidecar, { + **asdict(meta), + "spooled_at": now, + "attempts": 1, + "next_attempt_at": now + self._backoff(1), + "last_error": result.detail, + "last_status": result.status, + }) + shutil.move(str(file_path), str(dest)) + _fsync_dir(self.directory) + except Exception as e: + logger.error(f"Could not spool recording for call {meta.call_id} ({e}) — audio is lost.") + return None + logger.warning( + f"Upload of call {meta.call_id} failed ({result.detail}) — spooled for retry " + f"({self.stats()['files']} waiting)." + ) + self._wake.set() + return dest + + def _entries(self) -> List[Tuple[Path, Path, Dict[str, Any]]]: + """(audio, sidecar, data) oldest first. Cleans up half-written pairs.""" + if not self.directory.is_dir(): + return [] + entries = [] + sidecars = set() + for sidecar in self.directory.glob(f"*{SIDECAR_SUFFIX}"): + audio = sidecar.with_name(sidecar.name[: -len(SIDECAR_SUFFIX)]) + sidecars.add(audio.name) + if not audio.exists(): + sidecar.unlink(missing_ok=True) + continue + try: + data = json.loads(sidecar.read_text()) + if not data.get("call_id"): + raise ValueError("no call_id") + except Exception as e: + data = self._recover_sidecar(audio, sidecar, f"unreadable sidecar ({e})") + if data is None: + continue + entries.append((audio, sidecar, data)) + for audio in list(self.directory.iterdir()): + if audio.is_file() and not audio.name.endswith((SIDECAR_SUFFIX, ".tmp")) and audio.name not in sidecars: + sidecar = audio.with_name(audio.name + SIDECAR_SUFFIX) + data = self._recover_sidecar(audio, sidecar, "no metadata sidecar") + if data is not None: + entries.append((audio, sidecar, data)) + entries.sort(key=lambda e: (e[2].get("spooled_at", 0.0), e[0].name)) + return entries + + @staticmethod + def _remove(audio: Path, sidecar: Path) -> None: + # Audio first: a crash in between leaves a sidecar, which _entries() cleans. + audio.unlink(missing_ok=True) + sidecar.unlink(missing_ok=True) + + def stats(self) -> Dict[str, Any]: + entries = self._entries() + size = 0 + for audio, _, _ in entries: + try: + size += audio.stat().st_size + except OSError: + pass + oldest = entries[0][2].get("spooled_at") if entries else None + return { + "files": len(entries), + "bytes": size, + "oldest_age_seconds": round(self._clock() - oldest, 1) if oldest else None, + "last_result": self.last_result, + } + + def enforce_caps(self) -> int: + """Drop expired entries, then the oldest until under max_bytes. Returns how many went.""" + entries = self._entries() + now = self._clock() + dropped = 0 + kept = [] + for audio, sidecar, data in entries: + if now - data.get("spooled_at", now) > self.max_age_seconds: + logger.error( + f"Spooled call {data['call_id']} was never uploaded within " + f"{self.max_age_seconds / 3600:.0f}h ({data.get('attempts')} attempts, " + f"last: {data.get('last_error')}) — deleting its audio." + ) + self._remove(audio, sidecar) + dropped += 1 + else: + kept.append((audio, sidecar, data)) + sizes = [] + for audio, sidecar, data in kept: + try: + sizes.append(audio.stat().st_size) + except OSError: + sizes.append(0) + total = sum(sizes) + i = 0 + while total > self.max_bytes and i < len(kept): + audio, sidecar, data = kept[i] + logger.error( + f"Upload spool over {self.max_bytes} bytes — evicting the oldest, call {data['call_id']}." + ) + self._remove(audio, sidecar) + total -= sizes[i] + dropped += 1 + i += 1 + return dropped + + # ------------------------------------------------------------------ + # Retry + # ------------------------------------------------------------------ + + async def drain_once(self, force: bool = False) -> int: + """Attempt every due entry (every entry if `force`). Returns how many were uploaded.""" + self.enforce_caps() + uploaded = 0 + for audio, sidecar, data in self._entries(): + if not force and data.get("next_attempt_at", 0.0) > self._clock(): + continue + meta = UploadMeta(**{k: data.get(k) for k in UploadMeta.__dataclass_fields__}) + result = await self._upload(audio, meta) + self.last_result = {"call_id": meta.call_id, "kind": result.kind, "status": result.status, + "detail": result.detail, "epoch": self._clock()} + if result.kind == OK: + logger.info(f"Spooled call {meta.call_id} uploaded after {data.get('attempts')} failed attempt(s).") + self._remove(audio, sidecar) + uploaded += 1 + elif result.kind == REJECT: + logger.error( + f"C2 rejected spooled call {meta.call_id} ({result.detail}) — not retryable, deleting it." + ) + self._remove(audio, sidecar) + else: + attempts = int(data.get("attempts", 1)) + 1 + data.update(attempts=attempts, last_error=result.detail, last_status=result.status, + next_attempt_at=self._clock() + self._backoff(attempts)) + self._write_sidecar(sidecar, data) + logger.warning( + f"Retry {attempts} of spooled call {meta.call_id} failed ({result.detail}); next in " + f"{self._backoff(attempts):.0f}s." + ) + # C2 is still down: the rest would fail the same way. Leave them + # for their own timers rather than hammering it in one pass. + break + return uploaded + + def wake(self) -> None: + """Retry everything now (e.g. a new API key just arrived).""" + for audio, sidecar, data in self._entries(): + if data.get("next_attempt_at", 0.0) > self._clock(): + data["next_attempt_at"] = self._clock() + self._write_sidecar(sidecar, data) + self._wake.set() + + async def _run(self) -> None: + # Startup drain: whatever survived a restart is attempted immediately. + force = True + while True: + try: + await self.drain_once(force=force) + except asyncio.CancelledError: + raise + except Exception as e: + logger.error(f"Upload spool pass failed: {e}") + force = False + self._wake.clear() + try: + await asyncio.wait_for(self._wake.wait(), timeout=self.poll_seconds) + except asyncio.TimeoutError: + pass + + def start(self) -> None: + if self._task is None or self._task.done(): + pending = len(self._entries()) + if pending: + logger.info(f"Upload spool has {pending} recording(s) from before this start — draining.") + self._task = asyncio.create_task(self._run()) + + async def stop(self) -> None: + if self._task: + self._task.cancel() + try: + await self._task + except asyncio.CancelledError: + pass + self._task = None diff --git a/drb-edge-node/app/main.py b/drb-edge-node/app/main.py index 28156e6..4f26efa 100644 --- a/drb-edge-node/app/main.py +++ b/drb-edge-node/app/main.py @@ -111,8 +111,20 @@ async def on_call_end(data: dict): ) if audio_url: data["audio_url"] = audio_url - else: - logger.error(f"Audio upload failed for call {data['call_id']}. Verify C2_URL and Node API Key.") + elif settings.c2_url: + last = call_recorder.spool.last_result or {} + if last.get("kind") == "lost": + data["audio_upload"] = "lost" + elif last.get("kind") == "retry": + # The file is in the upload spool and will be retried; C2 attaches + # it to this call_id whenever it lands (node-26#44). + data["audio_upload"] = "spooled" + else: + data["audio_upload"] = "rejected" + logger.error( + f"Audio upload for call {data['call_id']} was rejected ({last.get('detail')}). " + "Verify the node is approved and its API key is current." + ) elif recording is not None and recording.all_silence: # Explicit policy: an all-silence recording is not uploaded. It has no # transcript value and silence is what makes Whisper invent text. @@ -179,6 +191,8 @@ async def on_api_key(payload: dict): if key: credentials.save_api_key(key) mqtt_manager.update_api_key(key) + # Uploads spooled on a 401 can go now rather than at their next backoff. + call_recorder.spool.wake() logger.info("Node API key received and saved.") diff --git a/drb-edge-node/app/routers/api.py b/drb-edge-node/app/routers/api.py index 716f80d..dddc78e 100644 --- a/drb-edge-node/app/routers/api.py +++ b/drb-edge-node/app/routers/api.py @@ -82,6 +82,9 @@ async def get_status(): # it was discarded rather than uploaded. Non-zero means either the # console is not decoding or something else is feeding drb_sink. "unattributed_segments": metadata_watcher.unattributed_segments, + # Recordings whose upload to C2 failed and are waiting for a retry + # (node-26#44), plus the outcome of the most recent upload attempt. + "upload_spool": call_recorder.spool.stats(), "active_tgid": active_tgid, "active_tgid_name": active_tgid_name, "active_call_id": metadata_watcher.active_call_id, diff --git a/drb-edge-node/tests/test_upload_spool.py b/drb-edge-node/tests/test_upload_spool.py new file mode 100644 index 0000000..8976e6d --- /dev/null +++ b/drb-edge-node/tests/test_upload_spool.py @@ -0,0 +1,230 @@ +""" +Upload spool (node-26#44): a failed upload must never delete the recording. + +Before the spool, upload_recording unlinked the file in a `finally`, so any C2 +outage, 5xx or timeout lost that call's audio for good. +""" +import asyncio +import json +from pathlib import Path + +import pytest + +from app.config import settings +from app.internal import upload_spool +from app.internal.call_recorder import CallRecorder +from app.internal.upload_spool import OK, REJECT, RETRY, UploadMeta, UploadResult, UploadSpool, classify + + +class Clock: + def __init__(self, t: float = 1_000_000.0): + self.t = t + + def __call__(self) -> float: + return self.t + + +class FakeUploader: + def __init__(self, *results: UploadResult): + self.results = list(results) + self.calls = [] + + async def __call__(self, path: Path, meta: UploadMeta) -> UploadResult: + assert path.exists(), "the uploader must be handed a file that exists" + self.calls.append((path.name, meta)) + return self.results.pop(0) if self.results else UploadResult(OK, url="gs://x") + + +def _recording(tmp_path: Path, name: str = "20261027_call-1.flac", size: int = 100) -> Path: + p = tmp_path / "rec" / name + p.parent.mkdir(parents=True, exist_ok=True) + p.write_bytes(b"f" * size) + return p + + +def _spool(tmp_path: Path, uploader, clock=None, **kw) -> UploadSpool: + args = dict(max_bytes=10_000, max_age_seconds=3600, backoff_min=10, backoff_max=80) + args.update(kw) + return UploadSpool(tmp_path / "spool", uploader, clock=clock or Clock(), **args) + + +# --------------------------------------------------------------------------- +# Classification +# --------------------------------------------------------------------------- + +@pytest.mark.parametrize("status,url,kind", [ + (200, "gs://b/o", OK), + (200, None, RETRY), # C2's GCS write failed: nothing stored + (401, None, RETRY), # a reissued key is on its way + (408, None, RETRY), + (429, None, RETRY), + (500, None, RETRY), + (503, None, RETRY), + (400, None, REJECT), + (403, None, REJECT), + (413, None, REJECT), +]) +def test_classify(status, url, kind): + assert classify(status, url) == kind + + +# --------------------------------------------------------------------------- +# upload_recording: delete only on success +# --------------------------------------------------------------------------- + +@pytest.fixture +def recorder(tmp_path, monkeypatch): + monkeypatch.setattr(settings, "c2_url", "https://c2.example") + rec = CallRecorder() + rec.spool = _spool(tmp_path, rec._post) + return rec + + +def _patch_post(monkeypatch, result: UploadResult): + async def fake(path, meta, mime, timeout=None): + return result + monkeypatch.setattr(upload_spool, "post_recording", fake) + + +async def test_failed_upload_keeps_the_file_in_the_spool(tmp_path, recorder, monkeypatch): + _patch_post(monkeypatch, UploadResult(RETRY, detail="ConnectError: refused")) + path = _recording(tmp_path) + url = await recorder.upload_recording(path, "call-1", talkgroup_id=101, talkgroup_name="Dispatch", + system_id="sys-1", audio_start_epoch=1.5, audio_end_epoch=9.25) + assert url is None + assert not path.exists() + spooled = tmp_path / "spool" / path.name + assert spooled.read_bytes() == b"f" * 100 + meta = json.loads((tmp_path / "spool" / (path.name + ".json")).read_text()) + assert meta["call_id"] == "call-1" and meta["talkgroup_id"] == 101 and meta["system_id"] == "sys-1" + assert meta["audio_start_epoch"] == 1.5 and meta["audio_end_epoch"] == 9.25 + assert meta["last_error"] == "ConnectError: refused" + assert recorder.spool.stats()["files"] == 1 + + +async def test_successful_upload_deletes_the_file(tmp_path, recorder, monkeypatch): + _patch_post(monkeypatch, UploadResult(OK, url="gs://b/call-1.flac", status=200)) + path = _recording(tmp_path) + assert await recorder.upload_recording(path, "call-1") == "gs://b/call-1.flac" + assert not path.exists() + assert recorder.spool.stats()["files"] == 0 + + +async def test_rejected_upload_is_not_spooled(tmp_path, recorder, monkeypatch): + _patch_post(monkeypatch, UploadResult(REJECT, status=403, detail="HTTP 403: Node is not approved")) + path = _recording(tmp_path) + assert await recorder.upload_recording(path, "call-1") is None + assert not path.exists() + assert recorder.spool.stats()["files"] == 0 + assert recorder.spool.last_result["kind"] == REJECT + + +async def test_real_post_network_error_is_retryable(tmp_path, monkeypatch): + monkeypatch.setattr(settings, "c2_url", "http://127.0.0.1:9") # discard port: refused + path = _recording(tmp_path) + result = await upload_spool.post_recording(path, UploadMeta(call_id="c"), "audio/flac") + assert result.kind == RETRY and result.status is None + assert path.exists(), "post_recording must never touch the file" + + +# --------------------------------------------------------------------------- +# Retry, startup drain, caps +# --------------------------------------------------------------------------- + +async def test_retry_backs_off_then_uploads(tmp_path): + clock = Clock() + up = FakeUploader(UploadResult(RETRY, status=503, detail="HTTP 503"), UploadResult(OK, url="gs://x")) + sp = _spool(tmp_path, up, clock=clock) + sp.add(_recording(tmp_path), UploadMeta(call_id="c1"), UploadResult(RETRY, detail="first")) + + assert await sp.drain_once() == 0 and up.calls == [] # not due yet + clock.t += 10 + assert await sp.drain_once() == 0 and len(up.calls) == 1 # retry 2 fails + data = json.loads(next((tmp_path / "spool").glob("*.json")).read_text()) + assert data["attempts"] == 2 and data["next_attempt_at"] == clock.t + 20 + clock.t += 20 + assert await sp.drain_once() == 1 + assert list((tmp_path / "spool").iterdir()) == [] + + +async def test_startup_drain_uploads_what_survived_a_restart(tmp_path): + clock = Clock() + first = _spool(tmp_path, FakeUploader(), clock=clock) + first.add(_recording(tmp_path, "a_c1.flac"), UploadMeta(call_id="c1", talkgroup_id=7), UploadResult(RETRY)) + first.add(_recording(tmp_path, "b_c2.flac"), UploadMeta(call_id="c2"), UploadResult(RETRY)) + + # A new process: same directory, nothing due by the sidecar timers. + up = FakeUploader() + restarted = _spool(tmp_path, up, clock=clock, poll_seconds=0.01) + restarted.start() + for _ in range(100): + if len(up.calls) == 2: + break + await asyncio.sleep(0.01) + await restarted.stop() + assert [m.call_id for _, m in up.calls] == ["c1", "c2"] + assert up.calls[0][1].talkgroup_id == 7 + assert list((tmp_path / "spool").iterdir()) == [] + + +async def test_rejected_on_retry_is_deleted(tmp_path): + clock = Clock() + sp = _spool(tmp_path, FakeUploader(UploadResult(REJECT, status=413)), clock=clock) + sp.add(_recording(tmp_path), UploadMeta(call_id="c1"), UploadResult(RETRY)) + await sp.drain_once(force=True) + assert sp.stats()["files"] == 0 + + +async def test_age_cap_drops_expired_entries(tmp_path): + clock = Clock() + sp = _spool(tmp_path, FakeUploader(), clock=clock, max_age_seconds=3600) + sp.add(_recording(tmp_path, "old_c1.flac"), UploadMeta(call_id="c1"), UploadResult(RETRY)) + clock.t += 1800 + sp.add(_recording(tmp_path, "new_c2.flac"), UploadMeta(call_id="c2"), UploadResult(RETRY)) + clock.t += 1801 + assert sp.enforce_caps() == 1 + assert [p.name for p in (tmp_path / "spool").glob("*.flac")] == ["new_c2.flac"] + + +async def test_size_cap_evicts_oldest_first(tmp_path): + clock = Clock() + sp = _spool(tmp_path, FakeUploader(), clock=clock, max_bytes=250) + for i in range(3): + sp.add(_recording(tmp_path, f"{i}_c{i}.flac", size=100), UploadMeta(call_id=f"c{i}"), UploadResult(RETRY)) + clock.t += 1 + assert sp.enforce_caps() == 1 + assert sorted(p.name for p in (tmp_path / "spool").glob("*.flac")) == ["1_c1.flac", "2_c2.flac"] + assert sp.stats()["bytes"] == 200 + + +async def test_half_written_pairs_are_cleaned(tmp_path): + sp = _spool(tmp_path, FakeUploader()) + d = tmp_path / "spool" + d.mkdir() + (d / "lost.flac.json").write_text(json.dumps({"call_id": "x", "spooled_at": 0})) # no audio + (d / "orphan.flac").write_bytes(b"a") # no sidecar, no call_id in name + assert sp.stats()["files"] == 0 + assert list(d.iterdir()) == [] + + +async def test_lost_or_corrupt_sidecar_is_rebuilt_from_the_filename(tmp_path): + """A power cut can zero a sidecar; the audio must survive it.""" + up = FakeUploader() + sp = _spool(tmp_path, up) + d = tmp_path / "spool" + d.mkdir() + (d / "20261027_120000_call-a.flac").write_bytes(b"a") + (d / "20261027_120001_call_b.flac").write_bytes(b"b") + (d / "20261027_120001_call_b.flac.json").write_text("") + assert sp.stats()["files"] == 2 + await sp.drain_once() + assert sorted(m.call_id for _, m in up.calls) == ["call-a", "call_b"] + + +async def test_wake_makes_everything_due_now(tmp_path): + clock = Clock() + up = FakeUploader() + sp = _spool(tmp_path, up, clock=clock) + sp.add(_recording(tmp_path), UploadMeta(call_id="c1"), UploadResult(RETRY, status=401)) + sp.wake() + assert await sp.drain_once() == 1 From a4a575ec9f2ece9322669c91d2d9a15a62c0e933 Mon Sep 17 00:00:00 2001 From: Logan Cusano Date: Tue, 27 Oct 2026 20:16:08 -0400 Subject: [PATCH 4/6] op25: pin gr310 to 5b58335; analog defaults to OP25 noise squelch (node-26#46, node-26#29) - op25-container/Dockerfile: clone gr310 then check out 5b58335 (OP25_COMMIT build arg). That is the gr310 merge tk_drb_conv was verified against for #29 and the first carrying PR #296's NBFM noise/voice squelch. Changing the RUN line also busts CI's cached unpinned clone. - Keys verified from op25_nbfm.py at that commit: nbfm_squelch_mode (power|noise|voice), nbfm_noise_squelch_db, nbfm_squelch_threshold, read from the channel dict. - op25-container: ConfigGenerator.nbfm_squelch_mode (enum, default noise) and nbfm_noise_squelch_db, written onto the analog channel. - edge-node: system config squelch_mode / squelch_db / squelch_threshold -> generate payload (validated, noise when absent); /api/listen and the dashboard's analog pane expose mode and level. #29's CI test of tk_drb_conv against the real tk_p25.py is not done here. Co-Authored-By: Claude Opus 5.5 --- drb-edge-node/app/internal/radio_config.py | 42 +++++- drb-edge-node/app/routers/api.py | 21 ++- drb-edge-node/app/templates/index.html | 28 +++- drb-edge-node/tests/test_conventional.py | 11 ++ drb-edge-node/tests/test_local_dashboard.py | 11 ++ op25-container/Dockerfile | 138 +++++++++--------- op25-container/app/models.py | 19 +++ op25-container/app/routers/op25_controller.py | 2 + op25-container/tests/test_conventional.py | 17 +++ 9 files changed, 218 insertions(+), 71 deletions(-) diff --git a/drb-edge-node/app/internal/radio_config.py b/drb-edge-node/app/internal/radio_config.py index 08b853b..4dc0892 100644 --- a/drb-edge-node/app/internal/radio_config.py +++ b/drb-edge-node/app/internal/radio_config.py @@ -9,7 +9,11 @@ C2 system documents (the `config` blob is passed through untouched): P25 conventional type "P25", config.mode "conventional", config.channels = [{freq: MHz, name, nac?}, ...] (node-26#24) Analog type "NBFM", config.channels = [{freq, name}] — or, from - before #24, config.control_channels[0] + before #24, config.control_channels[0]. Optional squelch + (node-26#46): config.squelch_mode "noise" (default) | + "voice" | "power", config.squelch_db (noise/voice: dB + of quieting to open) and config.squelch_threshold + (power: absolute dB, OP25 default -60) A missing mode is trunked, so every system saved before #24 behaves exactly as it did. Pure functions; main._generate_op25_config does the I/O. @@ -80,6 +84,41 @@ def conventional_channels(config: Dict[str, Any]) -> List[Dict[str, Any]]: return out +SQUELCH_MODES = ("noise", "voice", "power") +DEFAULT_SQUELCH_MODE = "noise" + + +def _optional_float(config: Dict[str, Any], key: str) -> Optional[float]: + value = (config or {}).get(key) + if value is None or (isinstance(value, str) and not value.strip()): + return None + try: + return float(value) + except (TypeError, ValueError): + raise ValueError(f"{key} {value!r} isn't a number.") + + +def analog_squelch(config: Dict[str, Any]) -> Dict[str, Any]: + """config -> the op25-container squelch fields for an analog channel. + + Noise squelch by default (node-26#46): OP25's power squelch compares an + absolute level that depends on dongle, gain and antenna, and left analog + open on static; noise squelch measures FM quieting and needs no + calibration. A config that says nothing gets noise mode. + """ + mode = str((config or {}).get("squelch_mode") or DEFAULT_SQUELCH_MODE).strip().lower() + if mode not in SQUELCH_MODES: + raise ValueError(f"Squelch mode {mode!r} isn't one of {', '.join(SQUELCH_MODES)}.") + out: Dict[str, Any] = {"nbfm_squelch_mode": mode} + db = _optional_float(config, "squelch_db") + if db is not None: + out["nbfm_noise_squelch_db"] = db + threshold = _optional_float(config, "squelch_threshold") + if threshold is not None: + out["nbfm_squelch"] = threshold + return out + + def analog_frequency(config: Dict[str, Any]) -> Optional[int]: chans = (config or {}).get("channels") or [] if chans and isinstance(chans[0], dict) and chans[0].get("freq") not in (None, ""): @@ -117,6 +156,7 @@ def build_generate_payload( if mode == ANALOG: freq = analog_frequency(config) payload["channels"] = [freq] if freq else [] + payload.update(analog_squelch(config)) return payload talkgroups = (config or {}).get("talkgroups", []) or [] diff --git a/drb-edge-node/app/routers/api.py b/drb-edge-node/app/routers/api.py index dddc78e..f0a8fbd 100644 --- a/drb-edge-node/app/routers/api.py +++ b/drb-edge-node/app/routers/api.py @@ -208,6 +208,12 @@ class ListenBody(BaseModel): frequencies: List[float] = [] talkgroups: List[Talkgroup] = [] # p25 only; empty = every talkgroup nac: Optional[str] = None # p25conv only, hex; empty = any NAC + # analog only (node-26#46): OP25 squelch. noise (default) needs no + # calibration; squelch_db = dB of quieting to open (OP25 default 8); + # squelch_threshold = absolute dB for power mode (OP25 default -60). + squelch_mode: Optional[Literal["noise", "voice", "power"]] = None + squelch_db: Optional[float] = None + squelch_threshold: Optional[float] = None # RTL-SDR (R820T/R828D) tuning range. @@ -242,14 +248,21 @@ async def listen(body: ListenBody): else: is_p25 = body.mode == "p25" label = (body.name or "").strip() or (f"P25 {first} MHz" if is_p25 else f"Analog {first} MHz") + sys_config = { + "control_channels": freqs, + "talkgroups": [tg.model_dump() for tg in body.talkgroups] if is_p25 else [], + } + if not is_p25: + sys_config["squelch_mode"] = body.squelch_mode or "noise" + if body.squelch_db is not None: + sys_config["squelch_db"] = body.squelch_db + if body.squelch_threshold is not None: + sys_config["squelch_threshold"] = body.squelch_threshold config = SystemConfig( system_id=f"local-{body.mode}", name=label, type="P25" if is_p25 else "NBFM", - config={ - "control_channels": freqs, - "talkgroups": [tg.model_dump() for tg in body.talkgroups] if is_p25 else [], - }, + config=sys_config, ) await _apply_override(config) return {"ok": True, "name": config.name} diff --git a/drb-edge-node/app/templates/index.html b/drb-edge-node/app/templates/index.html index 8847e2d..39e2333 100644 --- a/drb-edge-node/app/templates/index.html +++ b/drb-edge-node/app/templates/index.html @@ -216,6 +216,20 @@ +
+ + + Noise squelch opens on a real carrier and needs no tuning per site. +
+
+ + + Noise/voice: dB of quieting to open — 8 default, 12-15 for strong signals only, ~5 for weak ones. +