logan/node-26
1
0
Fork 0
Code Issues 14 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

call_recorder: spool failed uploads to disk and retry instead of deleting (node-26#44)

Browse Source 
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 <recordings>/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 <noreply@anthropic.com>
This commit is contained in:
Logan Cusano
2026-10-27 20:14:36 -04:00
co-authored by Claude Opus 5.5
parent 1b4fcefbf2
commit 6b884d5433
8 changed files with 752 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
drb-edge-node/app/internal/call_recorder.py
+53 -41
View File
@@ -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
drb-edge-node/app/internal/mqtt_manager.py
+8
View File
@@ -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):
drb-edge-node/app/internal/upload_spool.py
+416
View File
@@ -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:
<spool>/<recording name>.flac
<spool>/<recording name>.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_<call_id>.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