fix: relay Signals reset to plugins (#13510)

This PR will close #13501

# Description

This PR expands on [the relay of signals to running plugin
processes](https://github.com/nushell/nushell/pull/13181). The Ctrlc
relay has been generalized to SignalAction::Interrupt and when
reset_signal is called on the main EngineState, a SignalAction::Reset is
now relayed to running plugins.

# User-Facing Changes

The signal handler closure now takes a `signals::SignalAction`, while
previously it took no arguments. The handler will now be called on both
interrupt and reset. The method to register a handler on the plugin side
is now called `register_signal_handler` instead of
`register_ctrlc_handler`
[example](https://github.com/nushell/nushell/pull/13510/files#diff-3e04dff88fd0780a49778a3d1eede092ec729a1264b4ef07ca0d2baa859dad05L38).
This will only affect plugin authors who have started making use of
https://github.com/nushell/nushell/pull/13181, which isn't currently
part of an official release.

The change will also require all of user's plugins to be recompiled in
order that they don't error when a signal is received on the
PluginInterface.

# Testing

```
: example ctrlc
interrupt status: false
waiting for interrupt signal...
^Cinterrupt status: true
peace.
Error:   × Operation interrupted
   ╭─[display_output hook:1:1]
 1 │ if (term size).columns >= 100 { table -e } else { table }
   · ─┬
   ·  ╰── This operation was interrupted
   ╰────

: example ctrlc
interrupt status: false   <-- NOTE status is false
waiting for interrupt signal...
^Cinterrupt status: true
peace.
Error:   × Operation interrupted
   ╭─[display_output hook:1:1]
 1 │ if (term size).columns >= 100 { table -e } else { table }
   · ─┬
   ·  ╰── This operation was interrupted
   ╰────
   ```

crates/nu-protocol/src/pipeline/handlers.rs

@@ -0,0 +1,139 @@
+use std::fmt::Debug;
+use std::sync::{Arc, Mutex};
+
+use crate::{engine::Sequence, ShellError, SignalAction};
+
+/// Handler is a closure that can be sent across threads and shared.
+pub type Handler = Box<dyn Fn(SignalAction) + Send + Sync>;
+
+/// Manages a collection of handlers.
+#[derive(Clone)]
+pub struct Handlers {
+    /// List of handler tuples containing an ID and the handler itself.
+    handlers: Arc<Mutex<Vec<(usize, Handler)>>>,
+    /// Sequence generator for unique IDs.
+    next_id: Arc<Sequence>,
+}
+
+impl Debug for Handlers {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("Handlers")
+            .field("next_id", &self.next_id)
+            .finish()
+    }
+}
+
+/// HandlerGuard that unregisters a handler when dropped.
+#[derive(Clone)]
+pub struct HandlerGuard {
+    /// Unique ID of the handler.
+    id: usize,
+    /// Reference to the handlers list.
+    handlers: Arc<Mutex<Vec<(usize, Handler)>>>,
+}
+
+impl Drop for HandlerGuard {
+    /// Drops the `Guard`, removing the associated handler from the list.
+    fn drop(&mut self) {
+        if let Ok(mut handlers) = self.handlers.lock() {
+            handlers.retain(|(id, _)| *id != self.id);
+        }
+    }
+}
+
+impl Debug for HandlerGuard {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("Guard").field("id", &self.id).finish()
+    }
+}
+
+impl Handlers {
+    pub fn new() -> Handlers {
+        let handlers = Arc::new(Mutex::new(vec![]));
+        let next_id = Arc::new(Sequence::default());
+        Handlers { handlers, next_id }
+    }
+
+    /// Registers a new handler and returns an RAII guard which will unregister the handler when
+    /// dropped.
+    pub fn register(&self, handler: Handler) -> Result<HandlerGuard, ShellError> {
+        let id = self.next_id.next()?;
+        if let Ok(mut handlers) = self.handlers.lock() {
+            handlers.push((id, handler));
+        }
+
+        Ok(HandlerGuard {
+            id,
+            handlers: Arc::clone(&self.handlers),
+        })
+    }
+
+    /// Runs all registered handlers.
+    pub fn run(&self, action: SignalAction) {
+        if let Ok(handlers) = self.handlers.lock() {
+            for (_, handler) in handlers.iter() {
+                handler(action);
+            }
+        }
+    }
+}
+
+impl Default for Handlers {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::atomic::{AtomicBool, Ordering};
+
+    #[test]
+    /// Tests registering and running multiple handlers.
+    fn test_multiple_handlers() {
+        let handlers = Handlers::new();
+        let called1 = Arc::new(AtomicBool::new(false));
+        let called2 = Arc::new(AtomicBool::new(false));
+
+        let called1_clone = Arc::clone(&called1);
+        let called2_clone = Arc::clone(&called2);
+
+        let _guard1 = handlers.register(Box::new(move |_| {
+            called1_clone.store(true, Ordering::SeqCst);
+        }));
+        let _guard2 = handlers.register(Box::new(move |_| {
+            called2_clone.store(true, Ordering::SeqCst);
+        }));
+
+        handlers.run(SignalAction::Interrupt);
+
+        assert!(called1.load(Ordering::SeqCst));
+        assert!(called2.load(Ordering::SeqCst));
+    }
+
+    #[test]
+    /// Tests the dropping of a guard and ensuring the handler is unregistered.
+    fn test_guard_drop() {
+        let handlers = Handlers::new();
+        let called = Arc::new(AtomicBool::new(false));
+        let called_clone = Arc::clone(&called);
+
+        let guard = handlers.register(Box::new(move |_| {
+            called_clone.store(true, Ordering::Relaxed);
+        }));
+
+        // Ensure the handler is registered
+        assert_eq!(handlers.handlers.lock().unwrap().len(), 1);
+
+        drop(guard);
+
+        // Ensure the handler is removed after dropping the guard
+        assert_eq!(handlers.handlers.lock().unwrap().len(), 0);
+
+        handlers.run(SignalAction::Interrupt);
+
+        // Ensure the handler is not called after being dropped
+        assert!(!called.load(Ordering::Relaxed));
+    }
+}

crates/nu-protocol/src/pipeline/mod.rs

@@ -1,4 +1,5 @@
 pub mod byte_stream;
+mod handlers;
 pub mod list_stream;
 mod metadata;
 mod out_dest;
@@ -6,6 +7,7 @@ mod pipeline_data;
 mod signals;
 
 pub use byte_stream::*;
+pub use handlers::*;
 pub use list_stream::*;
 pub use metadata::*;
 pub use out_dest::*;

crates/nu-protocol/src/pipeline/signals.rs

@@ -4,6 +4,8 @@ use std::sync::{
     Arc,
 };
 
+use serde::{Deserialize, Serialize};
+
 /// Used to check for signals to suspend or terminate the execution of Nushell code.
 ///
 /// For now, this struct only supports interruption (ctrl+c or SIGINT).
@@ -75,9 +77,17 @@ impl Signals {
         self.signals.is_none()
     }
 
-    pub(crate) fn reset(&self) {
+    pub fn reset(&self) {
         if let Some(signals) = &self.signals {
             signals.store(false, Ordering::Relaxed);
         }
     }
 }
+
+/// The types of things that can be signaled. It's anticipated this will change as we learn more
+/// about how we'd like signals to be handled.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+pub enum SignalAction {
+    Interrupt,
+    Reset,
+}