Watcher: Move manual-mode uploads to a dedicated worker thread

developer-docs/reference/watcher.md

@@ -67,8 +67,9 @@ 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.
 - **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 polls 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.
-- **Heartbeat loop** sends periodic heartbeats (every 60 seconds) to the API. The payload includes the watcher version, instrument ID, watch directory, upload mode, per-interval activity counters, and process uptime; a final `status="stopped"` heartbeat is sent on graceful shutdown. In manual mode, the tick also polls the upload queue.
+- **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.
+- **Heartbeat loop** sends periodic heartbeats (every 60 seconds) to the API. The payload includes the watcher version, instrument ID, watch directory, upload mode, per-interval activity counters, and process uptime; a final `status="stopped"` heartbeat is sent on graceful shutdown.
 - **Event reporter** batches and flushes lifecycle events (started, stopped, file uploaded, errors) to the API. See [Observability](#observability) for the full taxonomy.
 - **Auto-updater** runs from the same heartbeat tick on every platform — not only Windows services. It polls `GET /watchers/:id/update-check` roughly hourly and applies new releases when the watcher has been idle long enough not to clobber an in-flight run. The full activity-window guard, mandatory-update behavior, and rollback flow are documented in [Upgrading the watcher](../guides/upgrading-the-watcher.md); auto-update is hard-disabled in the `preview` environment.
 
@@ -207,12 +208,12 @@ Upgrading an existing watcher is unaffected: the environment's database already
 ### Upload modes
 
 - **`auto`**: Files are uploaded to S3 immediately after run detection.
-- **`manual`**: Runs are reported to the API without uploading. The server decides which files to upload via a queue, polled on each heartbeat tick. Useful when uploads need human approval.
+- **`manual`**: Runs are reported to the API without uploading. The server decides which files to upload via a queue, polled by the upload worker thread every 60 seconds. Useful when uploads need human approval.
 
 Queued files are resolved against the current `watch_directory` (each queue entry carries a `relative_path` anchored to the root that was active when the file was detected). Two safeguards keep a stale queue entry from erroring forever (ENG-1397):
 
 - **On `watch_directory` change**: the server reverts every pending upload request for that instrument back to `detected` (clearing `upload_requested_at`) as soon as the new config is pushed, so the queue drains immediately. The reverted files remain re-requestable detections; an operator can queue them again from their new location.
-- **Per-file 3-try cap (`MAX_QUEUE_FILE_ATTEMPTS`)**: a queued file that keeps failing — missing on disk or failing to upload — is retried on at most three heartbeat polls. After that the watcher cancels the request server-side (revert to `detected`) so the file leaves the queue instead of re-erroring each tick. The attempt count resets on watcher restart, so a transient outage longer than three ticks is recovered on the next start.
+- **Per-file 3-try cap (`MAX_QUEUE_FILE_ATTEMPTS`)**: a queued file that keeps failing — missing on disk or failing to upload — is retried on at most three upload-queue polls. After that the watcher cancels the request server-side (revert to `detected`) so the file leaves the queue instead of re-erroring each poll. The attempt count resets on watcher restart, so a transient outage longer than three polls is recovered on the next start.
 
 ## Local state
 

watcher/pyproject.toml

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

@@ -71,6 +71,12 @@ def _resolve_watcher_log_dir() -> Path:
 SUPPORTED_ENVIRONMENTS: tuple[str, ...] = ("staging", "production", "preview")
 
 HEARTBEAT_INTERVAL_SECONDS = 60
+# Kept separate from ``HEARTBEAT_INTERVAL_SECONDS`` so the poll cadence can
+# diverge now that uploads no longer ride the heartbeat tick.
+UPLOAD_POLL_INTERVAL_SECONDS = 60
+# Bounded so a service stop doesn't hang on a large in-flight PUT; past this,
+# shutdown stops waiting for the worker and leaves the state DB open.
+UPLOAD_WORKER_STOP_TIMEOUT_SECONDS = 30
 DEFAULT_STABILITY_PERIOD_SECONDS = 5
 
 # Built-in presets for the ``init`` / ``config edit`` wizard.

watcher/src/data_hub_watcher/runtime.py

@@ -22,6 +22,7 @@
     DEFAULT_CONFIG_DIR,
     HEARTBEAT_INTERVAL_SECONDS,
     PRUNE_DAYS,
+    UPLOAD_WORKER_STOP_TIMEOUT_SECONDS,
 )
 from data_hub_watcher.events import EventReporter, EventType, WatcherEvent
 from data_hub_watcher.heartbeat import HeartbeatLoop, WatcherCounters
@@ -35,7 +36,7 @@
     clear_upgrade_result,
     read_upgrade_result,
 )
-from data_hub_watcher.uploader import Uploader
+from data_hub_watcher.uploader import Uploader, UploadQueueWorker
 
 logger = logging.getLogger(__name__)
 
@@ -74,6 +75,10 @@ class WatcherRuntime:
     # in their stop wait so a shutdown can be triggered from any thread.
     shutdown_event: threading.Event = field(default_factory=threading.Event)
     upgrade_restart_event: threading.Event = field(default_factory=threading.Event)
+    # Manual mode only: the thread that polls the upload queue off the
+    # heartbeat. ``None`` in auto mode, where uploads run on the monitor's
+    # stability-checker thread via the run detector's callback.
+    upload_worker: UploadQueueWorker | None = None
 
 
 @dataclass(frozen=True)
@@ -251,6 +256,11 @@ def _request_upgrade_restart(target_version: str) -> None:
 
     is_auto = inst.upload_mode == "auto"
 
+    # Shared with the manual-mode ``UploadQueueWorker`` so a shutdown can
+    # interrupt an in-flight upload's backoff and abort between queued files;
+    # unused (but harmless) in auto mode.
+    upload_stop_event = threading.Event()
+
     uploader = Uploader(
         client=client,
         state_db=state_db,
@@ -262,8 +272,13 @@ def _request_upgrade_restart(target_version: str) -> None:
         # Per-instrument knob, defaulted on the model so older configs
         # transparently inherit the new parallel-upload behaviour.
         upload_parallelism=inst.upload_parallelism,
+        stop_event=upload_stop_event,
     )
 
+    # Manual mode polls the server queue on its own thread; auto mode uploads
+    # via the run detector's callback on the stability-checker thread instead.
+    upload_worker = None if is_auto else UploadQueueWorker(uploader, stop_event=upload_stop_event)
+
     detector = RunDetector(
         pattern=inst.run_detection.pattern,
         instrument_id=inst.id,
@@ -287,19 +302,10 @@ def _request_upgrade_restart(target_version: str) -> None:
         seed_baseline=seed_baseline,
     )
 
-    # The heartbeat's `on_tick` hook is now multi-purpose:
-    #   1. In manual mode, poll the server's upload queue (uploads
-    #      naturally inherit the heartbeat cadence).
-    #   2. Always: feed the in-process auto-updater so it can count
-    #      idle ticks and run a server update-check roughly hourly.
-    # Each side wraps its own try/except so a failure on one side
-    # never blocks the other.
+    # Feeds the in-process auto-updater on every tick. Manual-mode upload
+    # polling used to run here too but moved to ``UploadQueueWorker`` so a
+    # slow upload can't delay a heartbeat.
     def _on_tick() -> None:
-        if not is_auto:
-            try:
-                uploader.poll_upload_queue()
-            except Exception:
-                logger.exception("Upload queue poll failed")
         try:
             updater.on_tick()
         except Exception:
@@ -329,6 +335,7 @@ def _on_tick() -> None:
         config_dir=effective_config_dir,
         shutdown_event=shutdown_event,
         upgrade_restart_event=upgrade_restart_event,
+        upload_worker=upload_worker,
     )
 
 
@@ -480,6 +487,10 @@ def start_runtime(rt: WatcherRuntime, *, started_message: str) -> None:
     rt.detector.hydrate_from_state_db()
 
     rt.heartbeat.start()
+    # Start manual-mode upload polling before the (potentially long) initial
+    # scan so queued uploads keep draining while the scan walks the backlog.
+    if rt.upload_worker is not None:
+        rt.upload_worker.start()
     rt.monitor.start()
 
 
@@ -551,6 +562,17 @@ def sync_config_to_api(
 def stop_runtime(rt: WatcherRuntime, *, stopped_message: str) -> None:
     """Shut everything down in reverse order and flush pending events."""
     rt.monitor.stop()
+    # ``StateDB.close`` assumes writer threads have joined; a large upload
+    # outliving the join once wrote post-close. Stop the worker first, and if
+    # it won't stop in time, skip the close rather than race it (the OS reaps).
+    upload_worker_stopped = True
+    if rt.upload_worker is not None:
+        upload_worker_stopped = rt.upload_worker.stop(timeout=UPLOAD_WORKER_STOP_TIMEOUT_SECONDS)
+        if not upload_worker_stopped:
+            logger.warning(
+                "Upload worker still running at shutdown; leaving the state DB "
+                "open so the in-flight upload can finish without a closed-DB error"
+            )
     rt.reporter.queue_event(
         WatcherEvent(
             event_type=EventType.WATCHER_STOPPED,
@@ -559,4 +581,5 @@ def stop_runtime(rt: WatcherRuntime, *, stopped_message: str) -> None:
     )
     rt.heartbeat.stop()
     rt.reporter.flush()
-    rt.state_db.close()
+    if upload_worker_stopped:
+        rt.state_db.close()

watcher/src/data_hub_watcher/uploader.py

@@ -22,6 +22,7 @@
 from data_hub_watcher.api_client import ApiError, DataHubClient
 from data_hub_watcher.constants import (
     MAX_QUEUE_FILE_ATTEMPTS,
+    UPLOAD_POLL_INTERVAL_SECONDS,
     UPLOAD_RETRY_BASE_DELAY,
     UPLOAD_RETRY_MAX,
 )
@@ -105,6 +106,7 @@ def __init__(
         watcher_id: str,
         watch_directory: Path,
         upload_parallelism: int = 1,
+        stop_event: threading.Event | None = None,
     ) -> None:
         self._client = client
         self._state_db = state_db
@@ -116,18 +118,22 @@ def __init__(
         if upload_parallelism < 1:
             raise ValueError(f"upload_parallelism must be >= 1, got {upload_parallelism}")
         self._parallelism = upload_parallelism
+        # When set (by the owning ``UploadQueueWorker``), a shutdown can
+        # interrupt the retry backoff and abort between queued files; ``None``
+        # for the one-shot ``upload`` CLI path, which never races a teardown.
+        self._stop_event = stop_event
         # Track consecutive upload-queue poll failures so the watcher
         # surfaces a ``kind=upload_queue_poll_failed`` event on the
         # 1st failure and every 10th repeat. The unthrottled case
-        # would emit one event per heartbeat tick during an outage,
-        # crowding out other signals on the dashboard.
-        # Mutated only from the heartbeat thread (manual mode), so
+        # would emit one event per poll during an outage, crowding out
+        # other signals on the dashboard.
+        # Mutated only from the upload worker thread (manual mode), so
         # not under any explicit lock.
         self._consecutive_queue_poll_failures = 0
         # Per-file upload-queue attempt bookkeeping, keyed by server file id.
         # Bounds retries before giving up (see ``_process_queued_file``) and
         # doubles as the emit-once throttle for the missing-file error. Pruned
-        # each poll, so a re-requested id starts fresh. Heartbeat thread only.
+        # each poll, so a re-requested id starts fresh. Upload worker thread only.
         self._queue_attempts: dict[int, _QueueAttempt] = {}
         # Single ``requests.Session`` shared across every S3 PUT
         # (parallel or serial). Keeps TLS connections alive between
@@ -230,10 +236,15 @@ def upload_files(self, run_id: str, files: list[FileInfo]) -> int:
     # Manual-mode: poll the server queue
     # ------------------------------------------------------------------
 
+    def _stop_requested(self) -> bool:
+        return self._stop_event is not None and self._stop_event.is_set()
+
     def poll_upload_queue(self) -> None:
         """Fetch the upload queue and process each file.
 
-        Intended to be called on heartbeat ticks in manual mode.
+        Driven by the manual-mode ``UploadQueueWorker`` loop on its own
+        thread, decoupled from the heartbeat so a slow upload can't starve
+        the liveness signal.
         """
         try:
             queue = self._client.get_upload_queue(self._watcher_id)
@@ -278,13 +289,19 @@ def poll_upload_queue(self) -> None:
 
         logger.info("Upload queue has %d file(s)", len(queue.files))
         for qf in queue.files:
+            # Bail between files on shutdown so the worker's ``stop()`` can
+            # join promptly instead of draining the whole queue; the
+            # remaining files are picked up on the next start's poll.
+            if self._stop_requested():
+                logger.info("Stop requested; deferring %d queued file(s)", len(queue.files))
+                break
             self._process_queued_file(qf)
 
     def _process_queued_file(self, qf: UploadQueueFile) -> None:
-        """Attempt one queued file, bounding retries across heartbeat polls.
+        """Attempt one queued file, bounding retries across polls.
 
-        Manual-mode polling runs every heartbeat, so a file that can't be
-        uploaded -- missing on disk after a watch-directory change, or a
+        Manual-mode polling repeats on the worker's cadence, so a file that
+        can't be uploaded -- missing on disk after a watch-directory change, or a
         persistent upload error -- would otherwise re-error forever. We cap
         attempts at ``MAX_QUEUE_FILE_ATTEMPTS`` and then cancel the request
         server-side (revert to ``detected``) so it leaves the queue. The
@@ -509,13 +526,15 @@ def _upload_single(self, path: Path, run_id: str) -> bool:
             return True
 
         last_exc: Exception | None = None
+        put_ok = False
 
         # Exponential backoff: 1s, 2s, 4s. Retries protect against transient
         # network errors common on lab-PC networks.
         assert presigned.upload_url is not None
         for attempt in range(UPLOAD_RETRY_MAX):
             try:
                 self._put_to_presigned_url(presigned.upload_url, path, content_type)
+                put_ok = True
                 break
             except Exception as exc:
                 last_exc = exc
@@ -528,8 +547,21 @@ def _upload_single(self, path: Path, run_id: str) -> bool:
                     exc,
                     delay,
                 )
-                time.sleep(delay)
-        else:
+                # Wait on the stop event when present so a shutdown cuts the
+                # backoff short; falls back to a plain sleep for the one-shot
+                # ``upload`` path that has no worker/event.
+                if self._stop_event is not None:
+                    self._stop_event.wait(delay)
+                else:
+                    time.sleep(delay)
+            # Abandon the remaining retries on shutdown. Returning False (not a
+            # hard failure event) leaves the request pending so the next start
+            # re-uploads it, rather than recording a spurious upload error.
+            if self._stop_requested():
+                logger.info("Stop requested mid-upload; deferring %s", path.name)
+                return False
+
+        if not put_ok:
             logger.error("Upload failed after %d attempts: %s", UPLOAD_RETRY_MAX, path.name)
             self._reporter.queue_event(
                 WatcherEvent(
@@ -583,3 +615,59 @@ def _upload_single(self, path: Path, run_id: str) -> bool:
         )
         logger.info("Uploaded %s → s3://%s/%s", path.name, s3_bucket, s3_key)
         return True
+
+
+class UploadQueueWorker:
+    """Polls the manual-mode upload queue on a dedicated long-lived thread.
+
+    Uploads used to run on the heartbeat tick, so a slow or large transfer
+    delayed heartbeats and the dashboard flagged a busy watcher as offline.
+    Owning the poll loop here keeps the heartbeat free, and the shared
+    ``stop_event`` lets a shutdown interrupt an upload so ``stop()`` can join
+    before ``StateDB.close`` runs (which assumes writer threads have joined).
+    """
+
+    def __init__(
+        self,
+        uploader: Uploader,
+        *,
+        stop_event: threading.Event,
+        interval_seconds: int = UPLOAD_POLL_INTERVAL_SECONDS,
+    ) -> None:
+        self._uploader = uploader
+        self._stop_event = stop_event
+        self._interval = interval_seconds
+        self._thread: threading.Thread | None = None
+
+    def start(self) -> None:
+        self._stop_event.clear()
+        self._thread = threading.Thread(target=self._run, daemon=True, name="upload-worker")
+        self._thread.start()
+
+    def stop(self, timeout: float | None = None) -> bool:
+        """Signal the loop and join it. Returns whether the thread exited.
+
+        A ``False`` return means an upload is still in flight past *timeout*
+        (a single S3 PUT can run up to its request timeout); the caller uses
+        this to avoid closing the state DB out from under the live upload.
+        """
+        self._stop_event.set()
+        if self._thread is None:
+            return True
+        self._thread.join(timeout=timeout)
+        return not self._thread.is_alive()
+
+    def _run(self) -> None:
+        # Wait first (parity with the previous heartbeat-driven cadence),
+        # then poll each interval until stop.
+        while not self._stop_event.wait(timeout=self._interval):
+            self._poll_once()
+
+    def _poll_once(self) -> None:
+        # ``poll_upload_queue`` already handles and reports poll failures; this
+        # guard only keeps an unexpected error from killing the thread and
+        # silently ending all future polls.
+        try:
+            self._uploader.poll_upload_queue()
+        except Exception:
+            logger.exception("Upload queue poll failed")