channel: drop queued messages when the Receiver drops

A queued Envelope::Call was stranded until the last Sender dropped, so a
caller parked in gen_server::call was never released with ServerDown when a
*named* server was request_stop'd — the registry's inbox Sender clone (lazy
prune) kept the channel Arc, and the queued reply_tx, alive indefinitely.

Receiver::Drop now drains the queue (items dropped after releasing the lock,
since a reply_tx drop reaches a different channel's lock + the scheduler),
restoring the documented ServerDown guarantee on every teardown path.

Adds tests/stop_with_queued_call.rs: deterministic pure-smarm reproducer.
This commit is contained in:
smarm-agent
2026-06-24 20:53:27 +00:00
parent 3e9c33377c
commit 6c2b7e91cf
2 changed files with 135 additions and 1 deletions
+15 -1
View File
@@ -132,7 +132,21 @@ impl<T> Drop for Sender<T> {
impl<T> Drop for Receiver<T> {
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<Inner> — 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);
}
}
+120
View File
@@ -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<Inner>` 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<Blocker> = 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<Receiver<()>>,
}
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<Mutex<Option<Result<(), CallError>>>> = 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::<Result<(), CallError>>();
// 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)"
);
}