Try and make watch client connection more resilient.

Author: nojaf

rewatch/EXPERIMENT_DAEMON.md

@@ -115,6 +115,7 @@ Eliminate lockfile limitation via gRPC daemon that owns build state. Commands (`
 | Client isolation       | client_id on all events                | Concurrent builds don't interfere, each sees only its own      |
 | Per-build CLI flags    | Passed to build, not stored            | Flags like `--warn-error` apply per-build without persisting   |
 | Scoping via path       | `get_scope_package_from_working_dir()` | Working dir determines scope, passed as parameter to commands  |
+| Connection resilience  | HTTP/2 keepalive + heartbeat + retry   | Daemon detects dead clients, watch client auto-reconnects      |
 
 ## How It Works
 
@@ -190,6 +191,16 @@ The watch client (`client/watch.rs`) uses a **daemon-driven** watching strategy:
 
 **Signal handling:** Watch client uses `tokio::signal::unix` to handle SIGTERM/SIGINT. On signal, sends `DisconnectRequest` to daemon before exiting cleanly.
 
+**Reconnection:** On gRPC stream errors, the watch client automatically reconnects:
+
+- `run()` wraps each session in a retry loop with exponential backoff (1s → 2s → 4s → ... → 30s cap)
+- `run_watch_session()` returns `SessionExit::Clean` (signal, no retry) or `SessionExit::StreamError` (retry)
+- On reconnect, `connect_or_start()` handles starting a new daemon if the old one died
+- File watchers are recreated fresh via `WatchPaths` event each session — no state carryover needed
+- Logs the last event received before the error for diagnostics
+
+**Heartbeat:** Daemon sends `Heartbeat` event every 30s on the watch stream. This ensures broken connections are detected even during idle periods (no build activity). The heartbeat write failing on the server side triggers tonic to drop the stream, firing `scopeguard` cleanup.
+
 **CompileType enum:** `Incremental`, `SourceCreated`, `SourceDeleted`, `SourceRenamed`, `ConfigChange`
 
 ### Daemon Lifecycle
@@ -199,6 +210,7 @@ The watch client (`client/watch.rs`) uses a **daemon-driven** watching strategy:
   - SIGTERM/SIGINT → graceful shutdown
   - SIGHUP → ignored (daemon survives terminal close)
   - All clients disconnected → shutdown
+- HTTP/2 keepalive: `http2_keepalive_interval(10s)` + `http2_keepalive_timeout(5s)` on tonic server — detects dead clients within ~15s even if no application traffic
 - Writes PID to `<root>/lib/bs/rescript.daemon.pid`
 - Writes socket path to `<root>/lib/bs/rescript.sock.path`
 - Cleans up socket, PID file, and socket path file on exit
@@ -312,6 +324,7 @@ Update after each session. Reflect current state, not history.
 - **std::sync::Mutex in async**: Works because locking only in `spawn_blocking`, but footgun
 - **No protocol versioning**: Client/daemon must be same version
 - **Unix-only**: No Windows support (needs named pipes or TCP)
+- **Reconnection is watch-only**: Build/clean/format clients don't retry on stream errors (they're short-lived, so restart is acceptable)
 
 ## Testing
 

rewatch/proto/rescript.proto

@@ -91,6 +91,9 @@ message DaemonEvent {
     FormatProgress format_progress = 21;
     FormatCheckFailed format_check_failed = 22;
     FormattedStdin formatted_stdin = 23;
+
+    // Keepalive
+    Heartbeat heartbeat = 34;
   }
 }
 
@@ -364,6 +367,16 @@ message FormattedStdin {
   string content = 2;  // Formatted content from stdin
 }
 
+// =============================================================================
+// Keepalive
+// =============================================================================
+
+// Periodic heartbeat sent on long-lived streams (e.g., watch) so broken
+// connections are detected even when no build activity is happening.
+message Heartbeat {
+  uint64 client_id = 1;
+}
+
 // =============================================================================
 // Request/Response Messages
 // =============================================================================

rewatch/src/client/debug.rs

@@ -419,6 +419,8 @@ impl App {
                     format!("{} ({})", e.command, output_info),
                 )
             }
+            // Heartbeats are keepalive signals — don't show in debug TUI
+            DaemonEventVariant::Heartbeat(_) => return,
         };
 
         self.logs.push(LogEntry {

rewatch/src/client/output.rs

@@ -324,6 +324,7 @@ This inconsistency will cause issues with package resolution.",
             | Event::BuildStarted(_)
             | Event::FileChanged(_)
             | Event::WatchPaths(_)
+            | Event::Heartbeat(_)
             | Event::FormatStarted(_)
             | Event::FormatFinished(_)
             | Event::FormatProgress(_)

rewatch/src/client/watch.rs

@@ -1,11 +1,12 @@
 use std::collections::{HashMap, HashSet};
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
-use std::time::SystemTime;
+use std::time::{Duration, SystemTime};
 
 use anyhow::Result;
 use notify::{Config, Event, RecommendedWatcher, RecursiveMode, Watcher};
 use tokio::sync::mpsc;
+use tokio::time::sleep;
 use tokio_stream::StreamExt;
 
 #[cfg(unix)]
@@ -116,23 +117,96 @@ fn setup_watchers(
     state
 }
 
-/// Run watch mode via the daemon
+/// Outcome of a single watch session.
+enum SessionExit {
+    /// User-initiated shutdown (signal). Don't retry.
+    Clean,
+    /// gRPC stream broke. Retry with reconnection.
+    StreamError(String),
+}
+
+/// Return a short description of a daemon event variant for diagnostic logging.
+fn event_description(event: &DaemonEventVariant) -> &'static str {
+    match event {
+        DaemonEventVariant::ClientConnected(_) => "ClientConnected",
+        DaemonEventVariant::ClientDisconnected(_) => "ClientDisconnected",
+        DaemonEventVariant::BuildStarted(_) => "BuildStarted",
+        DaemonEventVariant::BuildFinished(_) => "BuildFinished",
+        DaemonEventVariant::Cleaned(_) => "Cleaned",
+        DaemonEventVariant::Parsed(_) => "Parsed",
+        DaemonEventVariant::Compiling(_) => "Compiling",
+        DaemonEventVariant::Compiled(_) => "Compiled",
+        DaemonEventVariant::CleanedCompilerAssets(_) => "CleanedCompilerAssets",
+        DaemonEventVariant::CleanedJsFiles(_) => "CleanedJsFiles",
+        DaemonEventVariant::CircularDependency(_) => "CircularDependency",
+        DaemonEventVariant::UnallowedDependency(_) => "UnallowedDependency",
+        DaemonEventVariant::PackageTreeError(_) => "PackageTreeError",
+        DaemonEventVariant::ModuleNotFound(_) => "ModuleNotFound",
+        DaemonEventVariant::InitializationError(_) => "InitializationError",
+        DaemonEventVariant::CompilerWarning(_) => "CompilerWarning",
+        DaemonEventVariant::CompilerError(_) => "CompilerError",
+        DaemonEventVariant::ConfigWarning(_) => "ConfigWarning",
+        DaemonEventVariant::DuplicatedPackage(_) => "DuplicatedPackage",
+        DaemonEventVariant::MissingImplementation(_) => "MissingImplementation",
+        DaemonEventVariant::PackageNameMismatch(_) => "PackageNameMismatch",
+        DaemonEventVariant::FileChanged(_) => "FileChanged",
+        DaemonEventVariant::WatchPaths(_) => "WatchPaths",
+        DaemonEventVariant::Heartbeat(_) => "Heartbeat",
+        DaemonEventVariant::FormatStarted(_) => "FormatStarted",
+        DaemonEventVariant::FormatFinished(_) => "FormatFinished",
+        DaemonEventVariant::FormatProgress(_) => "FormatProgress",
+        DaemonEventVariant::FormatCheckFailed(_) => "FormatCheckFailed",
+        DaemonEventVariant::FormattedStdin(_) => "FormattedStdin",
+        DaemonEventVariant::JsPostBuildOutput(_) => "JsPostBuildOutput",
+    }
+}
+
+/// Run watch mode via the daemon, reconnecting on stream errors.
 pub async fn run(folder: &str, filter: Option<String>, after_build: Option<String>) -> Result<()> {
     let root = connection::find_project_root(Path::new(folder))?;
     let working_dir = Path::new(folder).canonicalize()?;
 
-    let mut client = connection::connect_or_start(&root).await?;
+    let mut delay = Duration::from_secs(1);
+    let max_delay = Duration::from_secs(30);
+
+    loop {
+        match run_watch_session(&root, &working_dir, &filter, &after_build).await {
+            Ok(SessionExit::Clean) => return Ok(()),
+            Ok(SessionExit::StreamError(msg)) => {
+                eprintln!(
+                    "Connection to daemon lost ({}). Reconnecting in {}s...",
+                    msg,
+                    delay.as_secs()
+                );
+                sleep(delay).await;
+                delay = (delay * 2).min(max_delay);
+            }
+            Err(e) => return Err(e),
+        }
+    }
+}
+
+/// Run a single watch session. Returns `SessionExit::Clean` on signal,
+/// `SessionExit::StreamError` on gRPC stream failure, or `Err` for
+/// non-recoverable errors (e.g. initialization failure).
+async fn run_watch_session(
+    root: &Path,
+    working_dir: &Path,
+    filter: &Option<String>,
+    after_build: &Option<String>,
+) -> Result<SessionExit> {
+    let mut client = connection::connect_or_start(root).await?;
 
     // Send watch request to daemon
     let request = WatchRequest {
         working_directory: working_dir.to_string_lossy().to_string(),
-        filter,
+        filter: filter.clone(),
     };
 
     let mut build_stream = client.watch(request).await?.into_inner();
 
     // Clone client for file/config change notifications
-    let mut notify_client = connection::connect(&root).await?;
+    let mut notify_client = connection::connect(root).await?;
 
     // Set up signal handlers for graceful shutdown
     #[cfg(unix)]
@@ -151,6 +225,9 @@ pub async fn run(folder: &str, filter: Option<String>, after_build: Option<Strin
     // Track client_id for graceful disconnect
     let mut client_id: Option<u64> = None;
 
+    // Track the last event received for diagnostic logging on stream errors
+    let mut last_event: &str = "none";
+
     loop {
         tokio::select! {
             // Handle SIGTERM for graceful shutdown
@@ -165,7 +242,7 @@ pub async fn run(folder: &str, filter: Option<String>, after_build: Option<Strin
                 if let Some(id) = client_id {
                     let _ = notify_client.disconnect(DisconnectRequest { client_id: id }).await;
                 }
-                break;
+                return Ok(SessionExit::Clean);
             }
 
             // Handle SIGINT (Ctrl-C) for graceful shutdown
@@ -180,7 +257,7 @@ pub async fn run(folder: &str, filter: Option<String>, after_build: Option<Strin
                 if let Some(id) = client_id {
                     let _ = notify_client.disconnect(DisconnectRequest { client_id: id }).await;
                 }
-                break;
+                return Ok(SessionExit::Clean);
             }
 
             // Handle file system events
@@ -254,6 +331,11 @@ pub async fn run(folder: &str, filter: Option<String>, after_build: Option<Strin
             Some(result) = build_stream.next() => {
                 match result {
                     Ok(event) => {
+                        // Track the last event for diagnostic logging
+                        if let Some(ref evt) = event.event {
+                            last_event = event_description(evt);
+                        }
+
                         // Check if this is a WatchPaths event — update our watchers
                         if let Some(DaemonEventVariant::WatchPaths(ref watch_paths)) = event.event {
                             let source_paths: Vec<(PathBuf, bool)> = watch_paths
@@ -275,6 +357,11 @@ pub async fn run(folder: &str, filter: Option<String>, after_build: Option<Strin
                             continue;
                         }
 
+                        // Silently consume heartbeats
+                        if matches!(event.event, Some(DaemonEventVariant::Heartbeat(_))) {
+                            continue;
+                        }
+
                         // Extract client_id from events (BuildStarted is the first event with it)
                         if client_id.is_none()
                             && let Some(DaemonEventVariant::BuildStarted(ref started)) = event.event
@@ -300,25 +387,23 @@ pub async fn run(folder: &str, filter: Option<String>, after_build: Option<Strin
 
                         // Run after_build command on successful completion
                         if is_success
-                            && let Some(ref cmd_str) = after_build
+                            && let Some(cmd_str) = after_build
                         {
                             cmd::run(cmd_str.clone());
                         }
                     }
                     Err(e) => {
-                        eprintln!("Stream error: {}", e);
-                        // Send disconnect to daemon so it can clean up
+                        eprintln!("Stream error (last event: {}): {}", last_event, e);
+                        // Best-effort disconnect — connection is likely broken
                         if let Some(id) = client_id {
                             let _ = notify_client.disconnect(DisconnectRequest { client_id: id }).await;
                         }
-                        break;
+                        return Ok(SessionExit::StreamError(e.to_string()));
                     }
                 }
             }
         }
     }
-
-    Ok(())
 }
 
 /// Check if a path is a config file (rescript.json) that we're watching.

rewatch/src/daemon.rs

@@ -9,6 +9,7 @@ mod work_queue;
 use std::collections::hash_map::DefaultHasher;
 use std::hash::{Hash, Hasher};
 use std::path::{Path, PathBuf};
+use std::time::Duration;
 
 use anyhow::Result;
 use tokio::net::UnixListener;
@@ -111,6 +112,8 @@ pub async fn start(root: PathBuf) -> Result<()> {
     };
 
     Server::builder()
+        .http2_keepalive_interval(Some(Duration::from_secs(10)))
+        .http2_keepalive_timeout(Some(Duration::from_secs(5)))
         .add_service(proto::rescript_daemon_server::RescriptDaemonServer::new(service))
         .serve_with_incoming_shutdown(uds_stream, shutdown_monitor(shutdown_rx))
         .await?;

rewatch/src/daemon/service.rs

@@ -140,6 +140,8 @@ fn event_matches_client(event: &DaemonEvent, client_id: u64) -> bool {
         Some(Event::FileChanged(_)) => true,
         // WatchPaths is targeted at the watch client
         Some(Event::WatchPaths(e)) => e.client_id == client_id,
+        // Heartbeat is targeted at a specific client
+        Some(Event::Heartbeat(e)) => e.client_id == client_id,
         None => false,
     }
 }
@@ -509,14 +511,30 @@ impl RescriptDaemon for DaemonService {
                 });
             });
 
-            while let Some(result) = event_stream.next().await {
-                match result {
-                    Ok(event) => {
-                        if event_matches_client(&event, client_id) {
-                            yield Ok(event);
+            // Send periodic heartbeats so broken connections are detected
+            // even when no build activity is happening.
+            let mut heartbeat_interval = tokio::time::interval(std::time::Duration::from_secs(30));
+            heartbeat_interval.tick().await; // consume the immediate first tick
+
+            loop {
+                tokio::select! {
+                    result = event_stream.next() => {
+                        match result {
+                            Some(Ok(event)) => {
+                                if event_matches_client(&event, client_id) {
+                                    yield Ok(event);
+                                }
+                            }
+                            Some(Err(_)) => continue,
+                            None => break,
                         }
                     }
-                    Err(_) => continue,
+                    _ = heartbeat_interval.tick() => {
+                        yield Ok(DaemonEvent {
+                            timestamp: timestamp(),
+                            event: Some(Event::Heartbeat(super::proto::Heartbeat { client_id })),
+                        });
+                    }
                 }
             }
         };