Add tests. Fix async problems.

Author: johnteee

CHANGELOG.md

@@ -39,6 +39,12 @@ All notable changes to this project are documented here. The format loosely foll
 - **`WillAsync` / `CountDownLatch` woke only one awaiter** — the single `Option<Waker>` slot dropped all but the last-registered waker, hanging concurrent awaiters; now stores `Vec<Waker>` and wakes all (draining before waking to stay reentrancy-safe). `WillAsync::start()` now also wakes wakers registered before `start()`, so poll-before-start no longer hangs.
 - **`SubscriptionFunc` buffered every value forever (memory leak)** — under `for_futures`, `on_next` unconditionally pushed each delivered value into an internal `cached` stream, but that stream was open from construction (`LinkedListAsync::new()` starts `alive=true`). A callback-only subscription (the common `subscribe_fn` case) has no stream consumer, so the buffer grew without bound. The `cached` stream now starts **closed** and only begins buffering once `as_stream()` opens it — matching the documented open/close intent. Streaming (`subscribe_as_stream`) is unaffected. Regression-guarded by `test_common_subscription_func_no_buffer_until_stream_opened`.
 - **`BlockingQueue::stop` doc/dead-code mismatch** — the method and module docs claimed it "drops the sender side" to close the queue, but the code did `let sender = …lock().unwrap(); drop(sender);` which dropped the `MutexGuard`, not the `Sender` (which lives behind `Arc<Mutex<..>>` and cannot be dropped there). Removed the misleading no-op and corrected both docs to state the real mechanism: `stop()` flips the `alive` flag (gating later `offer`/`take`), and a consumer already blocked in `take()` stays blocked until its own `timeout`. No behavior change (verified by `test_sync_blocking_queue_is_alive_and_stop`).
+- **`HandlerThreadInner::post` accepted work after stop** — `Handler::stop` promised "stop accepting new jobs," but `post()` always enqueued into the inner `BlockingQueue`. A post after stop could leak forever into an un-drained queue or wake a worker blocked in `take()` and execute one more job after stop. `post()` now rejects only the stopped state (`started && !alive`), preserving pre-start queueing. Regression-guarded by `test_handler_inner_post_after_stop_is_rejected` and `test_handler_post_before_start_is_preserved`.
+- **`ActorAsync::stop` left mailbox handles accepting sends** — stopping an actor cleared the actor alive flag but left the mailbox `BlockingQueue` alive, so `HandleAsync::send` after stop still queued messages that would never be processed. `stop()` now also stops the mailbox queue so later sends are rejected, while still not interrupting a thread already blocked in `take()` (shutdown redesign remains deferred). Regression-guarded by `test_actor_handle_send_after_stop_is_rejected`.
+- **`CountDownLatch::wait` released only one blocking waiter** — `countdown()` used `Condvar::notify_one()`, so when the count reached zero only one waiting thread was guaranteed to wake. It now uses `notify_all()` at zero, matching Java `CountDownLatch` semantics and the existing async multi-waker behavior. Regression-guarded by `test_sync_countdownlatch_releases_all_waiters`.
+- **Reentrant waker deadlock risks** — `WillAsync`, `CountDownLatch`, and `LinkedListAsync` now release state/waker locks before invoking `wake()`/`wake_all_wakers()`/`notify_all()`, so synchronous re-polling wakers cannot block on the same non-reentrant mutexes during wakeup. Verified by existing async stream/future tests.
+- **`Cor::stop` vestigial no-op** — `stop()` ran `drop(self.op_ch_sender.lock().unwrap());`, which (like the old `BlockingQueue::stop`) dropped the `MutexGuard`, not the shared `Sender`, so it never closed the op channel. Removed the no-op and clarified that cooperative stop is delivered solely by the `alive` flag (a `yield_from` to a stopped `Cor` returns `None` because `receive()` early-returns). Behavior-preserving; also removes one lock held under `started_alive`. Verified by `test_cor_*` (pure + `test_runtime`).
+- **`Publisher::delete_observer` closed streams for non-members and duplicates too early** — stream closure happened before confirming the subscription belonged to that publisher, so deleting a non-member could close another publisher's stream. With duplicate registrations, removing one occurrence closed the shared stream while another observer still remained. `delete_observer` now closes only after removing a matching observer, and only when no matching observer remains in that publisher. Regression-guarded by `test_publisher_delete_non_member_does_not_close_subscription_stream` and `test_publisher_delete_duplicate_keeps_stream_open_until_last_observer`.
 
 ---
 

src/actor.rs

@@ -7,11 +7,11 @@
 //!
 //! # Shutdown behavior
 //!
-//! [`ActorAsync::stop`] sets the alive flag only. The consumer thread is **not**
-//! joined; commented-out [`join`](std::thread::JoinHandle::join) code shows
-//! join-based shutdown was deferred. A thread blocked in [`take`](crate::sync::BlockingQueue::take)
-//! is not interrupted. README/examples that use `thread::sleep` after `stop` are
-//! demo synchronization only—not a supported shutdown contract.
+//! [`ActorAsync::stop`] clears the actor alive flag and stops the mailbox queue
+//! from accepting later sends. The consumer thread is **not** joined;
+//! commented-out [`join`](std::thread::JoinHandle::join) code shows join-based
+//! shutdown was deferred. A thread already blocked in
+//! [`take`](crate::sync::BlockingQueue::take) is not interrupted.
 
 use std::collections::HashMap;
 use std::sync::{
@@ -176,7 +176,8 @@ where
         alive.load(Ordering::SeqCst)
     }
 
-    /// Clears alive flag only; does not join the mailbox thread (see module docs).
+    /// Clears the actor alive flag and stops the mailbox accepting later sends;
+    /// does not join the mailbox thread (see module docs).
     pub fn stop(&mut self) {
         {
             let started_alive = self.started_alive.lock().unwrap();
@@ -190,6 +191,9 @@ where
             }
             alive.store(false, Ordering::SeqCst);
         }
+
+        self.queue.stop();
+
         // NOTE: Kill thread <- OS depending
         // let mut join_handle = self.join_handle.lock().unwrap();
         // join_handle
@@ -591,6 +595,27 @@ fn test_actor_handle_id_and_no_parent() {
     assert_eq!(true, actor.get_handle_child("missing").is_none());
 }
 
+#[test]
+fn test_actor_handle_send_after_stop_is_rejected() {
+    let mut actor = ActorAsync::new(
+        |_this: &mut ActorAsync<i32, i32>, _msg: i32, _ctx: &mut HashMap<String, i32>| {},
+    );
+    let mut handle = actor.get_handle();
+
+    {
+        let started_alive = actor.started_alive.lock().unwrap();
+        let (started, alive) = &*started_alive;
+        started.store(true, Ordering::SeqCst);
+        alive.store(true, Ordering::SeqCst);
+    }
+
+    actor.stop();
+    handle.send(7);
+
+    let mut queue = handle.queue.clone();
+    assert_eq!(None, queue.poll());
+}
+
 #[test]
 fn test_actor_context_accumulates_state() {
     use super::common::LinkedListAsync;

src/common.rs

@@ -178,9 +178,9 @@ impl<T> LinkedListAsync<T> {
                 if alive.load(Ordering::SeqCst) {
                     self.inner.lock().unwrap().push_back(input);
                 }
-
-                self.wake();
             }
+
+            self.wake();
         }
 
         #[cfg(not(feature = "for_futures"))]
@@ -197,8 +197,8 @@ impl<T> LinkedListAsync<T> {
     #[cfg(feature = "for_futures")]
     #[inline]
     fn wake(&self) {
-        let mut waker = self.waker.lock().unwrap();
-        if let Some(waker) = waker.take() {
+        let waker = { self.waker.lock().unwrap().take() };
+        if let Some(waker) = waker {
             waker.wake();
         }
     }
@@ -214,9 +214,9 @@ impl<T> LinkedListAsync<T> {
         {
             let alive = self.alive.lock().unwrap();
             alive.store(false, Ordering::SeqCst);
-
-            self.wake()
         }
+
+        self.wake();
     }
 }
 

src/cor.rs

@@ -356,26 +356,24 @@ impl<RETURN: Send + Sync + 'static, RECEIVE: Send + Sync + 'static> Cor<RETURN,
         alive.load(Ordering::SeqCst)
     }
 
-    /// Cooperative stop: `yield_from` targeting this coroutine returns `None`.
+    /// Cooperative stop: clears the `alive` flag so a `yield_from` targeting this
+    /// coroutine returns `None` (its `receive` early-returns without queueing).
     ///
-    /// Does not join an async worker thread; shutdown redesign is deferred.
+    /// Does not join an async worker thread; shutdown redesign is deferred. The
+    /// op channel is not closed here: the shared `Sender` lives behind
+    /// `Arc<Mutex<..>>` and cannot be dropped from one handle, so cooperative
+    /// stop is delivered by the `alive` flag alone.
     pub fn stop(&mut self) {
-        {
-            let started_alive = self.started_alive.lock().unwrap();
-            let (started, alive) = &*started_alive;
-
-            if !started.load(Ordering::SeqCst) {
-                return;
-            }
-            if !alive.load(Ordering::SeqCst) {
-                return;
-            }
-            alive.store(false, Ordering::SeqCst);
+        let started_alive = self.started_alive.lock().unwrap();
+        let (started, alive) = &*started_alive;
 
-            {
-                drop(self.op_ch_sender.lock().unwrap());
-            }
+        if !started.load(Ordering::SeqCst) {
+            return;
+        }
+        if !alive.load(Ordering::SeqCst) {
+            return;
         }
+        alive.store(false, Ordering::SeqCst);
     }
 
     fn receive(

src/handler.rs

@@ -30,7 +30,7 @@ pub trait Handler: Send + Sync + 'static {
     /// `true` while the worker loop may still execute jobs.
     fn is_alive(&mut self) -> bool;
 
-    /// Starts (or resumes) the worker loop.
+    /// Starts the worker loop once; later calls are no-ops after `is_started` is true.
     fn start(&mut self);
 
     /// Stops accepting new jobs. Does not join the worker thread or interrupt a
@@ -197,6 +197,14 @@ impl Handler for HandlerThreadInner {
     }
 
     fn post(&mut self, func: RawFunc) {
+        // Preserve pre-start queueing, but once a handler has been started and
+        // then stopped it must reject new work. Otherwise post-after-stop can
+        // either leak into an un-drained queue or wake a worker blocked in
+        // take() and execute one more job after stop().
+        if self.started.load(Ordering::SeqCst) && !self.alive.load(Ordering::SeqCst) {
+            return;
+        }
+
         let q = Arc::make_mut(&mut self.q);
 
         q.put(func);
@@ -277,6 +285,27 @@ fn test_handler_stop_before_start_is_noop() {
     assert_eq!(false, h.is_alive());
 }
 
+#[test]
+fn test_handler_inner_post_after_stop_is_rejected() {
+    let mut inner = HandlerThreadInner::new();
+    inner.started.store(true, Ordering::SeqCst);
+    inner.alive.store(false, Ordering::SeqCst);
+
+    inner.post(RawFunc::new(|| panic!("stopped handler accepted work")));
+
+    let mut q = inner.q.as_ref().clone();
+    assert_eq!(true, q.poll().is_none());
+}
+
+#[test]
+fn test_handler_post_before_start_is_preserved() {
+    let mut inner = HandlerThreadInner::new();
+    inner.post(RawFunc::new(|| {}));
+
+    let mut q = inner.q.as_ref().clone();
+    assert_eq!(true, q.poll().is_some());
+}
+
 #[test]
 fn test_handler_post_executes_job() {
     use super::sync::CountDownLatch;

src/publisher.rs

@@ -157,18 +157,22 @@ impl<X: Send + Sync + 'static> Observable<X, SubscriptionFunc<X>> for Publisher<
         self.observers.push(observer);
     }
     fn delete_observer(&mut self, observer: Arc<SubscriptionFunc<X>>) {
-        let id;
-        {
-            id = observer.get_id();
-
-            #[cfg(feature = "for_futures")]
-            observer.as_ref().clone().close_stream();
-        }
+        let id = observer.get_id();
 
         for (index, obs) in self.observers.iter().enumerate() {
             if obs.get_id() == id {
                 // println!("delete_observer({});", observer);
+                #[cfg(feature = "for_futures")]
+                {
+                    let removed = self.observers.remove(index);
+                    if !self.observers.iter().any(|obs| obs.get_id() == id) {
+                        removed.as_ref().clone().close_stream();
+                    }
+                }
+
+                #[cfg(not(feature = "for_futures"))]
                 self.observers.remove(index);
+
                 return;
             }
         }
@@ -533,6 +537,43 @@ fn test_publisher_delete_observer_directly() {
     assert_eq!(1, count.load(Ordering::SeqCst));
 }
 
+#[cfg(feature = "for_futures")]
+#[test]
+fn test_publisher_delete_non_member_does_not_close_subscription_stream() {
+    use super::common::SubscriptionFunc;
+
+    let mut owner = Publisher::new();
+    let mut stranger = Publisher::new();
+    let subscription = Arc::new(SubscriptionFunc::new(|_x: Arc<i32>| {}));
+    let stream = owner.subscribe_as_stream(subscription.clone());
+
+    // Removing from a publisher that never owned the subscription must not
+    // close the stream attached to the real publisher.
+    stranger.delete_observer(subscription);
+
+    owner.publish(9);
+    assert_eq!(Some(Arc::new(9)), stream.pop_front());
+}
+
+#[cfg(feature = "for_futures")]
+#[test]
+fn test_publisher_delete_duplicate_keeps_stream_open_until_last_observer() {
+    use super::common::SubscriptionFunc;
+
+    let mut publisher = Publisher::new();
+    let subscription = Arc::new(SubscriptionFunc::new(|_x: Arc<i32>| {}));
+    let stream = publisher.subscribe_as_stream(subscription.clone());
+    publisher.subscribe(subscription.clone());
+
+    publisher.delete_observer(subscription.clone());
+    publisher.publish(1);
+    assert_eq!(Some(Arc::new(1)), stream.pop_front());
+
+    publisher.delete_observer(subscription);
+    publisher.publish(2);
+    assert_eq!(None, stream.pop_front());
+}
+
 #[test]
 fn test_publisher_default_has_no_observers() {
     use std::sync::Arc;