diff --git a/uv.lock b/uv.lock index 161bae30..d9cfca36 100644 --- a/uv.lock +++ b/uv.lock @@ -418,7 +418,7 @@ requires-dist = [ [[package]] name = "data-hub-watcher" -version = "1.1.0" +version = "1.1.1" source = { editable = "watcher" } dependencies = [ { name = "click" }, diff --git a/watcher/pyproject.toml b/watcher/pyproject.toml index 26f6d541..c490262d 100644 --- a/watcher/pyproject.toml +++ b/watcher/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "data-hub-watcher" -version = "1.1.0" +version = "1.1.1" description = "File-watcher agent for lab instrument PCs that ingests data into Data Hub." readme = "README.md" requires-python = ">=3.12" diff --git a/watcher/src/data_hub_watcher/api_client.py b/watcher/src/data_hub_watcher/api_client.py index 79d55684..95fb4446 100644 --- a/watcher/src/data_hub_watcher/api_client.py +++ b/watcher/src/data_hub_watcher/api_client.py @@ -2,6 +2,7 @@ import logging import os from datetime import datetime, timezone +from email.utils import parsedate_to_datetime from typing import Any import requests @@ -34,11 +35,13 @@ def __init__( message: str, status_code: int = 0, detail: ApiErrorDetail | None = None, + retry_after_seconds: float | None = None, ) -> None: super().__init__(message) self.message = message self.status_code = status_code self.detail = detail + self.retry_after_seconds = retry_after_seconds DEFAULT_TIMEOUT: tuple[float, float] = (5, 30) # (connect, read) seconds @@ -79,6 +82,21 @@ def _url(self, path: str) -> str: def _handle_error(self, resp: requests.Response) -> None: """Parse an error body and raise `ApiError`.""" + retry_after: float | None = None + raw_retry_after = resp.headers.get("Retry-After") + if raw_retry_after: + try: + retry_after = max(0.0, float(raw_retry_after)) + except ValueError: + try: + retry_after = max( + 0.0, + ( + parsedate_to_datetime(raw_retry_after) - datetime.now(timezone.utc) + ).total_seconds(), + ) + except (TypeError, ValueError, OverflowError): + pass detail: ApiErrorDetail | None = None try: body = resp.json() @@ -89,7 +107,12 @@ def _handle_error(self, resp: requests.Response) -> None: msg = resp.text except Exception: msg = resp.text - raise ApiError(msg, status_code=resp.status_code, detail=detail) + raise ApiError( + msg, + status_code=resp.status_code, + detail=detail, + retry_after_seconds=retry_after, + ) def _request( self, diff --git a/watcher/src/data_hub_watcher/monitor.py b/watcher/src/data_hub_watcher/monitor.py index 4025b0d6..d62907d0 100644 --- a/watcher/src/data_hub_watcher/monitor.py +++ b/watcher/src/data_hub_watcher/monitor.py @@ -114,6 +114,10 @@ class FileMonitor: ``MAX_STABILITY_WAIT_SECONDS``. on_stable_file: Called with the `Path` of each stable file. + on_retry_tick: + Optional callback run after each stability check, including checks + with no pending files. Lets transient API reports retry without a new + filesystem event. state_db: Used to skip files that have already been uploaded. recursive: @@ -138,6 +142,7 @@ def __init__( event_reporter: EventReporter | None = None, seed_baseline: bool = False, max_stability_wait_seconds: int = MAX_STABILITY_WAIT_SECONDS, + on_retry_tick: Callable[[], None] | None = None, ) -> None: self._watch_dir = watch_directory self._patterns = file_patterns @@ -150,6 +155,7 @@ def __init__( self._stability_period = stability_period self._max_stability_wait_seconds = max_stability_wait_seconds self._on_stable = on_stable_file + self._on_retry_tick = on_retry_tick self._state_db = state_db self._recursive = recursive # Optional so unit tests that build a FileMonitor in isolation @@ -510,6 +516,12 @@ def _check_pending(self) -> None: error=str(exc), ) + if self._on_retry_tick is not None: + try: + self._on_retry_tick() + except Exception: + logger.exception("Run-report retry tick failed") + def _stat_in_dedup_index( index: dict[str, list[tuple[int, float]]], diff --git a/watcher/src/data_hub_watcher/run_detector.py b/watcher/src/data_hub_watcher/run_detector.py index 1f351eaa..48f05ef2 100644 --- a/watcher/src/data_hub_watcher/run_detector.py +++ b/watcher/src/data_hub_watcher/run_detector.py @@ -10,6 +10,7 @@ import logging import os import re +import time from collections.abc import Callable from dataclasses import dataclass, field from datetime import datetime, timezone @@ -72,6 +73,10 @@ class RunState: # field when a later-stabilising file reveals an earlier creation # time than what the server already knows. acquired_at_sent: float | None = None + # Transient report/PATCH failures need an in-process retry even when no + # other file arrives. The monitor's stability tick services this timer. + retry_at: float | None = None + retry_attempts: int = 0 def _run_acquired_at(files: list[FileInfo]) -> float | None: @@ -185,6 +190,37 @@ def on_stable_file(self, path: Path) -> None: else: self._update_run(run) + def retry_failed_reports(self) -> None: + """Retry one due run per monitor tick, independent of new file events.""" + now = time.monotonic() + for run in self._runs.values(): + if run.retry_at is None or run.retry_at > now: + continue + # Clear the due timer before sending; a failure schedules the next + # attempt, while a success leaves the run out of the retry queue. + run.retry_at = None + if run.reported: + self._update_run(run) + else: + self._report_new_run(run) + return + + @staticmethod + def _schedule_retry(run: RunState, exc: ApiError) -> None: + # 4xx validation/auth failures need operator action. 429, transport + # failures (status 0) and 5xx are transient. POST is safe to replay: + # the server upserts on (instrument_id, run_id). + if exc.status_code not in (0, 429) and exc.status_code < 500: + run.retry_at = None + return + run.retry_attempts += 1 + delay = min(5 * 2 ** min(run.retry_attempts - 1, 6), 300) + if exc.status_code == 429: + delay = max(delay, 60) + if exc.retry_after_seconds is not None: + delay = max(delay, exc.retry_after_seconds) + run.retry_at = time.monotonic() + delay + # ------------------------------------------------------------------ # run-ID extraction # ------------------------------------------------------------------ @@ -272,6 +308,8 @@ def _report_new_run(self, run: RunState) -> None: try: resp = self._client.report_run(self._instrument_id, payload) run.reported = True + run.retry_at = None + run.retry_attempts = 0 run.api_run_id = resp.id run.patched_file_count = len(run.files) run.acquired_at_sent = acquired @@ -288,7 +326,13 @@ def _report_new_run(self, run: RunState) -> None: ) logger.info("Reported new run %s (%d files)", run.run_id, len(run.files)) except ApiError as exc: - logger.warning("Failed to report run %s: %s (will retry)", run.run_id, exc.message) + self._schedule_retry(run, exc) + logger.warning( + "Failed to report run %s: %s%s", + run.run_id, + exc.message, + " (retry scheduled)" if run.retry_at is not None else "", + ) self._counters.errors += 1 self._reporter.report_error( "run_report_failed", @@ -345,6 +389,8 @@ def _update_run(self, run: RunState) -> None: ).isoformat() try: self._client.update_run(self._instrument_id, run.run_id, payload) + run.retry_at = None + run.retry_attempts = 0 run.patched_file_count = len(run.files) if earlier_acquired is not None: run.acquired_at_sent = earlier_acquired @@ -357,6 +403,7 @@ def _update_run(self, run: RunState) -> None: ) except ApiError as exc: logger.warning("Failed to update run %s: %s", run.run_id, exc.message) + self._schedule_retry(run, exc) self._counters.errors += 1 self._reporter.report_error( "run_report_failed", diff --git a/watcher/src/data_hub_watcher/runtime.py b/watcher/src/data_hub_watcher/runtime.py index 966a7c80..dec64c89 100644 --- a/watcher/src/data_hub_watcher/runtime.py +++ b/watcher/src/data_hub_watcher/runtime.py @@ -297,6 +297,7 @@ def _request_upgrade_restart(target_version: str) -> None: stability_period=inst.stability_period_seconds, max_stability_wait_seconds=inst.max_stability_wait_seconds, on_stable_file=detector.on_stable_file, + on_retry_tick=detector.retry_failed_reports, state_db=state_db, recursive=inst.run_detection.recursive, event_reporter=reporter, diff --git a/watcher/tests/test_run_report_retries.py b/watcher/tests/test_run_report_retries.py new file mode 100644 index 00000000..825d5947 --- /dev/null +++ b/watcher/tests/test_run_report_retries.py @@ -0,0 +1,115 @@ +"""Transient run-report failures retry without another file event.""" + +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import MagicMock + +import pytest +import requests + +from data_hub_watcher.api_client import ApiError, DataHubClient +from data_hub_watcher.monitor import FileMonitor +from data_hub_watcher.run_detector import RunDetector + + +def _detector(tmp_path: Path, client: MagicMock) -> RunDetector: + return RunDetector( + pattern=r"^(R1)_", + instrument_id="synthetic-instrument", + watcher_id="synthetic-watcher", + client=client, + state_db=MagicMock(), + event_reporter=MagicMock(), + counters=MagicMock(), + watch_directory=tmp_path, + ) + + +def test_transient_post_retries_without_new_file(tmp_path: Path, monkeypatch) -> None: + clock = [100.0] + monkeypatch.setattr("data_hub_watcher.run_detector.time.monotonic", lambda: clock[0]) + file = tmp_path / "R1_a.csv" + file.write_text("a") + client = MagicMock() + client.report_run.side_effect = [ApiError("temporary", 503), SimpleNamespace(id="run")] + detector = _detector(tmp_path, client) + + detector.on_stable_file(file) + detector.on_stable_file(file) # duplicate callbacks must not be needed + clock[0] = 104.0 + detector.retry_failed_reports() + assert client.report_run.call_count == 1 + + clock[0] = 105.0 + detector.retry_failed_reports() + assert client.report_run.call_count == 2 + assert detector._runs["R1"].reported + detector.retry_failed_reports() + assert client.report_run.call_count == 2 + + +def test_transient_patch_retries_backlog(tmp_path: Path, monkeypatch) -> None: + clock = [100.0] + monkeypatch.setattr("data_hub_watcher.run_detector.time.monotonic", lambda: clock[0]) + client = MagicMock() + client.report_run.return_value = SimpleNamespace(id="run") + client.update_run.side_effect = [ApiError("temporary", 503), MagicMock()] + detector = _detector(tmp_path, client) + first = tmp_path / "R1_a.csv" + second = tmp_path / "R1_b.csv" + first.write_text("a") + second.write_text("b") + + detector.on_stable_file(first) + detector.on_stable_file(second) + assert client.update_run.call_count == 1 + clock[0] = 105.0 + detector.retry_failed_reports() + assert client.update_run.call_count == 2 + assert detector._runs["R1"].patched_file_count == 2 + + +def test_permanent_error_does_not_schedule_retry(tmp_path: Path, monkeypatch) -> None: + monkeypatch.setattr("data_hub_watcher.run_detector.time.monotonic", lambda: 100.0) + file = tmp_path / "R1_a.csv" + file.write_text("a") + client = MagicMock() + client.report_run.side_effect = ApiError("invalid watcher", 400) + detector = _detector(tmp_path, client) + detector.on_stable_file(file) + detector.retry_failed_reports() + assert client.report_run.call_count == 1 + assert detector._runs["R1"].retry_at is None + + +def test_rate_limit_honors_retry_after(tmp_path: Path, monkeypatch) -> None: + monkeypatch.setattr("data_hub_watcher.run_detector.time.monotonic", lambda: 100.0) + response = requests.Response() + response.status_code = 429 + response._content = b"busy" + response.headers["Retry-After"] = "120" + with pytest.raises(ApiError) as raised: + DataHubClient("https://example.test")._handle_error(response) + assert raised.value.retry_after_seconds == 120 + + file = tmp_path / "R1_a.csv" + file.write_text("a") + client = MagicMock() + client.report_run.side_effect = raised.value + detector = _detector(tmp_path, client) + detector.on_stable_file(file) + assert detector._runs["R1"].retry_at == 220 + + +def test_monitor_ticks_retries_even_with_no_pending_file(tmp_path: Path) -> None: + retry = MagicMock() + monitor = FileMonitor( + watch_directory=tmp_path, + file_patterns=["*.csv"], + stability_period=1, + on_stable_file=MagicMock(), + on_retry_tick=retry, + state_db=MagicMock(), + ) + monitor._check_pending() + retry.assert_called_once_with()