feat: abort SSE connections on hot reload

SSE streams now return an error on reload instead of ending cleanly,
causing hyper to abort the connection. This triggers client retry
behavior (e.g., Datastar reconnects automatically).

Author: cablehead

Cargo.lock

@@ -32,6 +32,7 @@ hyper-util = { version = "0.1", features = ["full", "server-auto"] }
 tokio = { version = "1", features = ["full"] }
 tokio-stream = "0.1"
 tokio-util = { version = "0.7", features = ["io", "compat"] }
+futures-util = "0.3"
 
 bytes = "1.6.0"
 url = "2.5.0"

Cargo.toml

@@ -144,7 +144,9 @@ $ http-nu :3001 -w ./serve.nu
 ```
 
 This watches the script's directory for any changes (including included files)
-and hot-reloads the handler. Useful during development.
+and hot-reloads the handler. Useful during development. Active
+[SSE connections](#server-sent-events) are aborted on reload to trigger client
+reconnection.
 
 ### Reading from stdin
 

README.md

@@ -1,6 +1,8 @@
 use std::path::Path;
 use std::sync::{atomic::AtomicBool, Arc};
 
+use tokio_util::sync::CancellationToken;
+
 use nu_cli::{add_cli_context, gather_parent_env_vars};
 use nu_cmd_lang::create_default_context;
 use nu_command::add_shell_command_context;
@@ -29,6 +31,8 @@ use crate::Error;
 pub struct Engine {
     pub state: EngineState,
     pub closure: Option<Closure>,
+    /// Cancellation token triggered on engine reload
+    pub reload_token: CancellationToken,
 }
 
 impl Engine {
@@ -47,6 +51,7 @@ impl Engine {
         Ok(Self {
             state: engine_state,
             closure: None,
+            reload_token: CancellationToken::new(),
         })
     }
 
@@ -345,6 +350,8 @@ impl Engine {
 /// On error, prints to stderr and emits JSON to stdout, returning None.
 pub fn script_to_engine(base: &Engine, script: &str) -> Option<Engine> {
     let mut engine = base.clone();
+    // Fresh cancellation token for this engine instance
+    engine.reload_token = CancellationToken::new();
 
     if let Err(e) = engine.parse_closure(script) {
         log_error(&nu_utils::strip_ansi_string_likely(e.to_string()));

src/engine.rs

@@ -3,10 +3,11 @@ use std::sync::Arc;
 use std::time::Instant;
 
 use arc_swap::ArcSwap;
+use futures_util::StreamExt;
 use http_body_util::{combinators::BoxBody, BodyExt, Empty, Full, StreamBody};
 use hyper::body::{Bytes, Frame};
 use tokio_stream::wrappers::ReceiverStream;
-use tokio_stream::StreamExt;
+use tokio_util::sync::CancellationToken;
 use tower::Service;
 use tower_http::services::{ServeDir, ServeFile};
 
@@ -135,6 +136,7 @@ where
     // Phase 1: Log request
     log_request(request_id, &request);
 
+    let reload_token = engine.reload_token.clone();
     let (meta_rx, bridged_body) = spawn_eval_thread(engine, request, stream);
 
     // Wait for both:
@@ -158,7 +160,15 @@ where
 
     match &meta.body_type {
         ResponseBodyType::Normal => {
-            build_normal_response(&meta, Ok(body_result?), use_brotli, guard, start_time).await
+            build_normal_response(
+                &meta,
+                Ok(body_result?),
+                use_brotli,
+                guard,
+                start_time,
+                reload_token,
+            )
+            .await
         }
         ResponseBodyType::Static {
             root,
@@ -327,6 +337,7 @@ async fn build_normal_response(
     use_brotli: bool,
     guard: RequestGuard,
     start_time: Instant,
+    reload_token: CancellationToken,
 ) -> HTTPResult {
     let request_id = guard.request_id();
     let (inferred_content_type, body) = body_result?;
@@ -362,7 +373,8 @@ async fn build_normal_response(
     }
 
     // Add SSE-required headers for event streams
-    if content_type == "text/event-stream" {
+    let is_sse = content_type == "text/event-stream";
+    if is_sse {
         header_map.insert(
             hyper::header::CACHE_CONTROL,
             hyper::header::HeaderValue::from_static("no-cache"),
@@ -415,9 +427,29 @@ async fn build_normal_response(
         ResponseTransport::Stream(rx) => {
             if use_brotli {
                 compression::compress_stream(rx)
+            } else if is_sse {
+                // SSE streams abort on reload (error triggers client retry)
+                let stream = futures_util::stream::try_unfold(
+                    (ReceiverStream::new(rx), reload_token),
+                    |(mut data_rx, token)| async move {
+                        tokio::select! {
+                            biased;
+                            _ = token.cancelled() => {
+                                Err(std::io::Error::other("reload").into())
+                            }
+                            item = StreamExt::next(&mut data_rx) => {
+                                match item {
+                                    Some(data) => Ok(Some((Frame::data(Bytes::from(data)), (data_rx, token)))),
+                                    None => Ok(None),
+                                }
+                            }
+                        }
+                    },
+                );
+                BodyExt::boxed(StreamBody::new(stream))
             } else {
                 let stream = ReceiverStream::new(rx).map(|data| Ok(Frame::data(Bytes::from(data))));
-                StreamBody::new(stream).boxed()
+                BodyExt::boxed(StreamBody::new(stream))
             }
         }
     };

src/handler.rs

@@ -277,6 +277,8 @@ async fn serve(
     tokio::spawn(async move {
         while let Some(script) = rx.recv().await {
             if let Some(new_engine) = script_to_engine(&base_for_updates, &script) {
+                // Signal reload to cancel SSE streams on old engine
+                engine_updater.load().reload_token.cancel();
                 engine_updater.store(Arc::new(new_engine));
                 log_reloaded();
             }

src/main.rs

@@ -2008,3 +2008,101 @@ async fn test_binary_octet_stream_content_type() {
         "Expected application/octet-stream content-type for binary, got: {response}"
     );
 }
+
+#[tokio::test]
+async fn test_sse_cancelled_on_hot_reload() {
+    use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
+
+    // Spawn server with an SSE endpoint that streams indefinitely
+    let (mut child, mut stdin, addr_rx) = TestServerWithStdin::spawn("127.0.0.1:0", false);
+
+    // Send initial SSE script - stream many events slowly
+    let sse_script = r#"{|req|
+        .response {headers: {"Content-Type": "text/event-stream"}}
+        1..100 | each {|i|
+            sleep 100ms
+            $"data: event-($i)\n\n"
+        }
+    }"#;
+    stdin.write_all(sse_script.as_bytes()).await.unwrap();
+    stdin.write_all(b"\0").await.unwrap();
+    stdin.flush().await.unwrap();
+
+    // Wait for server to start
+    let address = tokio::time::timeout(std::time::Duration::from_secs(5), addr_rx)
+        .await
+        .expect("Server didn't start")
+        .expect("Channel closed");
+
+    tokio::time::sleep(std::time::Duration::from_millis(200)).await;
+
+    // Start curl to SSE endpoint
+    let mut sse_child = tokio::process::Command::new("curl")
+        .arg("-sN")
+        .arg(&address)
+        .stdout(std::process::Stdio::piped())
+        .spawn()
+        .expect("Failed to start curl");
+
+    let stdout = sse_child.stdout.take().expect("Failed to get stdout");
+    let mut reader = BufReader::new(stdout).lines();
+
+    // Read a few events to confirm SSE is working
+    let mut events_received = 0;
+    for _ in 0..3 {
+        if let Ok(Ok(Some(line))) =
+            tokio::time::timeout(std::time::Duration::from_secs(2), reader.next_line()).await
+        {
+            if line.starts_with("data:") {
+                events_received += 1;
+            }
+        }
+    }
+    assert!(
+        events_received >= 1,
+        "Should have received at least one SSE event before reload"
+    );
+
+    // Trigger hot reload with a different script
+    let new_script = r#"{|req| "reloaded"}"#;
+    stdin.write_all(new_script.as_bytes()).await.unwrap();
+    stdin.write_all(b"\0").await.unwrap();
+    stdin.flush().await.unwrap();
+
+    // Wait for reload to process
+    tokio::time::sleep(std::time::Duration::from_millis(200)).await;
+
+    // After reload, the SSE stream should be cancelled.
+    // With HTTP keep-alive, curl won't exit on its own, but no more events should arrive.
+    // Try to read another event - it should timeout (stream cancelled) or return None.
+    let more_events =
+        tokio::time::timeout(std::time::Duration::from_millis(500), reader.next_line()).await;
+
+    // Either timeout (stream stalled) or None (stream ended) is acceptable
+    let stream_stopped = match more_events {
+        Err(_) => true,                                   // Timeout - no more data
+        Ok(Ok(None)) => true,                             // Stream ended
+        Ok(Ok(Some(line))) => !line.starts_with("data:"), // Got something but not an event
+        Ok(Err(_)) => true,                               // Read error
+    };
+    assert!(stream_stopped, "SSE stream should stop after reload");
+
+    // Kill the curl process since it won't exit with keep-alive
+    sse_child.kill().await.ok();
+
+    // Verify the new handler works
+    let output = std::process::Command::new("curl")
+        .arg("-s")
+        .arg(&address)
+        .output()
+        .expect("curl failed");
+
+    assert_eq!(
+        String::from_utf8_lossy(&output.stdout).trim(),
+        "reloaded",
+        "New handler should be active after reload"
+    );
+
+    // Cleanup - kill the server
+    child.kill().await.ok();
+}