Make max stability wait configurable

developer-docs/watcher.md

@@ -65,7 +65,7 @@ Starts the file monitoring loop. Before entering the loop it:
 
 While running:
 
-- **File monitor** watches the directory for new/modified files using `watchdog` and waits for each file to stabilize (size + mtime unchanged for the configured stability period). Files that keep changing for longer than 5 minutes are abandoned and surface as a `stability_timeout` error event.
+- **File monitor** watches the directory for new/modified files using `watchdog` and waits for each file to stabilize (size + mtime unchanged for the configured stability period). Files that keep changing past `max_stability_wait_seconds` (default 5 minutes) are abandoned and surface as a `stability_timeout` error event.
 - **Run detector** groups stable files into runs by applying the configured regex to each file's relative path. The first file for a run triggers `POST /instruments/:id/runs`; subsequent files for the same run incrementally `PATCH` only the new entries onto the manifest. Files inside the watch tree that don't match the pattern emit a `pattern_mismatch` event (throttled to one per parent directory) so misconfigured patterns surface in the dashboard.
 - **Uploader** requests a presigned S3 URL from the API and uploads each file via HTTP PUT (auto mode), or processes the server's upload queue (manual mode). The watcher does not need AWS credentials. Each upload retries up to 3 times with exponential backoff (1, 2, 4 s) and is recorded locally with its SHA-256 so retries and restarts don't re-upload the same bytes. In manual mode, queue-poll failures are throttled (1st failure, then every 10th) to keep a sustained outage visible without flooding the events stream.
 - **Upload worker** (manual mode only) polls the server's upload queue on its own long-lived thread every 60 seconds, decoupled from the heartbeat so a slow or large upload can't delay heartbeats and make a busy watcher look offline. On shutdown it is stopped and joined before the state DB is closed. Auto mode has no worker: uploads run on the monitor's stability-checker thread via the run detector's upload callback.
@@ -162,6 +162,7 @@ instrument:
   enabled: true
   upload_mode: auto              # "auto" or "manual"
   stability_period_seconds: 5    # 1–300
+  max_stability_wait_seconds: 300  # 1–86400; must be >= stability_period_seconds
   run_detection:
     pattern: '^([^/]+)/'         # regex with one capture group (run ID)
     recursive: true
@@ -281,7 +282,7 @@ The watcher's primary observability surface is the per-watcher event log served
 | --- | --- |
 | `run_report_failed` | POST/PATCH against `/instruments/:id/runs[/:run_id]` failed. `details` includes `operation`, `status_code`, `file_count`. |
 | `config_sync_failed` | The startup `PUT /watchers/:id/config` (or its checksum probe) failed. |
-| `stability_timeout` | A file kept changing past 5 minutes and was abandoned. |
+| `stability_timeout` | A file kept changing past the configured max wait (default 5 minutes) and was abandoned. `details.max_wait_seconds` reports the cap used. |
 | `stable_callback_failed` | The on-stable-file callback raised. |
 | `pattern_mismatch` | A file inside the watch tree did not match `run_detection.pattern`. Throttled to one emission per parent directory per process. |
 | `events_dropped` | Synthetic event prepended after one or more prior batches were dropped. `details.dropped_count` reports the gap size. |

watcher/pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "data-hub-watcher"
-version = "0.5.3"
+version = "0.5.4"
 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/events.py

@@ -19,8 +19,9 @@
   "status_code", "error", "file_count"}``.
 * ``config_sync_failed`` -- the startup ``PUT /watchers/:id/config``
   call raised ``ApiError``. ``details = {"kind", "checksum", "error"}``.
-* ``stability_timeout`` -- a file kept changing past
-  ``MAX_STABILITY_WAIT_SECONDS`` and was abandoned.
+* ``stability_timeout`` -- a file kept changing past the configured
+  ``max_stability_wait_seconds`` (default ``MAX_STABILITY_WAIT_SECONDS``)
+  and was abandoned.
   ``details = {"kind", "path", "max_wait_seconds"}``.
 * ``stable_callback_failed`` -- the on-stable-file callback raised.
   ``details = {"kind", "path", "error"}``.

watcher/src/data_hub_watcher/models.py

@@ -8,6 +8,8 @@
 
 from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
 
+from data_hub_watcher.constants import MAX_STABILITY_WAIT_SECONDS
+
 logger = logging.getLogger(__name__)
 
 
@@ -41,6 +43,11 @@ class InstrumentConfig(BaseModel):
     enabled: bool = True
     upload_mode: Literal["auto", "manual"] = "auto"
     stability_period_seconds: int = Field(default=5, ge=1, le=300)
+    # Cap on how long a still-changing file may stay pending before the
+    # monitor abandons it with a stability_timeout event. Defaults to
+    # MAX_STABILITY_WAIT_SECONDS so existing YAMLs keep today's 5-minute
+    # behaviour; raise for instruments that write large files for longer.
+    max_stability_wait_seconds: int = Field(default=MAX_STABILITY_WAIT_SECONDS, ge=1, le=86400)
     # Maximum number of files to upload concurrently for a single
     # detected run (auto mode) or queue poll. The default of 4 balances
     # S3 throughput against lab-PC network and CPU budgets; instruments
@@ -68,6 +75,19 @@ def _validate_directory(cls, v: Path) -> Path:
             raise ValueError(f"Watch directory is not a directory: {expanded}")
         return expanded
 
+    @model_validator(mode="after")
+    def _max_wait_covers_stability_period(self) -> InstrumentConfig:
+        # A file must stay unchanged for stability_period_seconds before
+        # it can become stable; if the abandon cap is shorter, every file
+        # that takes any time to finish writing would time out first.
+        if self.max_stability_wait_seconds < self.stability_period_seconds:
+            raise ValueError(
+                "max_stability_wait_seconds "
+                f"({self.max_stability_wait_seconds}) must be >= "
+                f"stability_period_seconds ({self.stability_period_seconds})"
+            )
+        return self
+
 
 class WatcherConfig(BaseModel):
     version: Literal[1]

watcher/src/data_hub_watcher/monitor.py

@@ -108,6 +108,10 @@ class FileMonitor:
     stability_period:
         Seconds a file's size + mtime must remain unchanged before it is
         considered stable.
+    max_stability_wait_seconds:
+        Seconds from first sighting after which a still-changing file is
+        abandoned with a ``stability_timeout`` event. Defaults to
+        ``MAX_STABILITY_WAIT_SECONDS``.
     on_stable_file:
         Called with the `Path` of each stable file.
     state_db:
@@ -133,6 +137,7 @@ def __init__(
         recursive: bool = False,
         event_reporter: EventReporter | None = None,
         seed_baseline: bool = False,
+        max_stability_wait_seconds: int = MAX_STABILITY_WAIT_SECONDS,
     ) -> None:
         self._watch_dir = watch_directory
         self._patterns = file_patterns
@@ -143,6 +148,7 @@ def __init__(
         # per file; this is one regex match instead.
         self._matches_name = _compile_pattern_matcher(file_patterns)
         self._stability_period = stability_period
+        self._max_stability_wait_seconds = max_stability_wait_seconds
         self._on_stable = on_stable_file
         self._state_db = state_db
         self._recursive = recursive
@@ -446,12 +452,13 @@ def _check_pending(self) -> None:
 
         A file is "stable" when its size and mtime haven't changed for the full
         stability period.  If the file keeps changing beyond
-        MAX_STABILITY_WAIT_SECONDS it is abandoned — this guards against files
-        that are continuously appended to (e.g. active log streams).
+        ``max_stability_wait_seconds`` it is abandoned — this guards against
+        files that are continuously appended to (e.g. active log streams).
         """
         now = time.monotonic()
         stable: list[Path] = []
         timed_out: list[Path] = []
+        max_wait = self._max_stability_wait_seconds
 
         with self._lock:
             for path, pf in list(self._pending.items()):
@@ -471,22 +478,22 @@ def _check_pending(self) -> None:
                 if elapsed_since_change >= self._stability_period:
                     stable.append(path)
                     del self._pending[path]
-                elif (now - pf.first_seen) >= MAX_STABILITY_WAIT_SECONDS:
+                elif (now - pf.first_seen) >= max_wait:
                     timed_out.append(path)
                     del self._pending[path]
 
         for path in timed_out:
             logger.error(
                 "File %s did not stabilise within %ds — skipping",
                 path,
-                MAX_STABILITY_WAIT_SECONDS,
+                max_wait,
             )
             if self._reporter is not None:
                 self._reporter.report_error(
                     "stability_timeout",
-                    f"File {path.name} did not stabilise within {MAX_STABILITY_WAIT_SECONDS}s",
+                    f"File {path.name} did not stabilise within {max_wait}s",
                     path=str(path),
-                    max_wait_seconds=MAX_STABILITY_WAIT_SECONDS,
+                    max_wait_seconds=max_wait,
                 )
 
         for path in stable:

watcher/src/data_hub_watcher/runtime.py

@@ -295,6 +295,7 @@ def _request_upgrade_restart(target_version: str) -> None:
         watch_directory=inst.watch_directory,
         file_patterns=inst.file_patterns,
         stability_period=inst.stability_period_seconds,
+        max_stability_wait_seconds=inst.max_stability_wait_seconds,
         on_stable_file=detector.on_stable_file,
         state_db=state_db,
         recursive=inst.run_detection.recursive,

watcher/tests/test_monitor_events.py

@@ -3,13 +3,13 @@
 Two anomalies that previously only logged locally now surface as
 ``EventType.ERROR`` events:
 
-* ``kind=stability_timeout`` -- a file kept changing past
-  ``MAX_STABILITY_WAIT_SECONDS`` and was abandoned.
+* ``kind=stability_timeout`` -- a file kept changing past the
+  configured ``max_stability_wait_seconds`` and was abandoned.
 * ``kind=stable_callback_failed`` -- the on-stable-file callback raised.
 
-Tests drive ``_check_pending`` directly (with monkey-patched ``time``
-and a forced-out-of-window ``first_seen``) so we don't have to wait
-the real 5-minute stability window in unit tests.
+Tests drive ``_check_pending`` directly (with a forced-out-of-window
+``first_seen``) so we don't have to wait the real stability window in
+unit tests.
 """
 
 from __future__ import annotations
@@ -20,6 +20,7 @@
 
 import pytest
 
+from data_hub_watcher.constants import MAX_STABILITY_WAIT_SECONDS
 from data_hub_watcher.events import EventReporter
 from data_hub_watcher.monitor import FileMonitor, _PendingFile
 from data_hub_watcher.state import StateDB
@@ -45,11 +46,13 @@ def _make_monitor(
     *,
     on_stable_file: MagicMock | None = None,
     reporter: MagicMock | None = None,
+    max_stability_wait_seconds: int = MAX_STABILITY_WAIT_SECONDS,
 ) -> FileMonitor:
     return FileMonitor(
         watch_directory=watch_dir,
         file_patterns=["*.csv"],
         stability_period=1,
+        max_stability_wait_seconds=max_stability_wait_seconds,
         on_stable_file=on_stable_file or MagicMock(),
         state_db=state_db,
         recursive=False,
@@ -59,8 +62,6 @@ def _make_monitor(
 
 class TestStabilityTimeout:
     def test_emits_event_for_abandoned_file(self, watch_dir: Path, state_db: StateDB) -> None:
-        from data_hub_watcher.constants import MAX_STABILITY_WAIT_SECONDS
-
         reporter = MagicMock(spec=EventReporter)
         monitor = _make_monitor(watch_dir, state_db, reporter=reporter)
 
@@ -69,8 +70,8 @@ def test_emits_event_for_abandoned_file(self, watch_dir: Path, state_db: StateDB
         st = f.stat()
 
         # Manually plant a pending entry whose first_seen is older than
-        # MAX_STABILITY_WAIT_SECONDS so _check_pending classifies it as
-        # timed out without waiting 5 minutes in the test.
+        # the default max wait so _check_pending classifies it as timed
+        # out without waiting in the test.
         now = time.monotonic()
         monitor._pending[f] = _PendingFile(
             path=f,
@@ -91,15 +92,42 @@ def test_emits_event_for_abandoned_file(self, watch_dir: Path, state_db: StateDB
         assert call.kwargs["path"] == str(f)
         assert call.kwargs["max_wait_seconds"] == MAX_STABILITY_WAIT_SECONDS
 
+    def test_uses_configured_max_wait(self, watch_dir: Path, state_db: StateDB) -> None:
+        reporter = MagicMock(spec=EventReporter)
+        custom_max_wait = 10
+        monitor = _make_monitor(
+            watch_dir,
+            state_db,
+            reporter=reporter,
+            max_stability_wait_seconds=custom_max_wait,
+        )
+
+        f = watch_dir / "growing.csv"
+        f.write_text("a")
+        st = f.stat()
+        now = time.monotonic()
+        monitor._pending[f] = _PendingFile(
+            path=f,
+            size=st.st_size,
+            mtime=st.st_mtime,
+            first_seen=now - custom_max_wait - 1,
+            last_changed=now,
+        )
+
+        monitor._check_pending()
+
+        assert f not in monitor._pending
+        call = reporter.report_error.call_args
+        assert call.args[0] == "stability_timeout"
+        assert call.kwargs["max_wait_seconds"] == custom_max_wait
+
     def test_no_reporter_does_not_crash(self, watch_dir: Path, state_db: StateDB) -> None:
         """A FileMonitor built without a reporter must still function.
 
         Tests that build FileMonitor in isolation pass reporter=None.
         The timeout path used to log only; it should keep doing so
         without raising when the reporter is absent.
         """
-        from data_hub_watcher.constants import MAX_STABILITY_WAIT_SECONDS
-
         monitor = _make_monitor(watch_dir, state_db, reporter=None)
         f = watch_dir / "growing.csv"
         f.write_text("a")

watcher/tests/test_run_detection_config.py

@@ -1,10 +1,9 @@
 """Unit tests for ``RunDetectionConfig`` validation.
 
-Also covers the ``InstrumentConfig.upload_parallelism`` field added
-alongside the parallel-upload optimisation -- it lives here rather
-than in its own file because the existing ``RunDetectionConfig``
-suite is the closest neighbour and the field is a small additive
-concern.
+Also covers small additive ``InstrumentConfig`` fields
+(``upload_parallelism``, ``max_stability_wait_seconds``) -- they live
+here rather than in their own files because this suite is the closest
+neighbour.
 """
 
 from __future__ import annotations
@@ -13,6 +12,7 @@
 import pytest
 from pydantic import ValidationError
 
+from data_hub_watcher.constants import MAX_STABILITY_WAIT_SECONDS
 from data_hub_watcher.models import InstrumentConfig, RunDetectionConfig
 
 
@@ -54,6 +54,8 @@ def _make_instrument(
     tmp_path: Path,
     *,
     upload_parallelism: int | None = None,
+    stability_period_seconds: int | None = None,
+    max_stability_wait_seconds: int | None = None,
 ) -> InstrumentConfig:
     watch_dir = tmp_path / "data"
     watch_dir.mkdir()
@@ -66,6 +68,10 @@ def _make_instrument(
     }
     if upload_parallelism is not None:
         kwargs["upload_parallelism"] = upload_parallelism
+    if stability_period_seconds is not None:
+        kwargs["stability_period_seconds"] = stability_period_seconds
+    if max_stability_wait_seconds is not None:
+        kwargs["max_stability_wait_seconds"] = max_stability_wait_seconds
     return InstrumentConfig(**kwargs)  # type: ignore[arg-type]
 
 
@@ -100,3 +106,47 @@ def test_negative_rejected(self, tmp_path: Path) -> None:
     def test_above_cap_rejected(self, tmp_path: Path) -> None:
         with pytest.raises(ValidationError, match="less than or equal to 32"):
             _make_instrument(tmp_path, upload_parallelism=64)
+
+
+class TestMaxStabilityWaitSeconds:
+    """Validation rules for ``InstrumentConfig.max_stability_wait_seconds``.
+
+    Defaulted to ``MAX_STABILITY_WAIT_SECONDS`` so existing config YAMLs
+    keep today's 5-minute abandon behaviour. Must be at least the
+    stability period so a file can become stable before it is abandoned.
+    """
+
+    def test_defaults_to_constant(self, tmp_path: Path) -> None:
+        cfg = _make_instrument(tmp_path)
+        assert cfg.max_stability_wait_seconds == MAX_STABILITY_WAIT_SECONDS
+
+    def test_explicit_minimum_of_one(self, tmp_path: Path) -> None:
+        cfg = _make_instrument(
+            tmp_path,
+            stability_period_seconds=1,
+            max_stability_wait_seconds=1,
+        )
+        assert cfg.max_stability_wait_seconds == 1
+
+    def test_explicit_maximum_of_one_day(self, tmp_path: Path) -> None:
+        cfg = _make_instrument(tmp_path, max_stability_wait_seconds=86400)
+        assert cfg.max_stability_wait_seconds == 86400
+
+    def test_zero_rejected(self, tmp_path: Path) -> None:
+        with pytest.raises(ValidationError, match="greater than or equal to 1"):
+            _make_instrument(tmp_path, max_stability_wait_seconds=0)
+
+    def test_above_cap_rejected(self, tmp_path: Path) -> None:
+        with pytest.raises(ValidationError, match="less than or equal to 86400"):
+            _make_instrument(tmp_path, max_stability_wait_seconds=86401)
+
+    def test_shorter_than_stability_period_rejected(self, tmp_path: Path) -> None:
+        with pytest.raises(
+            ValidationError,
+            match="max_stability_wait_seconds .* must be >= stability_period_seconds",
+        ):
+            _make_instrument(
+                tmp_path,
+                stability_period_seconds=60,
+                max_stability_wait_seconds=30,
+            )

web/lib/docs.ts

@@ -7,6 +7,7 @@ const DOCS_ORIGIN =
 // The docs site is served under `/docs` on the product's domain (Vercel
 // Microfrontends), so every docs link hangs off `${DOCS_ORIGIN}/docs`.
 export const DOCS_URL = `${DOCS_ORIGIN}/docs`;
-export const QUICKSTART_DOCS_URL = `${DOCS_URL}/quickstart`;
-export const ADD_INSTRUMENT_DOCS_URL = `${DOCS_URL}/adding-an-instrument`;
-export const MANAGING_TOKENS_DOCS_URL = `${DOCS_URL}/managing-tokens`;
+// Slugs must match data-hub-docs `content/docs/*.mdx` (and meta.json).
+export const QUICKSTART_DOCS_URL = `${DOCS_URL}/overview`;
+export const ADD_INSTRUMENT_DOCS_URL = `${DOCS_URL}/set-up-an-instrument`;
+export const MANAGING_TOKENS_DOCS_URL = `${DOCS_URL}/manage-tokens`;