fix: let an interrupted follow read stop instead of parking for the next frame

Author: cablehead

src/nu/commands/cat_command.rs

@@ -170,13 +170,16 @@ impl Command for CatCommand {
             // shared runtime; the consumer dropping the ListStream cancels it
             // (the L1 fd-leak fix). Driven off the runtime in real use.
             let mut rx = self.store.read(options);
+            let rt = self.store.runtime();
+            let signals = engine_state.signals().clone();
+            let waiting = signals.clone();
             let stream = ListStream::new(
                 std::iter::from_fn(move || {
-                    let frame = rx.blocking_recv()?; // parks off-runtime; None when producer done/cancelled
+                    let frame = crate::nu::util::follow_recv(&mut rx, rt.as_ref(), &waiting)?;
                     Some(to_value(&frame))
                 }),
                 span,
-                Signals::empty(),
+                signals,
             );
             return Ok(PipelineData::ListStream(stream, None));
         }

src/nu/commands/last_command.rs

@@ -1,7 +1,7 @@
 use nu_engine::CallExt;
 use nu_protocol::engine::{Call, Command, EngineState, Stack};
 use nu_protocol::{
-    Category, ListStream, PipelineData, ShellError, Signals, Signature, SyntaxShape, Type, Value,
+    Category, ListStream, PipelineData, ShellError, Signature, SyntaxShape, Type, Value,
 };
 
 use crate::nu::util;
@@ -90,13 +90,16 @@ impl Command for LastCommand {
                 .build();
 
             let mut rx = self.store.read(options);
+            let rt = self.store.runtime();
+            let signals = engine_state.signals().clone();
+            let waiting = signals.clone();
             let stream = ListStream::new(
                 std::iter::from_fn(move || {
-                    let frame = rx.blocking_recv()?; // parks off-runtime; None when producer done/cancelled
+                    let frame = util::follow_recv(&mut rx, rt.as_ref(), &waiting)?;
                     Some(util::frame_to_value(&frame, span, with_timestamp))
                 }),
                 span,
-                Signals::empty(),
+                signals,
             );
 
             return Ok(PipelineData::ListStream(stream, None));

src/nu/test_commands.rs

@@ -763,6 +763,88 @@ mod tests {
     // socket. That runtime is only dropped when the block_on future returns,
     // which requires the forwarding loop to exit.
     //
+    // A following read has to notice interruption. Its iterator parks
+    // waiting for the next frame on the topic, and on an idle topic that wait
+    // has no end. http-nu gives each request its own `Signals` and triggers
+    // them when the client disconnects, so without this the thread and the
+    // store's follow task stay parked until some unrelated append happens to
+    // land on that topic.
+    #[test]
+    fn test_cat_follow_stops_when_interrupted() {
+        use crate::nu::commands;
+        use nu_protocol::Signals;
+        use std::sync::atomic::{AtomicBool, Ordering};
+        use std::sync::Arc;
+        use std::time::Duration;
+
+        let (store, mut engine) = setup_test_env();
+        engine
+            .add_commands(vec![Box::new(commands::cat_command::CatCommand::new(
+                store.clone(),
+            ))])
+            .unwrap();
+
+        let flag = Arc::new(AtomicBool::new(false));
+        engine.state.set_signals(Signals::new(flag.clone()));
+
+        // Nothing is ever appended to this topic, so the only way out of the
+        // follow is the signal.
+        let (done_tx, done_rx) = std::sync::mpsc::channel();
+        let engine_clone = engine.clone();
+        std::thread::spawn(move || {
+            let out = engine_clone.eval(PipelineData::empty(), ".cat -f -T idle".to_string());
+            let drained = out.map(|d| d.into_iter().count());
+            let _ = done_tx.send(drained.is_ok());
+        });
+
+        // Let it reach the park before interrupting.
+        std::thread::sleep(Duration::from_millis(200));
+        flag.store(true, Ordering::SeqCst);
+
+        let finished = done_rx.recv_timeout(Duration::from_secs(5));
+        assert!(
+            finished.is_ok(),
+            "a following read ignored its interrupt and stayed parked",
+        );
+    }
+
+    // `.last -f` has the same follow branch and the same wait.
+    #[test]
+    fn test_last_follow_stops_when_interrupted() {
+        use crate::nu::commands;
+        use nu_protocol::Signals;
+        use std::sync::atomic::{AtomicBool, Ordering};
+        use std::sync::Arc;
+        use std::time::Duration;
+
+        let (store, mut engine) = setup_test_env();
+        engine
+            .add_commands(vec![Box::new(commands::last_command::LastCommand::new(
+                store.clone(),
+            ))])
+            .unwrap();
+
+        let flag = Arc::new(AtomicBool::new(false));
+        engine.state.set_signals(Signals::new(flag.clone()));
+
+        let (done_tx, done_rx) = std::sync::mpsc::channel();
+        let engine_clone = engine.clone();
+        std::thread::spawn(move || {
+            let out = engine_clone.eval(PipelineData::empty(), ".last idle -f".to_string());
+            // Assert on the count, not just that the thread ended: a bad
+            // invocation would error out at once and pass vacuously.
+            let _ = done_tx.send(out.map(|d| d.into_iter().count()));
+        });
+
+        std::thread::sleep(Duration::from_millis(200));
+        flag.store(true, Ordering::SeqCst);
+
+        let finished = done_rx
+            .recv_timeout(Duration::from_secs(5))
+            .expect("a following read ignored its interrupt and stayed parked");
+        assert!(finished.is_ok(), ".last -f failed to run: {finished:?}");
+    }
+
     // In `--follow` mode the store's read() future parks on the broadcast
     // receiver waiting for the next append. When the ListStream consumer is
     // dropped (client disconnects, pipeline ends early), the spawned thread is

src/nu/util.rs

@@ -271,3 +271,38 @@ pub fn write_pipeline_to_cas(
         PipelineData::Empty => Ok(None),
     }
 }
+
+/// How long a following read waits before it checks whether it was
+/// interrupted. The wait ends the moment a frame arrives, so this bounds only
+/// how late an interrupt is noticed, not how late a frame is delivered.
+const FOLLOW_INTERRUPT_CHECK: std::time::Duration = std::time::Duration::from_millis(100);
+
+/// The next frame of a following read, or `None` when the producer is done or
+/// the read was interrupted.
+///
+/// A follow parks until the next frame lands on its topic, and on a quiet
+/// topic that never happens. An unbounded wait therefore ignores interruption
+/// completely: the thread, and the store's follow task behind it, stay parked
+/// until some unrelated append wakes them. Waiting in slices fixes that and
+/// costs nothing, because the timeout returns as soon as a frame arrives.
+///
+/// `rt` is the store's runtime. Without one there is nothing to drive the
+/// timeout on, so the wait falls back to blocking and stays uninterruptible.
+pub fn follow_recv(
+    rx: &mut tokio::sync::mpsc::Receiver<crate::store::Frame>,
+    rt: Option<&tokio::runtime::Handle>,
+    signals: &nu_protocol::Signals,
+) -> Option<crate::store::Frame> {
+    let Some(rt) = rt else {
+        return rx.blocking_recv();
+    };
+    loop {
+        if signals.interrupted() {
+            return None;
+        }
+        match rt.block_on(async { tokio::time::timeout(FOLLOW_INTERRUPT_CHECK, rx.recv()).await }) {
+            Ok(frame) => return frame,
+            Err(_elapsed) => continue,
+        }
+    }
+}

src/store/mod.rs

@@ -994,6 +994,14 @@ impl Store {
         ]
     }
 
+    /// The runtime this store's tasks run on, if it was opened inside one.
+    ///
+    /// A following read needs it to wait on its receiver with a timeout, so
+    /// it can notice an interrupt instead of parking until the next frame.
+    pub fn runtime(&self) -> Option<tokio::runtime::Handle> {
+        self.rt.clone()
+    }
+
     /// The [`Fsync`] policy this store was opened with.
     pub fn fsync(&self) -> Fsync {
         self.fsync.mode