Watcher: Emit a startup heartbeat before the initial directory scan

watcher/pyproject.toml

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

@@ -70,8 +70,34 @@ def __init__(
         self._first_failure_at: float | None = None
 
     def start(self) -> None:
+        """Send an immediate heartbeat, then run periodic heartbeats in a thread.
+
+        The synchronous startup beat is critical because ``start()`` is
+        always called immediately before ``FileMonitor.start()`` runs the
+        initial directory scan, which can block the main thread for
+        several minutes on large or networked watch volumes. Without the
+        immediate beat, the dashboard would see nothing for the full
+        scan window plus up to one ``interval``, and the watcher would
+        be indistinguishable from a hung process — especially painful on
+        Windows where the service runs headless and the only liveness
+        signal an operator has is what shows up in Data Hub.
+
+        We also flush the event reporter on this initial tick so the
+        ``WATCHER_STARTED`` event the runtime queued moments ago becomes
+        visible right away rather than piggy-backing on the first
+        loop-driven flush ~``interval`` seconds later.
+        """
         self._start_time = time.monotonic()
         self._stop_event.clear()
+        # Synchronous startup beat + flush so the dashboard sees the
+        # watcher come up before the (potentially long-running) initial
+        # scan begins. ``_send_heartbeat`` and ``flush`` both swallow
+        # network errors internally, so a startup with the API
+        # unreachable still proceeds — the next loop tick will retry
+        # and emit the standard ``heartbeat_recovered`` event when the
+        # link comes back.
+        self._send_heartbeat(status="watching")
+        self._event_reporter.flush()
         self._thread = threading.Thread(target=self._run, daemon=True, name="heartbeat")
         self._thread.start()
 

watcher/tests/test_heartbeat.py

@@ -1,14 +1,25 @@
 """Unit tests for ``HeartbeatLoop``.
 
-The recovery path is the only thing exercised here: heartbeat
-*failures* are by definition unreportable in real time (the network
-they depend on is down), so the watcher accumulates a counter and
-emits a ``kind=heartbeat_recovered`` event on the first successful
-heartbeat after one or more failures. That recovery event is the
-only signal an operator gets that the watcher went dark.
+Two behaviours are exercised here:
+
+1. The recovery path: heartbeat *failures* are by definition
+   unreportable in real time (the network they depend on is down), so
+   the watcher accumulates a counter and emits a
+   ``kind=heartbeat_recovered`` event on the first successful
+   heartbeat after one or more failures. That recovery event is the
+   only signal an operator gets that the watcher went dark.
+
+2. The startup path: ``start()`` must send a synchronous heartbeat
+   *and* flush queued events before spawning the loop thread, so the
+   dashboard sees the watcher come up immediately rather than after a
+   full ``interval`` of silence — and, in particular, before the
+   ``FileMonitor.start()`` call that runs the (potentially multi-minute)
+   initial directory scan on the same thread.
 """
 
 from __future__ import annotations
+import threading
+import time
 from unittest.mock import MagicMock
 
 from data_hub_watcher.api_client import ApiError
@@ -73,3 +84,136 @@ def test_failures_alone_do_not_emit_event(self) -> None:
             loop._send_heartbeat()
 
         reporter.report_error.assert_not_called()
+
+
+class TestHeartbeatStartupBeat:
+    """``start()`` must send an immediate heartbeat + flush before looping.
+
+    The runtime calls ``HeartbeatLoop.start()`` *just before*
+    ``FileMonitor.start()``, which runs the synchronous initial
+    directory scan. On large or networked watch volumes that scan can
+    block the main thread for several minutes. Without a synchronous
+    startup beat the dashboard sees no liveness signal — and the
+    ``WATCHER_STARTED`` event the runtime queued moments earlier
+    doesn't get flushed — for that whole window, which is
+    indistinguishable from a hung process.
+    """
+
+    def _stop(self, loop: HeartbeatLoop) -> None:
+        """Tear down the daemon thread spawned by ``start()``.
+
+        ``HeartbeatLoop.stop()`` would also send a final ``status=stopped``
+        heartbeat, which would muddy the assertions in these tests. We
+        only need the loop thread to exit so the test process doesn't
+        leak threads, so we set the stop event and join directly.
+        """
+        loop._stop_event.set()
+        if loop._thread is not None:
+            loop._thread.join(timeout=5)
+
+    def test_start_sends_immediate_watching_heartbeat(self) -> None:
+        client = MagicMock()
+        reporter = MagicMock(spec=EventReporter)
+        loop = _make_loop(client, reporter)
+
+        loop.start()
+        try:
+            assert client.send_heartbeat.call_count >= 1
+            first_call = client.send_heartbeat.call_args_list[0]
+            assert first_call.args[0] == "w-test"
+            payload = first_call.args[1]
+            assert payload["status"] == "watching"
+            assert payload["instrument_id"] == "inst-1"
+            assert payload["watch_directory"] == "/tmp/watch"
+        finally:
+            self._stop(loop)
+
+    def test_start_flushes_event_reporter_before_loop_thread(self) -> None:
+        """The flush must run before the loop thread is spawned, so
+        ``WATCHER_STARTED`` (queued by the runtime moments earlier)
+        reaches the dashboard before any long-running initial scan
+        starts on the main thread.
+        """
+        client = MagicMock()
+        reporter = MagicMock(spec=EventReporter)
+        # Capture the live thread state at the moment ``flush`` runs so
+        # we can prove the call originates from the main thread, not
+        # from the loop thread that ``start()`` spawns afterwards.
+        flush_thread_names: list[str] = []
+        flush_called_before_thread_alive: list[bool] = []
+
+        def _record_flush() -> None:
+            flush_thread_names.append(threading.current_thread().name)
+            flush_called_before_thread_alive.append(
+                loop._thread is None or not loop._thread.is_alive()
+            )
+
+        reporter.flush.side_effect = _record_flush
+        loop = _make_loop(client, reporter)
+
+        loop.start()
+        try:
+            assert reporter.flush.call_count >= 1
+            assert flush_thread_names[0] == threading.current_thread().name
+            assert flush_called_before_thread_alive[0] is True
+        finally:
+            self._stop(loop)
+
+    def test_start_proceeds_when_initial_heartbeat_fails(self) -> None:
+        """A startup-time API outage must not prevent the loop from running.
+
+        ``_send_heartbeat`` already swallows ``ApiError``/``Exception``
+        and bumps the consecutive-failure counter; ``flush`` likewise
+        retries internally and re-queues on failure. ``start()`` must
+        therefore still spawn the loop thread so subsequent ticks can
+        recover and emit the standard ``heartbeat_recovered`` event.
+        """
+        client = MagicMock()
+        client.send_heartbeat.side_effect = ApiError("offline", status_code=0)
+        reporter = MagicMock(spec=EventReporter)
+        loop = _make_loop(client, reporter)
+
+        loop.start()
+        try:
+            assert loop._thread is not None
+            assert loop._thread.is_alive()
+            # The failed startup beat must have been recorded so the
+            # next successful tick emits ``heartbeat_recovered``.
+            assert loop._consecutive_heartbeat_failures >= 1
+        finally:
+            self._stop(loop)
+
+    def test_startup_beat_fires_before_loop_would_have(self) -> None:
+        """End-to-end timing guard: the first heartbeat must land
+        well before the configured ``interval``. With the old
+        loop-only behaviour the first beat was deferred by the full
+        interval, so this test would fail (or hang) without the
+        synchronous startup beat. We use a generous 1 s interval and
+        a 100 ms ceiling so the assertion is robust against CI
+        scheduling jitter.
+        """
+        client = MagicMock()
+        reporter = MagicMock(spec=EventReporter)
+        loop = HeartbeatLoop(
+            client=client,
+            watcher_id="w-test",
+            # 1 s is plenty long enough that a loop-driven first beat
+            # would not have fired by the time we measure.
+            interval_seconds=1,
+            event_reporter=reporter,
+            instrument_id="inst-1",
+            watch_directory="/tmp/watch",
+            upload_mode="auto",
+        )
+
+        before = time.monotonic()
+        loop.start()
+        try:
+            elapsed = time.monotonic() - before
+            assert client.send_heartbeat.call_count >= 1
+            assert elapsed < 0.1, (
+                f"Startup heartbeat took {elapsed:.3f}s — must fire "
+                "synchronously, not via the loop's interval wait."
+            )
+        finally:
+            self._stop(loop)