diff --git a/drb-edge-node/app/config.py b/drb-edge-node/app/config.py index 8d28973..0bcd22e 100644 --- a/drb-edge-node/app/config.py +++ b/drb-edge-node/app/config.py @@ -32,6 +32,9 @@ class Settings(BaseSettings): config_path: str = "/configs" recordings_path: str = "/recordings" + # Offline call buffer — how many call_end events to keep while disconnected + offline_call_buffer_size: int = 35 + class Config: env_file = ".env" diff --git a/drb-edge-node/app/internal/mqtt_manager.py b/drb-edge-node/app/internal/mqtt_manager.py index 19cf452..294bd4a 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 from datetime import datetime, timezone +from collections import deque from typing import Optional, Callable, Awaitable, Dict, Any import paho.mqtt.client as mqtt from app.config import settings @@ -23,6 +24,8 @@ class MQTTManager: self.on_config_push: Optional[ConfigCallback] = None self.on_api_key: Optional[ApiKeyCallback] = None + self._offline_buffer = deque(maxlen=settings.offline_call_buffer_size) + nid = settings.node_id self._t_checkin = f"nodes/{nid}/checkin" self._t_status = f"nodes/{nid}/status" @@ -64,6 +67,7 @@ class MQTTManager: logger.info("MQTT connected.") asyncio.run_coroutine_threadsafe(self._publish_checkin(), self._loop) asyncio.run_coroutine_threadsafe(self._maybe_request_key(), self._loop) + asyncio.run_coroutine_threadsafe(self._flush_offline_buffer(), self._loop) else: logger.error(f"MQTT connect refused: {reason_code}") @@ -130,7 +134,23 @@ class MQTTManager: "timestamp": datetime.now(timezone.utc).isoformat(), **data, } - self._publish(self._t_metadata, payload, qos=1) + if not self._connected: + if event_type == "call_end": + self._offline_buffer.append((self._t_metadata, payload)) + logger.warning(f"MQTT offline. Buffered call_end event for {data.get('call_id')}") + else: + logger.debug(f"MQTT offline. Dropping metadata event: {event_type}") + else: + self._publish(self._t_metadata, payload, qos=1) + + async def _flush_offline_buffer(self): + if not self._offline_buffer: + return + count = len(self._offline_buffer) + logger.info(f"Relaying {count} buffered call_end events from offline queue.") + while self._offline_buffer: + topic, payload = self._offline_buffer.popleft() + self._publish(topic, payload, qos=1) async def _maybe_request_key(self): """After connecting, wait for any retained api_key message to arrive. diff --git a/drb-edge-node/app/models.py b/drb-edge-node/app/models.py index 77f668d..441efa7 100644 --- a/drb-edge-node/app/models.py +++ b/drb-edge-node/app/models.py @@ -37,6 +37,7 @@ class NodeConfig(BaseModel): enforce_override_timeout: bool = True override_system_id: Optional[str] = None override_config: Optional[SystemConfig] = None + offline_call_buffer_size: int = 35 # max call_end events to buffer while MQTT is offline class CallEvent(BaseModel):