From 00cfca1a57b2c8a810703419eaa7fc5b4d23b27f Mon Sep 17 00:00:00 2001 From: core Date: Sat, 30 May 2026 12:53:02 +0500 Subject: [PATCH] go2rtc: auto-reconcile streams so live/recording self-heal Live and recording both read from the go2rtc restream. If go2rtc's runtime stream table gets emptied (camera clear via API, go2rtc restart with a stale/empty config, or a silently-failed startup sync_all), every restream returns 404 and both live and recording go dark until a manual `docker restart nvr-go2rtc`. Add a background Reconciler that every 30s diffs the desired streams against go2rtc /api/streams and PUTs only the missing ones (existing streams are left untouched so active producers/consumers are not disrupted). VERSION -> 0.0.115. Co-Authored-By: Claude Opus 4.8 (1M context) --- VERSION | 2 +- backend/app/main.py | 3 ++ backend/app/runtime.py | 2 + backend/app/services/go2rtc.py | 73 ++++++++++++++++++++++++++++++++++ 4 files changed, 79 insertions(+), 1 deletion(-) diff --git a/VERSION b/VERSION index 58ca857..f64402b 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.0.114 +0.0.115 diff --git a/backend/app/main.py b/backend/app/main.py index af150e9..660c71c 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -48,9 +48,11 @@ async def lifespan(app: FastAPI): os.makedirs(runtime.config.storage.path, exist_ok=True) await go2rtc.sync_all(runtime.config) # потоки (вкл. _main) в go2rtc для рестрима + runtime.reconciler = go2rtc.Reconciler(runtime.config) # самовосстановление потоков go2rtc runtime.supervisor.start_all() runtime.indexer.start() runtime.retention.start() + runtime.reconciler.start() log.info("NVR запущен: камер=%d, хранилище=%s", len(runtime.config.cameras), runtime.config.storage.path) @@ -58,6 +60,7 @@ async def lifespan(app: FastAPI): yield finally: log.info("остановка NVR…") + await runtime.reconciler.stop() await runtime.indexer.stop() await runtime.retention.stop() await runtime.supervisor.stop_all() diff --git a/backend/app/runtime.py b/backend/app/runtime.py index 84681b2..f8ed490 100644 --- a/backend/app/runtime.py +++ b/backend/app/runtime.py @@ -9,6 +9,7 @@ if TYPE_CHECKING: from .recorder.indexer import Indexer from .recorder.supervisor import Supervisor from .services.retention import Retention + from .services.go2rtc import Reconciler from .api.ws import ConnectionManager @@ -18,6 +19,7 @@ class Runtime: supervisor: "Supervisor" indexer: "Indexer" retention: "Retention" + reconciler: "Reconciler" ws: "ConnectionManager" diff --git a/backend/app/services/go2rtc.py b/backend/app/services/go2rtc.py index 8fa0d95..5721564 100644 --- a/backend/app/services/go2rtc.py +++ b/backend/app/services/go2rtc.py @@ -14,6 +14,7 @@ go2rtc-потоки на камеру: from __future__ import annotations import asyncio +import json import logging import os import urllib.parse @@ -27,6 +28,7 @@ from ..config import Camera, Config log = logging.getLogger("nvr.go2rtc") GO2RTC_YAML = os.environ.get("GO2RTC_CONFIG", "/app/go2rtc.yaml") +RECONCILE_INTERVAL = 30 # сек — период сверки рантайма go2rtc с конфигом def _rtsp_host(config: Config) -> str: @@ -120,3 +122,74 @@ async def sync_all(config: Config) -> None: for cam in config.cameras: for name, src in _streams_for(cam).items(): await _call("PUT", _put(base, name, src)) + + +def _get_stream_names(base: str) -> set[str]: + """Имена потоков, реально загруженных в рантайм go2rtc (/api/streams).""" + req = urllib.request.Request(f"{base}/api/streams", method="GET") + with urllib.request.urlopen(req, timeout=5) as resp: + data = json.loads(resp.read().decode("utf-8") or "{}") + return set(data.keys()) + + +async def reconcile(config: Config) -> int: + """Дотягивает в go2rtc недостающие потоки (самовосстановление). + + go2rtc читает go2rtc.yaml только при старте, а рантайм-таблицу держит через API. + Если её опустошить (DELETE при чистке камер, перезапуск go2rtc со старым/пустым + файлом, молча провалившийся стартовый sync_all) — рестрим начинает отдавать 404, + и live, и запись «слепнут». Здесь сверяем желаемый набор с /api/streams и PUT-им + ТОЛЬКО отсутствующие — существующие не трогаем, чтобы не рвать активных + producer'ов/consumer'ов. Возвращает число восстановленных потоков.""" + if not config.cameras: + return 0 + base = config.live.go2rtc_url.rstrip("/") + try: + existing = await asyncio.to_thread(_get_stream_names, base) + except Exception as exc: # noqa: BLE001 — go2rtc может быть недоступен/перезапускаться + log.debug("go2rtc reconcile: /api/streams недоступен: %s", exc) + return 0 + added = 0 + for cam in config.cameras: + for name, src in _streams_for(cam).items(): + if name not in existing: + await _call("PUT", _put(base, name, src)) + added += 1 + if added: + log.info("go2rtc reconcile: восстановлено потоков %d", added) + write_go2rtc_yaml(config) # и файл приводим в соответствие — для будущего рестарта + return added + + +class Reconciler: + """Фоновая сверка потоков go2rtc с конфигом каждые RECONCILE_INTERVAL секунд. + + Держит ссылку на тот же объект Config, что и остальной рантайм (его .cameras + мутируется in-place при add/clear), поэтому всегда видит актуальный список.""" + + def __init__(self, config: Config, interval: int = RECONCILE_INTERVAL): + self.config = config + self.interval = interval + self._task: asyncio.Task | None = None + self._stopping = False + + def start(self) -> None: + self._stopping = False + self._task = asyncio.create_task(self._loop(), name="go2rtc-reconcile") + + async def stop(self) -> None: + self._stopping = True + if self._task: + self._task.cancel() + try: + await self._task + except asyncio.CancelledError: + pass + + async def _loop(self) -> None: + while not self._stopping: + await asyncio.sleep(self.interval) + try: + await reconcile(self.config) + except Exception: # noqa: BLE001 + log.exception("go2rtc reconcile failed")