Support fallback NTP servers for edge time sync (#79)

src/sciot/config.py

@@ -529,6 +529,12 @@ def _validate_http_server_transport(
     _host(config, "host", errors, path="communication.http.host")
     _port(config, "port", errors, path="communication.http.port")
     _required_string(config, "ntp_server", errors, path="communication.http.ntp_server")
+    _optional_string_list(
+        config,
+        "ntp_fallback_servers",
+        errors,
+        path="communication.http.ntp_fallback_servers",
+    )
     _model_reference(config, "model", model_registry, errors, path="communication.http.model")
     _validate_endpoints(
         config.get("endpoints"),
@@ -577,6 +583,12 @@ def _validate_websocket_transport(
         errors,
         path="communication.websocket.ntp_server",
     )
+    _optional_string_list(
+        config,
+        "ntp_fallback_servers",
+        errors,
+        path="communication.websocket.ntp_fallback_servers",
+    )
     _model_reference(
         config,
         "model",
@@ -595,6 +607,12 @@ def _validate_mqtt_transport(
     _port(config, "broker_port", errors, path="communication.mqtt.broker_port")
     _required_string(config, "client_id", errors, path="communication.mqtt.client_id")
     _required_string(config, "ntp_server", errors, path="communication.mqtt.ntp_server")
+    _optional_string_list(
+        config,
+        "ntp_fallback_servers",
+        errors,
+        path="communication.mqtt.ntp_fallback_servers",
+    )
     _model_reference(config, "model", model_registry, errors, path="communication.mqtt.model")
 
     topics = config.get("topics")
@@ -852,6 +870,22 @@ def _optional_string_or_none(
         errors.append(f"{path}: must be a non-empty string or null")
 
 
+def _optional_string_list(
+    config: Mapping[str, Any],
+    key: str,
+    errors: list[str],
+    *,
+    path: str,
+):
+    if key not in config or config[key] is None:
+        return
+    value = config[key]
+    if not isinstance(value, list) or not all(
+        isinstance(item, str) and item.strip() for item in value
+    ):
+        errors.append(f"{path}: must be a list of non-empty strings")
+
+
 def _host(
     config: Mapping[str, Any],
     key: str,

src/server/communication/http_server.py

@@ -3,6 +3,7 @@
 from fastapi import FastAPI, HTTPException, Request
 from starlette.middleware.gzip import GZipMiddleware
 from server.communication.inference_protocol import InvalidInferencePayload
+from server.communication.ntp_sync import resync_ntp_fallback, sync_with_ntp_fallback
 from server.logger.log import logger
 import asyncio
 import ntplib
@@ -29,6 +30,7 @@ def __init__(
         last_offloading_layer: int,
         request_handler: "RequestHandler",
         settings: dict[str, Any] | None = None,
+        ntp_fallback_servers: list[str] | None = None,
     ):
         self.app = FastAPI()
         self.settings = settings
@@ -67,6 +69,7 @@ def __init__(
         # Set up NTP client
         self.ntp_client = ntplib.NTPClient()
         self.ntp_server = ntp_server
+        self.ntp_fallback_servers = ntp_fallback_servers or []
         self.offset = self._sync_with_ntp()
         self.start_timestamp = self._get_current_time()
         self._setup_routes()
@@ -80,45 +83,36 @@ def _schedule_resync_ntp(self):
         self._ntp_resync_timer = timer
 
     def _sync_with_ntp(self) -> float:
-        """Synchronize with NTP server at startup with exponential backoff.
+        """Synchronize with the primary NTP server, falling back to any
+        configured `ntp_fallback_servers` if it is unreachable.
 
-        Retries up to 10 times with increasing delays (1s, 2s, 4s, ... capped
-        at 30s).  Falls back to offset=0.0 if NTP is unreachable so the server
-        can still start.
+        Retries up to 10 rounds with increasing delays (1s, 2s, 4s, ... capped
+        at 30s), trying every server each round.  Falls back to offset=0.0 if
+        none of the servers are reachable so the server can still start.
         """
-        max_retries = 10
-        backoff = 1.0
-        for attempt in range(max_retries):
-            try:
-                response = self.ntp_client.request(self.ntp_server)
-                offset = response.offset
-                # Schedule periodic re-sync (every 10 minutes)
-                self._schedule_resync_ntp()
-                return offset
-            except Exception as e:
-                logger.warning(
-                    f"NTP sync attempt {attempt + 1}/{max_retries} failed, "
-                    f"retrying in {backoff:.1f}s: {e}"
-                )
-                time.sleep(backoff)
-                backoff = min(backoff * 2, 30.0)
-
-        logger.warning(
-            "NTP sync failed after all retries -- falling back to offset=0.0"
+        offset, server = sync_with_ntp_fallback(
+            self.ntp_client,
+            [self.ntp_server, *self.ntp_fallback_servers],
+            logger.warning,
         )
-        # Still schedule periodic re-sync so we recover later
+        if server is not None and server != self.ntp_server:
+            logger.warning(f"NTP synced against fallback server {server}")
+        # Schedule periodic re-sync regardless of success/failure so we can
+        # recover later.
         self._schedule_resync_ntp()
-        return 0.0
+        return offset
 
     def _resync_ntp(self):
         """Periodically re-sync with NTP server in the background."""
         if self._stopped:
             return
-        try:
-            response = self.ntp_client.request(self.ntp_server)
-            self.offset = response.offset
-        except Exception as e:
-            logger.warning(f"NTP re-sync failed: {e}")
+        offset = resync_ntp_fallback(
+            self.ntp_client,
+            [self.ntp_server, *self.ntp_fallback_servers],
+            logger.warning,
+        )
+        if offset is not None:
+            self.offset = offset
         # Schedule next re-sync regardless of success/failure
         self._schedule_resync_ntp()
 

src/server/communication/mqtt_client.py

@@ -8,6 +8,7 @@
 from concurrent.futures import ThreadPoolExecutor
 
 from server.logger.log import logger
+from server.communication.ntp_sync import resync_ntp_fallback, sync_with_ntp_fallback
 from server.communication.request_handler import RequestHandler
 from server.communication.inference_protocol import InvalidInferencePayload
 from server.communication.transport_protocol import (
@@ -34,6 +35,7 @@ def __init__(
         input_width: int,
         last_offloading_layer: int,
         request_handler: RequestHandler,
+        ntp_fallback_servers: list[str] | None = None,
     ):
         self.broker_url = broker_url
         self.broker_port = broker_port
@@ -56,6 +58,7 @@ def __init__(
         # Set up NTP client
         self.ntp_client = ntplib.NTPClient()
         self.ntp_server = ntp_server
+        self.ntp_fallback_servers = ntp_fallback_servers or []
         self.offset = self.sync_with_ntp()
         self.start_timestamp = self.get_current_time()
 
@@ -130,42 +133,32 @@ def on_connect(self, client, userdata, flags, reason_code, properties=None):
             logger.debug(f"Connection failed with code {reason_code}")
 
     def sync_with_ntp(self) -> float:
-        """Synchronize with NTP server at startup with exponential backoff.
+        """Synchronize with the primary NTP server, falling back to any
+        configured `ntp_fallback_servers` if it is unreachable.
 
-        Retries up to 10 times with increasing delays (1s, 2s, 4s, ... capped
-        at 30s).  Falls back to offset=0.0 if NTP is unreachable.
+        Retries up to 10 rounds with increasing delays (1s, 2s, 4s, ... capped
+        at 30s), trying every server each round.  Falls back to offset=0.0 if
+        none of the servers are reachable.
         """
-        max_retries = 10
-        backoff = 1.0
-        for attempt in range(max_retries):
-            try:
-                response = self.ntp_client.request(self.ntp_server)
-                offset = response.offset
-                logger.debug(f"Synchronized with NTP server. Offset: {offset} seconds")
-                # Schedule periodic re-sync (every 10 minutes)
-                self._schedule_resync_ntp()
-                return offset
-            except Exception as e:
-                logger.warning(
-                    f"NTP sync attempt {attempt + 1}/{max_retries} failed, "
-                    f"retrying in {backoff:.1f}s: {e}"
-                )
-                time.sleep(backoff)
-                backoff = min(backoff * 2, 30.0)
-
-        logger.warning(
-            "NTP sync failed after all retries -- falling back to offset=0.0"
+        offset, server = sync_with_ntp_fallback(
+            self.ntp_client,
+            [self.ntp_server, *self.ntp_fallback_servers],
+            logger.warning,
         )
+        if server is not None:
+            logger.debug(f"Synchronized with NTP server {server}. Offset: {offset} seconds")
         self._schedule_resync_ntp()
-        return 0.0
+        return offset
 
     def _resync_ntp(self):
         """Periodically re-sync with NTP server in the background."""
-        try:
-            response = self.ntp_client.request(self.ntp_server)
-            self.offset = response.offset
-        except Exception as e:
-            logger.warning(f"NTP re-sync failed: {e}")
+        offset = resync_ntp_fallback(
+            self.ntp_client,
+            [self.ntp_server, *self.ntp_fallback_servers],
+            logger.warning,
+        )
+        if offset is not None:
+            self.offset = offset
         # Schedule next re-sync regardless of success/failure
         self._schedule_resync_ntp()
 

src/server/communication/ntp_sync.py

@@ -0,0 +1,63 @@
+"""Shared NTP sync-with-fallback helpers for the http, websocket and mqtt transports.
+
+A single flaky/unreachable NTP server previously meant every retry attempt
+kept hitting the same host. These helpers cycle through a primary server
+plus optional fallback servers so a reachable server (e.g. a local one) is
+tried before giving up and falling back to offset=0.0.
+"""
+
+import time
+from typing import Callable
+
+import ntplib
+
+
+def sync_with_ntp_fallback(
+    ntp_client: ntplib.NTPClient,
+    servers: list[str],
+    log_warning: Callable[[str], None],
+    *,
+    max_retries: int = 10,
+    initial_backoff: float = 1.0,
+    max_backoff: float = 30.0,
+    sleep: Callable[[float], None] = time.sleep,
+) -> tuple[float, str | None]:
+    """Synchronize with the first reachable server, retrying the whole list.
+
+    Retries up to `max_retries` rounds with increasing delay between rounds
+    (1s, 2s, 4s, ... capped at `max_backoff`), trying every server in
+    `servers` (in order) each round. Returns `(offset, server)` for the
+    first successful sync, or `(0.0, None)` if every server was unreachable
+    on every round.
+    """
+    backoff = initial_backoff
+    for attempt in range(max_retries):
+        for server in servers:
+            try:
+                response = ntp_client.request(server)
+                return response.offset, server
+            except Exception as e:
+                log_warning(
+                    f"NTP sync attempt {attempt + 1}/{max_retries} against "
+                    f"{server} failed: {e}"
+                )
+        sleep(backoff)
+        backoff = min(backoff * 2, max_backoff)
+
+    log_warning("NTP sync failed after all retries -- falling back to offset=0.0")
+    return 0.0, None
+
+
+def resync_ntp_fallback(
+    ntp_client: ntplib.NTPClient,
+    servers: list[str],
+    log_warning: Callable[[str], None],
+) -> float | None:
+    """Try every server once (periodic background re-sync), first success wins."""
+    for server in servers:
+        try:
+            response = ntp_client.request(server)
+            return response.offset
+        except Exception as e:
+            log_warning(f"NTP re-sync against {server} failed: {e}")
+    return None

src/server/communication/websocket_server.py

@@ -7,6 +7,7 @@
     inference_response,
     registration_response,
 )
+from server.communication.ntp_sync import resync_ntp_fallback, sync_with_ntp_fallback
 import ntplib
 import threading
 import time
@@ -25,6 +26,7 @@ def __init__(
         input_width: int,
         last_offloading_layer: int,
         request_handler: RequestHandler,
+        ntp_fallback_servers: list[str] | None = None,
     ):
         self.app = FastAPI()
         self.host = host
@@ -45,6 +47,7 @@ def __init__(
         # Set up NTP client
         self.ntp_client = ntplib.NTPClient()
         self.ntp_server = ntp_server
+        self.ntp_fallback_servers = ntp_fallback_servers or []
         self.offset = self._sync_with_ntp()
         self.start_timestamp = self._get_current_time()
         self._setup_routes()
@@ -58,43 +61,34 @@ def _schedule_resync_ntp(self):
         self._ntp_resync_timer = timer
 
     def _sync_with_ntp(self) -> float:
-        """Synchronize with NTP server at startup with exponential backoff.
+        """Synchronize with the primary NTP server, falling back to any
+        configured `ntp_fallback_servers` if it is unreachable.
 
-        Retries up to 10 times with increasing delays (1s, 2s, 4s, ... capped
-        at 30s).  Falls back to offset=0.0 if NTP is unreachable.
+        Retries up to 10 rounds with increasing delays (1s, 2s, 4s, ... capped
+        at 30s), trying every server each round.  Falls back to offset=0.0 if
+        none of the servers are reachable.
         """
-        max_retries = 10
-        backoff = 1.0
-        for attempt in range(max_retries):
-            try:
-                response = self.ntp_client.request(self.ntp_server)
-                offset = response.offset
-                # Schedule periodic re-sync (every 10 minutes)
-                self._schedule_resync_ntp()
-                return offset
-            except Exception as e:
-                logger.warning(
-                    f"NTP sync attempt {attempt + 1}/{max_retries} failed, "
-                    f"retrying in {backoff:.1f}s: {e}"
-                )
-                time.sleep(backoff)
-                backoff = min(backoff * 2, 30.0)
-
-        logger.warning(
-            "NTP sync failed after all retries -- falling back to offset=0.0"
+        offset, server = sync_with_ntp_fallback(
+            self.ntp_client,
+            [self.ntp_server, *self.ntp_fallback_servers],
+            logger.warning,
         )
+        if server is not None and server != self.ntp_server:
+            logger.warning(f"NTP synced against fallback server {server}")
         self._schedule_resync_ntp()
-        return 0.0
+        return offset
 
     def _resync_ntp(self):
         """Periodically re-sync with NTP server in the background."""
         if self._stopped:
             return
-        try:
-            response = self.ntp_client.request(self.ntp_server)
-            self.offset = response.offset
-        except Exception as e:
-            logger.warning(f"NTP re-sync failed: {e}")
+        offset = resync_ntp_fallback(
+            self.ntp_client,
+            [self.ntp_server, *self.ntp_fallback_servers],
+            logger.warning,
+        )
+        if offset is not None:
+            self.offset = offset
         # Schedule next re-sync regardless of success/failure
         self._schedule_resync_ntp()