Retry transient watcher run reports without new file events

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"

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,

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]]],

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",

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,

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()