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