From bda15cc8fe102887ed51c6ff99c66bff3cc8a647 Mon Sep 17 00:00:00 2001 From: ff225 Date: Wed, 22 Jul 2026 11:30:39 +0200 Subject: [PATCH] feat(server): support fallback NTP servers for edge time sync (#79) A single remote NTP server (e.g. the public pool) can introduce synchronization errors that hurt telemetry accuracy. All three transports (http, websocket, mqtt) now try an optional ordered list of fallback servers -- e.g. a local NTP server first -- before giving up on sync, sharing the retry/backoff logic via a new server.communication.ntp_sync module. Closes #79 --- src/sciot/config.py | 34 ++++++++ src/server/communication/http_server.py | 54 ++++++------ src/server/communication/mqtt_client.py | 51 +++++------ src/server/communication/ntp_sync.py | 63 ++++++++++++++ src/server/communication/websocket_server.py | 50 +++++------ src/server/edge/run_edge.py | 3 + src/server/settings.yaml | 6 ++ tests/unit/test_config_validation.py | 30 +++++++ tests/unit/test_ntp_sync.py | 91 ++++++++++++++++++++ 9 files changed, 295 insertions(+), 87 deletions(-) create mode 100644 src/server/communication/ntp_sync.py create mode 100644 tests/unit/test_ntp_sync.py diff --git a/src/sciot/config.py b/src/sciot/config.py index 25a2503..39f7fcd 100644 --- a/src/sciot/config.py +++ b/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, diff --git a/src/server/communication/http_server.py b/src/server/communication/http_server.py index 6517ce2..8505323 100644 --- a/src/server/communication/http_server.py +++ b/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() diff --git a/src/server/communication/mqtt_client.py b/src/server/communication/mqtt_client.py index b254c3f..04b6463 100644 --- a/src/server/communication/mqtt_client.py +++ b/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() diff --git a/src/server/communication/ntp_sync.py b/src/server/communication/ntp_sync.py new file mode 100644 index 0000000..92a5db2 --- /dev/null +++ b/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 diff --git a/src/server/communication/websocket_server.py b/src/server/communication/websocket_server.py index 6f9dd86..26eed71 100644 --- a/src/server/communication/websocket_server.py +++ b/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() diff --git a/src/server/edge/run_edge.py b/src/server/edge/run_edge.py index 1bf0741..98d1152 100644 --- a/src/server/edge/run_edge.py +++ b/src/server/edge/run_edge.py @@ -42,6 +42,7 @@ def create_enabled_transports( port=websocket_config["port"], endpoint=websocket_config["endpoint"], ntp_server=websocket_config["ntp_server"], + ntp_fallback_servers=websocket_config.get("ntp_fallback_servers"), input_height=model_config["input_height"], input_width=model_config["input_width"], last_offloading_layer=model_config["last_offloading_layer"], @@ -60,6 +61,7 @@ def create_enabled_transports( port=http_config["port"], endpoints=http_config["endpoints"], ntp_server=http_config["ntp_server"], + ntp_fallback_servers=http_config.get("ntp_fallback_servers"), input_height=model_config["input_height"], input_width=model_config["input_width"], last_offloading_layer=model_config["last_offloading_layer"], @@ -81,6 +83,7 @@ def create_enabled_transports( protocol=mqtt.MQTTv311, subscribed_topics=mqtt_config["topics"], ntp_server=mqtt_config["ntp_server"], + ntp_fallback_servers=mqtt_config.get("ntp_fallback_servers"), input_height=model_config["input_height"], input_width=model_config["input_width"], last_offloading_layer=model_config["last_offloading_layer"], diff --git a/src/server/settings.yaml b/src/server/settings.yaml index b8d5e69..4611f92 100644 --- a/src/server/settings.yaml +++ b/src/server/settings.yaml @@ -10,6 +10,8 @@ communication: host: 0.0.0.0 model: fomo_96x96 ntp_server: 0.it.pool.ntp.org + # ntp_fallback_servers: + # - 1.it.pool.ntp.org port: 23456 timeout_keep_alive: 30 backlog: 2048 @@ -22,6 +24,8 @@ communication: client_id: edge model: fomo_96x96 ntp_server: 0.it.pool.ntp.org + # ntp_fallback_servers: + # - 1.it.pool.ntp.org topics: device_inference_result: device_01/model_inference_result device_input: device_01/input_data @@ -32,6 +36,8 @@ communication: host: 0.0.0.0 model: fomo_96x96 ntp_server: 0.it.pool.ntp.org + # ntp_fallback_servers: + # - 1.it.pool.ntp.org port: 8080 offloading_algo: diff --git a/tests/unit/test_config_validation.py b/tests/unit/test_config_validation.py index a12e088..2295029 100644 --- a/tests/unit/test_config_validation.py +++ b/tests/unit/test_config_validation.py @@ -192,6 +192,26 @@ def test_documented_example_configs_are_valid(path, loader): ), "offloading.adaptive_risk.device_load_weight: must be greater than or equal to 0", ), + ( + lambda cfg: cfg["communication"]["http"].update( + ntp_fallback_servers="0.pool.ntp.org" + ), + "communication.http.ntp_fallback_servers: must be a list of non-empty strings", + ), + ( + lambda cfg: ( + cfg["communication"]["mode"].append("mqtt"), + cfg["communication"]["mqtt"].update(ntp_fallback_servers=[""]), + ), + "communication.mqtt.ntp_fallback_servers: must be a list of non-empty strings", + ), + ( + lambda cfg: ( + cfg["communication"]["mode"].append("websocket"), + cfg["communication"]["websocket"].update(ntp_fallback_servers=[1]), + ), + "communication.websocket.ntp_fallback_servers: must be a list of non-empty strings", + ), ], ) def test_invalid_server_config_reports_field_errors(mutate, expected_error): @@ -202,6 +222,16 @@ def test_invalid_server_config_reports_field_errors(mutate, expected_error): validate_server_config(config) +def test_server_config_accepts_optional_ntp_fallback_servers(): + config = _server_config() + config["communication"]["http"]["ntp_fallback_servers"] = [ + "1.it.pool.ntp.org", + "2.it.pool.ntp.org", + ] + + validate_server_config(config) + + def test_offloading_config_accepts_static_and_adaptive_risk(): config = _server_config() diff --git a/tests/unit/test_ntp_sync.py b/tests/unit/test_ntp_sync.py new file mode 100644 index 0000000..ddbfa7a --- /dev/null +++ b/tests/unit/test_ntp_sync.py @@ -0,0 +1,91 @@ +"""Tests for the shared NTP sync-with-fallback helpers (SCIOT-79).""" + +from unittest.mock import Mock + +from server.communication.ntp_sync import resync_ntp_fallback, sync_with_ntp_fallback + + +def _response(offset): + response = Mock() + response.offset = offset + return response + + +def test_sync_uses_primary_server_when_reachable(): + client = Mock() + client.request.return_value = _response(1.5) + warnings = [] + + offset, server = sync_with_ntp_fallback( + client, ["primary.example", "fallback.example"], warnings.append + ) + + assert (offset, server) == (1.5, "primary.example") + client.request.assert_called_once_with("primary.example") + assert warnings == [] + + +def test_sync_falls_back_to_next_server_when_primary_unreachable(): + client = Mock() + client.request.side_effect = [OSError("unreachable"), _response(2.5)] + warnings = [] + + offset, server = sync_with_ntp_fallback( + client, ["primary.example", "fallback.example"], warnings.append + ) + + assert (offset, server) == (2.5, "fallback.example") + assert client.request.call_args_list == [ + (("primary.example",),), + (("fallback.example",),), + ] + assert len(warnings) == 1 + + +def test_sync_returns_zero_offset_when_every_server_unreachable(): + client = Mock() + client.request.side_effect = OSError("unreachable") + warnings = [] + sleeps = [] + + offset, server = sync_with_ntp_fallback( + client, + ["primary.example", "fallback.example"], + warnings.append, + max_retries=2, + initial_backoff=0.01, + sleep=sleeps.append, + ) + + assert (offset, server) == (0.0, None) + # Two rounds x two servers = 4 attempts, plus 2 sleeps between rounds. + assert client.request.call_count == 4 + assert sleeps == [0.01, 0.02] + assert "falling back to offset=0.0" in warnings[-1] + + +def test_resync_tries_each_server_once_and_returns_first_success(): + client = Mock() + client.request.side_effect = [OSError("unreachable"), _response(3.0)] + warnings = [] + + offset = resync_ntp_fallback( + client, ["primary.example", "fallback.example"], warnings.append + ) + + assert offset == 3.0 + assert client.request.call_count == 2 + assert len(warnings) == 1 + + +def test_resync_returns_none_when_every_server_unreachable(): + client = Mock() + client.request.side_effect = OSError("unreachable") + warnings = [] + + offset = resync_ntp_fallback( + client, ["primary.example", "fallback.example"], warnings.append + ) + + assert offset is None + assert client.request.call_count == 2