fix tests and add spawn link

Author: unbalancedparentheses

concurrency/src/registry.rs

@@ -193,12 +193,17 @@ pub fn clear() {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use std::sync::Mutex;
 
-    // Helper to ensure test isolation
+    // Mutex to serialize tests that need an isolated registry
+    static TEST_MUTEX: Mutex<()> = Mutex::new(());
+
+    // Helper to ensure test isolation - clears registry and holds lock
     fn with_clean_registry<F, R>(f: F) -> R
     where
         F: FnOnce() -> R,
     {
+        let _guard = TEST_MUTEX.lock().unwrap();
         clear();
         let result = f();
         clear();
@@ -325,33 +330,28 @@ mod tests {
 
     #[test]
     fn test_count() {
-        // Note: Due to parallel test execution, we can't rely on absolute counts.
-        // Instead, we verify that registration increases count and unregistration decreases it.
-        let pid1 = Pid::new();
-        let pid2 = Pid::new();
-
-        // Use unique names to avoid conflicts with parallel tests
-        let name1 = format!("count_test_{}", pid1.id());
-        let name2 = format!("count_test_{}", pid2.id());
-
-        let count_before = count();
-        register(&name1, pid1).unwrap();
-        let count_after_first = count();
-        assert!(
-            count_after_first > count_before,
-            "Count should increase after registration"
-        );
-
-        register(&name2, pid2).unwrap();
-        let count_after_second = count();
-        assert!(
-            count_after_second > count_after_first,
-            "Count should increase after second registration"
-        );
-
-        // Clean up our registrations
-        unregister(&name1);
-        unregister(&name2);
+        // Use with_clean_registry for test isolation
+        with_clean_registry(|| {
+            let pid1 = Pid::new();
+            let pid2 = Pid::new();
+
+            let name1 = format!("count_test_{}", pid1.id());
+            let name2 = format!("count_test_{}", pid2.id());
+
+            assert_eq!(count(), 0, "Registry should be empty");
+
+            register(&name1, pid1).unwrap();
+            assert_eq!(count(), 1, "Count should be 1 after first registration");
+
+            register(&name2, pid2).unwrap();
+            assert_eq!(count(), 2, "Count should be 2 after second registration");
+
+            unregister(&name1);
+            assert_eq!(count(), 1, "Count should be 1 after unregistration");
+
+            unregister(&name2);
+            assert_eq!(count(), 0, "Count should be 0 after all unregistrations");
+        });
     }
 
     #[test]

concurrency/src/tasks/gen_server.rs

@@ -412,24 +412,65 @@ pub trait GenServer: Send + Sized {
         GenServerHandle::new(self)
     }
 
-    /// Tokio tasks depend on a coolaborative multitasking model. "work stealing" can't
-    /// happen if the task is blocking the thread. As such, for sync compute task
+    /// Tokio tasks depend on a collaborative multitasking model. "Work stealing" can't
+    /// happen if the task is blocking the thread. As such, for sync compute tasks
     /// or other blocking tasks need to be in their own separate thread, and the OS
     /// will manage them through hardware interrupts.
-    /// Start blocking provides such thread.
+    /// `start_blocking` provides such a thread.
     fn start_blocking(self) -> GenServerHandle<Self> {
         GenServerHandle::new_blocking(self)
     }
 
-    /// For some "singleton" GenServers that run througout the whole execution of the
+    /// For some "singleton" GenServers that run throughout the whole execution of the
     /// program, it makes sense to run in their own dedicated thread to avoid interference
     /// with the rest of the tasks' runtime.
-    /// The use of tokio::task::spawm_blocking is not recommended for these scenarios
-    /// as it is a limited thread pool better suited for blocking IO tasks that eventually end
+    /// The use of `tokio::task::spawn_blocking` is not recommended for these scenarios
+    /// as it is a limited thread pool better suited for blocking IO tasks that eventually end.
     fn start_on_thread(self) -> GenServerHandle<Self> {
         GenServerHandle::new_on_thread(self)
     }
 
+    /// Start the GenServer and create a bidirectional link with another process.
+    ///
+    /// This is equivalent to calling `start()` followed by `link()`, but as an
+    /// atomic operation. If the link fails, the GenServer is stopped.
+    ///
+    /// # Example
+    ///
+    /// ```ignore
+    /// let parent = ParentServer::new().start();
+    /// let child = ChildServer::new().start_linked(&parent)?;
+    /// // Now if either crashes, the other will be notified
+    /// ```
+    fn start_linked(self, other: &impl HasPid) -> Result<GenServerHandle<Self>, LinkError> {
+        let handle = self.start();
+        handle.link(other)?;
+        Ok(handle)
+    }
+
+    /// Start the GenServer and set up monitoring from another process.
+    ///
+    /// This is equivalent to calling `start()` followed by `monitor()`, but as an
+    /// atomic operation. The monitoring process will receive a DOWN message when
+    /// this GenServer exits.
+    ///
+    /// # Example
+    ///
+    /// ```ignore
+    /// let supervisor = SupervisorServer::new().start();
+    /// let (worker, monitor_ref) = WorkerServer::new().start_monitored(&supervisor)?;
+    /// // supervisor will receive DOWN message when worker exits
+    /// ```
+    fn start_monitored(
+        self,
+        monitor_from: &impl HasPid,
+    ) -> Result<(GenServerHandle<Self>, MonitorRef), LinkError> {
+        let handle = self.start();
+        let monitor_ref = monitor_from.pid();
+        let actual_ref = process_table::monitor(monitor_ref, handle.pid())?;
+        Ok((handle, actual_ref))
+    }
+
     fn run(
         self,
         handle: &GenServerHandle<Self>,

concurrency/src/threads/gen_server.rs

@@ -290,10 +290,10 @@ pub enum InitResult<G: GenServer> {
 }
 
 pub trait GenServer: Send + Sized {
-    type CallMsg: Clone + Send + Sized;
-    type CastMsg: Clone + Send + Sized;
+    type CallMsg: Clone + Send + Sized + Sync;
+    type CastMsg: Clone + Send + Sized + Sync;
     type OutMsg: Send + Sized;
-    type Error: Debug;
+    type Error: Debug + Send;
 
     fn start(self) -> GenServerHandle<Self> {
         GenServerHandle::new(self)
@@ -305,6 +305,31 @@ pub trait GenServer: Send + Sized {
         GenServerHandle::new(self)
     }
 
+    /// Start the GenServer and create a bidirectional link with another process.
+    ///
+    /// This is equivalent to calling `start()` followed by `link()`, but as an
+    /// atomic operation. If the link fails, the GenServer is stopped.
+    fn start_linked(self, other: &impl HasPid) -> Result<GenServerHandle<Self>, LinkError> {
+        let handle = self.start();
+        handle.link(other)?;
+        Ok(handle)
+    }
+
+    /// Start the GenServer and set up monitoring from another process.
+    ///
+    /// This is equivalent to calling `start()` followed by `monitor()`, but as an
+    /// atomic operation. The monitoring process will receive a DOWN message when
+    /// this GenServer exits.
+    fn start_monitored(
+        self,
+        monitor_from: &impl HasPid,
+    ) -> Result<(GenServerHandle<Self>, MonitorRef), LinkError> {
+        let handle = self.start();
+        let monitor_ref = monitor_from.pid();
+        let actual_ref = process_table::monitor(monitor_ref, handle.pid())?;
+        Ok((handle, actual_ref))
+    }
+
     fn run(
         self,
         handle: &GenServerHandle<Self>,