diff --git a/src/channel.rs b/src/channel.rs index f14fc34..d0aa0f6 100644 --- a/src/channel.rs +++ b/src/channel.rs @@ -132,7 +132,21 @@ impl Drop for Sender { impl Drop for Receiver { fn drop(&mut self) { - self.inner.lock().receiver_alive = false; + // The only consumer is gone: queued messages can never be delivered. + // Drop them now instead of stranding them until the last Sender goes + // away (a registry entry under lazy prune can keep a Sender — and thus + // the Arc — alive long after the server exits). Dropping a queued + // Envelope::Call drops its reply_tx, waking any caller parked in `call` + // with ServerDown, so the documented guarantee holds on *every* teardown + // path, not only the all-senders-drop one. Drain under the lock, then + // run item destructors after releasing it (a reply_tx drop reaches into + // a *different* channel's lock + the scheduler, so it must not nest). + let drained = { + let mut g = self.inner.lock(); + g.receiver_alive = false; + std::mem::take(&mut g.queue) + }; + drop(drained); } } diff --git a/tests/stop_with_queued_call.rs b/tests/stop_with_queued_call.rs new file mode 100644 index 0000000..d48d068 --- /dev/null +++ b/tests/stop_with_queued_call.rs @@ -0,0 +1,120 @@ +//! Reproducer: a *named* gen_server stopped with `request_stop` while a `call` +//! sits **un-dequeued** in its inbox does NOT release the parked caller with +//! `CallError::ServerDown`. The caller parks forever, contradicting the +//! documented gen_server guarantee ("Any caller currently waiting in `call` +//! sees `Err(ServerDown)`"). +//! +//! Root cause (channel.rs): `Receiver::Drop` only flips `receiver_alive = false` +//! and never drains `queue`. The queued `Envelope::Call(_, reply_tx)` therefore +//! survives as long as the channel `Arc` does — and for a *named* server +//! the registry holds a `Sender` clone (lazy prune) that keeps the `Arc` alive +//! after the server is gone. So the queued `reply_tx` is never dropped, the +//! caller's `reply_rx` never closes, and `reply_rx.recv()` parks forever. +//! +//! Anonymous servers happen to dodge this: when their last `GenServerRef` +//! drops, every `Sender` drops, the `Arc` refcount hits zero, `Inner` (and its +//! queue) is dropped, and the queued `reply_tx` goes with it — waking the +//! caller. The bug is specific to "a `Sender` outlives the `Receiver`", which a +//! registry entry guarantees for every named server. + +use smarm::{ + call, channel, init, request_stop, spawn, Config, GenServer, GenServerBuilder, GenServerName, + CallError, Receiver, RecvTimeoutError, +}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +const BLOCKER: GenServerName = GenServerName::new("repro_blocker"); + +/// A server that, on its single cast, parks forever on a gate channel the test +/// never feeds. This deterministically holds the server *inside a handler* (not +/// at the inbox recv), so any subsequent `call` queues behind it and stays +/// un-dequeued — exactly the state `request_stop` then has to clean up. +struct Blocker { + gate: Option>, +} + +impl GenServer for Blocker { + type Call = (); + type Reply = (); + type Cast = (); + type Info = (); + type Timer = (); + + // Trivial + instant: if this ever ran for the queued call, the caller would + // get Ok(()) immediately. It must NOT run — the server is parked on the gate + // when the stop arrives. + fn handle_call(&mut self, _req: ()) {} + + // Park forever (until cancelled). recv() on an open channel with no message + // parks the actor; the gate sender is held by the test and never fires. + fn handle_cast(&mut self, _op: ()) { + if let Some(gate) = self.gate.take() { + let _ = gate.recv(); + } + } +} + +#[test] +fn named_server_request_stop_releases_queued_caller_with_server_down() { + // Final observation, asserted after the run. + // Some(Err(ServerDown)) -> contract honored (fixed) + // None -> caller never released; parked past the 3s + // bound (bug reproduced) + let outcome: Arc>>> = Arc::new(Mutex::new(None)); + let outcome_w = outcome.clone(); + + init(Config::exact(2)).run(move || { + // Gate the server will park on. Held for the whole run so the server's + // gate.recv() parks (rather than seeing Disconnected and returning). + let (gate_tx, gate_rx) = channel::<()>(); + + // Channel the queued caller reports its result back on. + let (res_tx, res_rx) = channel::>(); + + // 1. Start the named server and keep its ref alive. + let server = GenServerBuilder::new(Blocker { gate: Some(gate_rx) }) + .named(BLOCKER) + .start() + .expect("name should be free"); + let spid = server.pid(); + + // 2. Send the cast and let the server dequeue it and park on the gate. + server.cast(()).expect("server is live"); + smarm::sleep(Duration::from_millis(100)); + + // 3. A separate caller issues a by-name `call`. The server is parked on + // the gate, so this Call envelope queues un-dequeued; the caller then + // parks on its reply channel. + spawn(move || { + let r = call(BLOCKER, ()); + let _ = res_tx.send(r); + }); + smarm::sleep(Duration::from_millis(100)); + + // 4. Stop the server. Its loop unwinds out of the gate.recv() and drops + // the inbox Receiver — at which point the queued caller is *supposed* + // to be released with ServerDown. + request_stop(spid); + + // 5. Bounded wait. A correct runtime releases the caller in well under + // 3s; the bug leaves it parked, so we time out. + let observed = match res_rx.recv_timeout(Duration::from_secs(3)) { + Ok(r) => Some(r), + Err(RecvTimeoutError::Timeout) => None, + Err(RecvTimeoutError::Disconnected) => None, + }; + *outcome_w.lock().unwrap() = observed; + + // Keep the gate sender alive until the very end. + drop(gate_tx); + }); + + let observed = outcome.lock().unwrap().take(); + assert_eq!( + observed, + Some(Err(CallError::ServerDown)), + "queued caller was not released with ServerDown after the named server \ + was request_stop'd (None = parked forever => bug reproduced)" + ); +}