File viewer: Open instantly by backgrounding the FS watcher subscribe

- Move the blocking FSEvents subscribe off `open_session`'s critical path onto the watcher-manager thread (`spawn_watcher_manager`). Open drops from ~118ms idle (unbounded and fat-tailed under load: 195-730ms under a synthetic FS-event flood) to ~0.3ms regardless of system load. This is a prod win too: every viewer open previously paid this latency under the 2s `viewer_open` timeout, so a busy machine or slow/network path could time out the open.
- Close the open→subscribe missed-append window with `catch_up_after_subscribe`: right after subscribing, re-stat the file and, if on-disk size exceeds what the live backend covers, drive the same `Grew` path a real event would. Comparing against live backend coverage is correct regardless of upgrade timing (mid-upgrade queues into `pending_grew`; post-upgrade tail-extends or emits).
- Fix the `search_poll_no_active_search` 8s-nextest-cap flake: the test did ~zero CPU work; 100% of its time was that one `fseventsd`-bound subscribe, whose tail crossed 8s under the saturated check suite.
- Tests that inject synthetic watcher events now call `wait_for_watcher_subscribed()`, since the watcher isn't live the instant open returns.

Author: vdavid

apps/desktop/src-tauri/src/file_viewer/CLAUDE.md

@@ -98,7 +98,7 @@ Classification per debounce window:
 - `Replaced` when the inode changed (rename + atomic replace, log rotation).
 - `MetadataOnly` when nothing observable changed.
 
-Per-session, a manager thread (`spawn_watcher_manager_thread`) consumes events on the subscription channel:
+Per-session, a manager thread (`spawn_watcher_manager`) does the FSEvents subscribe itself (see the gotcha below), then consumes events on the subscription channel:
 
 - Always emits `viewer:file-changed:<sid>` with `{ kind: "grew", newSize }` or `{ kind: "rotated" }`.
 - `Grew` with `upgrading` or `rebuilding` in flight: queues `pending_grew = Some(new_size.max(prev))` (drain-and-swap
@@ -118,12 +118,20 @@ Per-session, a manager thread (`spawn_watcher_manager_thread`) consumes events o
 
 ## Gotchas (tail mode)
 
-**Watcher subscribe happens AFTER `SESSIONS.insert` but BEFORE the upgrade spawn.** `notify-debouncer-full::new_debouncer`
-plus `debouncer.watch` need the session already in `SESSIONS` so the manager thread can look it up. They run before the
-upgrade thread spawn so the watcher captures any append that lands during the upgrade window; otherwise an append
-arriving between SESSIONS.insert and the watcher's first event would be observed by no one (the upgrade has already
-stored its LineIndex covering only the pre-append EOF, and no later FS event ever fires for that one append). Pinned by
-`tail_mode_on_extends_backend_when_watcher_reports_grew` and `test_append_during_upgrade_not_dropped`.
+**The FSEvents subscribe runs on the manager thread, NOT inline in `open_session`.** `notify-debouncer-full::new_debouncer`
++ `debouncer.watch` is a blocking, `fseventsd`-bound call: ~100 ms idle on macOS and seconds under load (measured: a
+0.3 s test became >8 s under the full check suite, tripping the nextest cap; a synthetic FS-event flood pushed the
+subscribe alone from 118 ms to 730 ms). Doing it inline made every viewer open pay that latency and risked the 2 s
+`viewer_open` timeout on a busy system. So `open_session` spawns `spawn_watcher_manager`, which subscribes on its own
+thread and only then runs the poll loop. Open is now sub-millisecond regardless of system load. **Don't move the
+subscribe back inline.** Because the subscribe no longer precedes the upgrade thread, an append could land in the
+open→subscribe window and fire no event (the watcher's size baseline is the on-disk EOF at subscribe time). That window
+is closed by `catch_up_after_subscribe`: right after subscribing, it re-stats the file and, if the on-disk size exceeds
+what the active backend currently covers, drives the same `Grew` path a real event would — correct whether the
+ByteSeek→LineIndex upgrade has stored yet (mid-upgrade it queues into `pending_grew`; post-upgrade it tail-extends or
+emits). Tests that inject synthetic watcher events must call `wait_for_watcher_subscribed()` after `open_session` first,
+since the watcher isn't live the instant open returns. Pinned by `tail_mode_on_extends_backend_when_watcher_reports_grew`
+and `test_append_during_upgrade_not_dropped`.
 
 **Tail-extend race against an encoding rebuild.** `apply_tail_extend` snapshots the active backend `Arc`, drops the
 SESSIONS lock, runs `extend_to_boxed` (multi-second on a multi-MB append), then re-acquires the lock. If an encoding

apps/desktop/src-tauri/src/file_viewer/session.rs

@@ -4,7 +4,7 @@
 //! API for the frontend. Sessions are cached by ID and cleaned up on close.
 
 use std::collections::HashMap;
-use std::path::PathBuf;
+use std::path::{Path, PathBuf};
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, LazyLock, Mutex, RwLock};
 use std::thread;
@@ -24,7 +24,7 @@ use super::full_load::FullLoadBackend;
 use super::line_index::LineIndexBackend;
 use super::range_read::{RangeEnd, read_range as do_read_range};
 use super::search_matcher::{Matcher, SearchMode};
-use super::watcher::{VIEWER_WATCHER_MANAGER, ViewerSubscription, WatcherEvent};
+use super::watcher::{VIEWER_WATCHER_MANAGER, WatcherEvent};
 use super::{
     BackendCapabilities, FULL_LOAD_THRESHOLD, FileViewerBackend, LineChunk, MAX_SEARCH_MATCHES, SearchMatch,
     SeekTarget, ViewerError,
@@ -361,26 +361,17 @@ pub fn open_session(path: &str) -> Result<ViewerOpenResult, ViewerError> {
     let session_path = session.path.clone();
     SESSIONS.lock_ignore_poison().insert(session_id.clone(), session);
 
-    // Subscribe to filesystem events BEFORE spawning the upgrade thread.
-    // If the upgrade thread is spawned first and finishes between the spawn
-    // and the watcher subscribe, any append on disk during that window is
-    // observed by no one: the upgrade thread already stored its LineIndex
-    // (covering only the pre-append EOF), and the watcher hasn't subscribed
-    // yet so no event ever fires. Subscribing first means the watcher is
-    // live throughout the upgrade window; any append queues into
-    // `pending_grew` and the upgrade's drain-and-swap critical section
-    // picks it up.
-    //
-    // The watcher manager thread itself locks SESSIONS to look up the
-    // session on each event, so it gracefully tolerates the brief window
-    // between this insert and the manager thread's first `recv`.
+    // Attach the filesystem watcher off the critical path. The FSEvents
+    // subscribe is a blocking, `fseventsd`-bound call (~100ms idle, seconds
+    // under load), so doing it inline would make every viewer open pay that
+    // latency and risk the 2s `viewer_open` timeout on a busy system. The
+    // manager thread does the subscribe itself, then a catch-up re-stat closes
+    // the open→subscribe window for any append that landed before the watcher
+    // went live. See `spawn_watcher_manager`.
     //
     // Tests can opt out via `CMDR_VIEWER_DISABLE_WATCHER=1`.
     if std::env::var("CMDR_VIEWER_DISABLE_WATCHER").is_err() {
-        match VIEWER_WATCHER_MANAGER.subscribe(&session_path) {
-            Ok(sub) => spawn_watcher_manager_thread(session_id.clone(), sub, watcher_stop_for_thread),
-            Err(e) => debug!("viewer watcher subscribe failed for {}: {}", session_path.display(), e),
-        }
+        spawn_watcher_manager(session_id.clone(), session_path, watcher_stop_for_thread);
     }
 
     // If we're using ByteSeek, start background upgrade to LineIndex with timeout
@@ -822,7 +813,7 @@ pub fn write_range_to_file(
     read_id: u64,
     anchor: RangeEnd,
     focus: RangeEnd,
-    dest_path: &std::path::Path,
+    dest_path: &Path,
 ) -> Result<(), ViewerError> {
     let text = read_range(session_id, read_id, anchor, focus)?;
 
@@ -1189,13 +1180,38 @@ pub fn reload(session_id: &str) -> Result<(), ViewerError> {
     Ok(())
 }
 
-/// Manager thread spawned once per session. Owns the `ViewerSubscription`
-/// (kept off `ViewerSession` because the channel receiver isn't `Sync`).
-/// Drops the subscription when `stop` flips (set by `close_session`) or when
-/// the upstream channel disconnects; the subscription's `Drop` then
-/// unregisters the path from the shared singleton.
-fn spawn_watcher_manager_thread(session_id: String, sub: ViewerSubscription, stop: Arc<AtomicBool>) {
+/// Manager thread spawned once per session. Does the (blocking,
+/// `fseventsd`-bound) FSEvents subscribe itself — off `open_session`'s critical
+/// path — then owns the resulting `ViewerSubscription` (kept off `ViewerSession`
+/// because the channel receiver isn't `Sync`) and runs the event loop. Drops the
+/// subscription when `stop` flips (set by `close_session`) or when the upstream
+/// channel disconnects; the subscription's `Drop` then unregisters the path from
+/// the shared singleton.
+fn spawn_watcher_manager(session_id: String, path: PathBuf, stop: Arc<AtomicBool>) {
     thread::spawn(move || {
+        // `close_session` may have run before this thread got scheduled; skip
+        // the blocking subscribe entirely in that case (nothing registered yet,
+        // so nothing to unregister).
+        if stop.load(Ordering::Relaxed) {
+            return;
+        }
+
+        let sub = match VIEWER_WATCHER_MANAGER.subscribe(&path) {
+            Ok(sub) => sub,
+            Err(e) => {
+                debug!("viewer watcher subscribe failed for {}: {}", path.display(), e);
+                return;
+            }
+        };
+
+        // `stop` may have flipped while we were subscribing. Returning here
+        // drops `sub`, which unregisters the path.
+        if stop.load(Ordering::Relaxed) {
+            return;
+        }
+
+        catch_up_after_subscribe(&session_id, &path);
+
         // Poll with a timeout so we periodically check `stop` even if the
         // file's idle: this is the only way the manager exits when
         // `close_session` runs without the file changing first.
@@ -1214,6 +1230,33 @@ fn spawn_watcher_manager_thread(session_id: String, sub: ViewerSubscription, sto
     });
 }
 
+/// Closes the open→subscribe missed-append window. Because the subscribe runs
+/// in the background and blocks for an unbounded time, an append can land
+/// between open and the watcher going live — and the watcher's size baseline is
+/// the on-disk EOF *at subscribe time*, so that append would fire no event and
+/// stay invisible. We compare the current on-disk size against what the active
+/// backend covers and, if the file grew, drive the same path a real `Grew`
+/// event would. Comparing against live backend coverage (rather than a captured
+/// open-time EOF) makes this correct regardless of whether the ByteSeek →
+/// LineIndex upgrade has already stored: mid-upgrade it queues into
+/// `pending_grew` (drained by the swap), post-upgrade it tail-extends or emits.
+fn catch_up_after_subscribe(session_id: &str, path: &Path) {
+    let Ok(meta) = std::fs::metadata(path) else {
+        return;
+    };
+    let on_disk = meta.len();
+    let covered = {
+        let sessions = SESSIONS.lock_ignore_poison();
+        let Some(session) = sessions.get(session_id) else {
+            return; // session closed while we were subscribing
+        };
+        session.backend.load_full().total_bytes()
+    };
+    if on_disk > covered {
+        handle_watcher_event(session_id, WatcherEvent::Grew(on_disk));
+    }
+}
+
 fn emit_file_changed(session_id: &str, kind: &'static str, new_size: Option<u64>) {
     let Some(handle) = app_handle() else {
         return;

apps/desktop/src-tauri/src/file_viewer/session_test.rs

@@ -35,6 +35,23 @@ fn write_test_file(dir: &Path, name: &str, content: &str) -> PathBuf {
     file
 }
 
+/// Wait until a freshly-opened session's watcher subscribe has landed, so
+/// `test_only_emit` reliably reaches a subscriber. The subscribe runs on a
+/// background thread off `open_session`'s critical path (see
+/// `spawn_watcher_manager`), so tests that inject synthetic watcher events must
+/// sync on it instead of assuming the watcher is live the moment open returns.
+/// Each test runs in its own nextest process, so `watch_count() > 0` reflects
+/// only the session(s) opened in this test.
+fn wait_for_watcher_subscribed() {
+    for _ in 0..200 {
+        if super::watcher::VIEWER_WATCHER_MANAGER.watch_count() > 0 {
+            return;
+        }
+        thread::sleep(Duration::from_millis(10));
+    }
+    panic!("watcher subscribe did not land within 2s");
+}
+
 #[test]
 fn open_small_file_uses_full_load() {
     let dir = create_test_dir("small");
@@ -1251,6 +1268,9 @@ fn tail_mode_on_extends_backend_when_watcher_reports_grew() {
     let result = session::open_session(path.to_str().unwrap()).unwrap();
     let sid = result.session_id.clone();
     let original_bytes = result.total_bytes;
+    // Sync on the background subscribe before mutating, so the catch-up re-stat
+    // stays a no-op and the explicit emit below is the driver under test.
+    wait_for_watcher_subscribed();
 
     // Wait for the background ByteSeek → LineIndex upgrade to finish so
     // we're testing the post-upgrade fast path (extend an existing
@@ -1496,6 +1516,9 @@ fn test_session_emits_file_changed_on_append() {
     let file = write_test_file(&dir, "emit.log", &initial);
     let result = session::open_session(file.to_str().unwrap()).unwrap();
     let sid = result.session_id.clone();
+    // Sync on the background subscribe before mutating the file, so the
+    // catch-up re-stat stays a no-op and the explicit emit below is the driver.
+    wait_for_watcher_subscribed();
     session::set_tail_mode(&sid, true).unwrap();
 
     let mut content = initial.clone();
@@ -1536,6 +1559,7 @@ fn test_session_tail_mode_off_does_not_extend_index() {
     let file = write_test_file(&dir, "tailoff.log", initial);
     let result = session::open_session(file.to_str().unwrap()).unwrap();
     let sid = result.session_id.clone();
+    wait_for_watcher_subscribed();
     assert!(!session::test_only_tail_mode(&sid));
 
     let original_lines = session::get_session_status(&sid).unwrap().total_lines;
@@ -1580,6 +1604,7 @@ fn test_session_rotation_reopens_backend() {
     let result = session::open_session(file.to_str().unwrap()).unwrap();
     let sid = result.session_id.clone();
     let original_bytes = result.total_bytes;
+    wait_for_watcher_subscribed();
 
     // Replace the file with a longer one.
     let new_path = dir.join("rot.log.new");
@@ -1619,6 +1644,7 @@ fn test_session_close_stops_watcher() {
     let result = session::open_session(file.to_str().unwrap()).unwrap();
     let sid = result.session_id.clone();
 
+    wait_for_watcher_subscribed();
     let canonical = fs::canonicalize(&file).unwrap();
     // The subscription is alive: a test-only emit reaches it.
     let sent_before = super::watcher::test_only_emit(&canonical, super::watcher::WatcherEvent::MetadataOnly);