fix: stop leaking a tokio runtime per .cat/.last, consolidate read commands

src/api.rs

@@ -282,7 +282,7 @@ async fn handle_stream_cat(
     accept_type: AcceptType,
     with_timestamp: bool,
 ) -> HTTPResult {
-    let rx = store.read(options).await;
+    let rx = store.read(options);
     let stream = ReceiverStream::new(rx);
 
     let accept_type_clone = accept_type.clone();
@@ -564,15 +564,13 @@ async fn handle_last_get(
 
     // Follow mode: use ReadOptions::last to get historical + live frames
     // This emits xs.threshold after historical frames
-    let rx = store
-        .read(
-            ReadOptions::builder()
-                .last(last)
-                .maybe_topic(topic.map(|t| t.to_string()))
-                .follow(FollowOption::On)
-                .build(),
-        )
-        .await;
+    let rx = store.read(
+        ReadOptions::builder()
+            .last(last)
+            .maybe_topic(topic.map(|t| t.to_string()))
+            .follow(FollowOption::On)
+            .build(),
+    );
 
     let stream = tokio_stream::wrappers::ReceiverStream::new(rx).map(move |frame| {
         let mut bytes = serialize_frame(&frame, with_timestamp).into_bytes();
@@ -637,7 +635,7 @@ fn empty() -> BoxBody<Bytes, BoxError> {
 /// Direct `.append`) plus the VFS modules registered so far, so eval scripts
 /// can `use` them, the same builtins the runners get.
 fn eval_engine(store: &Store) -> Result<nu::Engine, String> {
-    let mut engine = nu::prepared_base(store, nu::ReadMode::Stream, true)
+    let mut engine = nu::prepared_base(store, true)
         .map_err(|e| format!("Failed to build nushell engine: {e}"))?;
     // Modules registered up to now: a fresh time-ordered id exceeds every
     // already-appended frame.
@@ -655,10 +653,19 @@ async fn handle_eval(store: &Store, body: hyper::body::Incoming) -> HTTPResult {
 
     let engine = eval_engine(store)?;
 
-    // Execute the script
-    let result = engine
-        .eval(nu_protocol::PipelineData::empty(), script)
-        .map_err(|e| format!("Script evaluation failed:\n{e}"))?;
+    // Execute the script on a dedicated thread, not on this tokio runtime
+    // thread. handle_eval is an async handler, so engine.eval would otherwise
+    // run on a runtime thread. The script can call .cat/.last, whose historical
+    // path parks on tokio's blocking_recv, which panics ("Cannot block the
+    // current thread from within a runtime") on a runtime thread. Running the
+    // eval on a plain std::thread avoids that.
+    let result = std::thread::scope(|scope| {
+        scope
+            .spawn(|| engine.eval(nu_protocol::PipelineData::empty(), script))
+            .join()
+            .expect("eval thread panicked")
+    })
+    .map_err(|e| format!("Script evaluation failed:\n{e}"))?;
 
     // Format output based on PipelineData type according to spec
     match result {
@@ -916,12 +923,12 @@ mod tests {
         // Add streaming commands
         engine
             .add_commands(vec![
-                Box::new(
-                    crate::nu::commands::cat_stream_command::CatStreamCommand::new(store.clone()),
-                ),
-                Box::new(
-                    crate::nu::commands::last_stream_command::LastStreamCommand::new(store.clone()),
-                ),
+                Box::new(crate::nu::commands::cat_command::CatCommand::new(
+                    store.clone(),
+                )),
+                Box::new(crate::nu::commands::last_command::LastCommand::new(
+                    store.clone(),
+                )),
                 Box::new(crate::nu::commands::append_command::AppendCommand::new(
                     store.clone(),
                 )),

src/lib.rs

@@ -47,9 +47,7 @@
 //! )?;
 //!
 //! // Replay history, then follow live appends to drive a UI.
-//! let mut rx = store
-//!     .read(ReadOptions::builder().follow(FollowOption::On).build())
-//!     .await;
+//! let mut rx = store.read(ReadOptions::builder().follow(FollowOption::On).build());
 //! while let Some(frame) = rx.recv().await {
 //!     println!("{} {}", frame.id, frame.topic);
 //! }

src/nu/commands/cat_command.rs

@@ -1,9 +1,12 @@
 use nu_engine::CallExt;
 use nu_protocol::engine::{Call, Command, EngineState, Stack};
 use nu_protocol::shell_error::generic::GenericError;
-use nu_protocol::{Category, PipelineData, ShellError, Signature, SyntaxShape, Type};
+use nu_protocol::{
+    Category, ListStream, PipelineData, ShellError, Signals, Signature, SyntaxShape, Type, Value,
+};
+use std::time::Duration;
 
-use crate::store::{ReadOptions, Store};
+use crate::store::{FollowOption, ReadOptions, Store};
 
 #[derive(Clone)]
 pub struct CatCommand {
@@ -16,6 +19,22 @@ impl CatCommand {
     }
 }
 
+// Parse a Scru128Id, boxing the (large) ShellError so the returned Result stays
+// small (avoids clippy::result_large_err). Callers unbox at the `?` boundary.
+fn parse_id(
+    s: &str,
+    name: &str,
+    span: nu_protocol::Span,
+) -> Result<scru128::Scru128Id, Box<ShellError>> {
+    s.parse().map_err(|e| {
+        Box::new(ShellError::Generic(GenericError::new(
+            format!("Invalid {name}"),
+            format!("Failed to parse Scru128Id: {e}"),
+            span,
+        )))
+    })
+}
+
 impl Command for CatCommand {
     fn name(&self) -> &str {
         ".cat"
@@ -24,6 +43,15 @@ impl Command for CatCommand {
     fn signature(&self) -> Signature {
         Signature::build(".cat")
             .input_output_types(vec![(Type::Nothing, Type::Any)])
+            .switch("follow", "long poll for new events", Some('f'))
+            .named(
+                "pulse",
+                SyntaxShape::Int,
+                "interval in ms for synthetic xs.pulse events",
+                Some('p'),
+            )
+            .switch("new", "skip existing, only show new", Some('n'))
+            .switch("detail", "include all frame fields", Some('d'))
             .named(
                 "limit",
                 SyntaxShape::Int,
@@ -76,52 +104,97 @@ impl Command for CatCommand {
         call: &Call,
         _input: PipelineData,
     ) -> Result<PipelineData, ShellError> {
-        let limit: Option<usize> = call.get_flag(engine_state, stack, "limit")?;
-        let last: Option<usize> = call.get_flag(engine_state, stack, "last")?;
+        let follow = call.has_flag(engine_state, stack, "follow")?;
+        let pulse: Option<i64> = call.get_flag(engine_state, stack, "pulse")?;
+        let new = call.has_flag(engine_state, stack, "new")?;
+        let detail = call.has_flag(engine_state, stack, "detail")?;
+        let with_timestamp = call.has_flag(engine_state, stack, "with-timestamp")?;
+        let limit: Option<i64> = call.get_flag(engine_state, stack, "limit")?;
+        let last: Option<i64> = call.get_flag(engine_state, stack, "last")?;
         let after: Option<String> = call.get_flag(engine_state, stack, "after")?;
         let from: Option<String> = call.get_flag(engine_state, stack, "from")?;
         let topic: Option<nu_protocol::Value> = call.get_flag(engine_state, stack, "topic")?;
         let topic: Option<String> = topic
             .map(crate::nu::util::topic_value_to_string)
             .transpose()?;
-        let with_timestamp = call.has_flag(engine_state, stack, "with-timestamp")?;
 
-        // Helper to parse Scru128Id
-        let parse_id = |s: &str, name: &str| -> Result<scru128::Scru128Id, ShellError> {
-            s.parse().map_err(|e| {
-                ShellError::Generic(GenericError::new(
-                    format!("Invalid {name}"),
-                    format!("Failed to parse Scru128Id: {e}"),
-                    call.head,
-                ))
-            })
-        };
+        let span = call.head;
 
-        let after: Option<scru128::Scru128Id> =
-            after.as_deref().map(|s| parse_id(s, "after")).transpose()?;
-        let from: Option<scru128::Scru128Id> =
-            from.as_deref().map(|s| parse_id(s, "from")).transpose()?;
+        let after: Option<scru128::Scru128Id> = after
+            .as_deref()
+            .map(|s| parse_id(s, "after", span))
+            .transpose()
+            .map_err(|e| *e)?;
+        let from: Option<scru128::Scru128Id> = from
+            .as_deref()
+            .map(|s| parse_id(s, "from", span))
+            .transpose()
+            .map_err(|e| *e)?;
 
+        // Build ReadOptions
+        let following = pulse.is_some() || follow;
         let options = ReadOptions::builder()
+            .follow(if let Some(pulse_ms) = pulse {
+                FollowOption::WithHeartbeat(Duration::from_millis(pulse_ms as u64))
+            } else if follow {
+                FollowOption::On
+            } else {
+                FollowOption::Off
+            })
+            .new(new)
             .maybe_after(after)
             .maybe_from(from)
-            .maybe_limit(limit)
-            .maybe_last(last)
+            .maybe_limit(limit.map(|l| l as usize))
+            .maybe_last(last.map(|l| l as usize))
             .maybe_topic(topic)
             .build();
 
-        let frames: Vec<_> = self.store.read_sync(options).collect();
+        // Shape one frame into a Value, stripping `ttl` unless `--detail`.
+        let to_value = move |frame: &crate::store::Frame| {
+            let value = crate::nu::util::frame_to_value(frame, span, with_timestamp);
+            if detail {
+                return value;
+            }
+            match value {
+                Value::Record { val, .. } => {
+                    let mut filtered = val.into_owned();
+                    filtered.remove("ttl");
+                    Value::record(filtered, span)
+                }
+                v => v,
+            }
+        };
 
-        use nu_protocol::Value;
+        if following {
+            // Follow mode: stream lazily. The follow/heartbeat task runs on the
+            // 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 stream = ListStream::new(
+                std::iter::from_fn(move || {
+                    let frame = rx.blocking_recv()?; // parks off-runtime; None when producer done/cancelled
+                    Some(to_value(&frame))
+                }),
+                span,
+                Signals::empty(),
+            );
+            return Ok(PipelineData::ListStream(stream, None));
+        }
 
-        let output = Value::list(
-            frames
-                .into_iter()
-                .map(|frame| crate::nu::util::frame_to_value(&frame, call.head, with_timestamp))
-                .collect(),
-            call.head,
+        // Historical mode: stream lazily, same shape as the follow branch. The
+        // producer closes the channel once replay completes, so from_fn ends.
+        // blocking_recv parks the caller thread; callers that reach `.cat`
+        // during an actor's async setup run the config eval on a dedicated
+        // thread (see parse_config), so this never parks a runtime thread.
+        let mut rx = self.store.read(options);
+        let stream = ListStream::new(
+            std::iter::from_fn(move || {
+                let frame = rx.blocking_recv()?; // None when the producer finishes replay
+                Some(to_value(&frame))
+            }),
+            span,
+            Signals::empty(),
         );
-
-        Ok(PipelineData::Value(output, None))
+        Ok(PipelineData::ListStream(stream, None))
     }
 }