From e7d3650bff229e88c555d3beedd0e40d3300f1c5 Mon Sep 17 00:00:00 2001 From: Johnathan Falk Date: Wed, 12 Aug 2026 11:07:52 -0400 Subject: [PATCH] fix(agent): wake the shutdown waiter that notify_waiters left behind `Shutdown::stop` signalled with `Notify::notify_waiters`, which wakes only the tasks already parked at that instant and stores no permit. Both waits -- the session loop and the reconnect backoff -- awaited `notified()` directly, with no level-triggered read of the flag `stop` had already set. When `stop` landed while the client was anywhere else (mid-recovery, mid-register, mid-dispatch) the signal was lost. The session loop then parked on a notification already gone, the inbound stream stayed open and silent, `session` never returned, and `run` never reached its `is_stopped` check. That is a permanent hang, not a slow one: 26 hours parked on a condvar at zero CPU on CI, ~1 run in 50 locally. It presented as a hung `cargo test --workspace` and was twice mistaken for a stuck runner, because a hang leaves no failing test behind to point at. `Shutdown::cancelled` registers as a waiter before reading the flag, and `stop` writes the flag before notifying. Two opposed orderings, so no interleaving misses both. The regression test asserts under a timeout on purpose -- the bug hangs rather than fails, so a bare assertion would reproduce it instead of reporting it. Verified by sabotaging `cancelled` back to the naive form and confirming the test goes red. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01U9keBRnGXBP2DiQmrcqm33 --- changelog.d/20260812-shutdown-lost-wakeup.md | 34 ++++++++ crates/transcodarr-agent/src/client.rs | 83 +++++++++++++++++++- 2 files changed, 113 insertions(+), 4 deletions(-) create mode 100644 changelog.d/20260812-shutdown-lost-wakeup.md diff --git a/changelog.d/20260812-shutdown-lost-wakeup.md b/changelog.d/20260812-shutdown-lost-wakeup.md new file mode 100644 index 0000000..7977367 --- /dev/null +++ b/changelog.d/20260812-shutdown-lost-wakeup.md @@ -0,0 +1,34 @@ + + + + + +### Fixed + +#### The agent could park forever instead of shutting down + +`Shutdown::stop` signalled waiters with `Notify::notify_waiters`, which wakes +only the tasks already parked at the instant it runs and leaves no permit for +one that arrives later. Both places that waited on it — the session loop and the +reconnect backoff — awaited `notified()` directly, with no level-triggered check +of the flag that `stop` had already set. + +So whenever `stop` landed while the client was somewhere other than parked on +that await — mid-recovery, mid-register, mid-dispatch — the signal was lost. The +session loop then parked on a notification that had already been and gone, while +the inbound stream stayed open and silent. `session` never returned, so `run` +never reached the `is_stopped` check that would have ended it. + +This was not a slow shutdown. It was a permanent one: a CI job sat parked on a +condvar at zero CPU for 26 hours, and the local suite reproduced it in roughly +1 run in 50. It presented as a hung `cargo test --workspace` and was twice +mistaken for a stuck runner, because a hang leaves no failing test behind to +point at. + +`Shutdown::cancelled` replaces both waits. It registers as a waiter *before* it +reads the flag, and `stop` writes the flag *before* it notifies — two opposed +orderings, so no interleaving can miss both. + +The regression test asserts under a `timeout`, deliberately: the bug hangs +rather than fails, and an assertion alone would reproduce the disease instead of +reporting it. diff --git a/crates/transcodarr-agent/src/client.rs b/crates/transcodarr-agent/src/client.rs index 2655365..77f55c7 100644 --- a/crates/transcodarr-agent/src/client.rs +++ b/crates/transcodarr-agent/src/client.rs @@ -1,7 +1,7 @@ // file: crates/transcodarr-agent/src/client.rs -// version: 1.0.0 +// version: 1.1.0 // guid: 0f5d8c31-97b4-42ae-b6d0-58e19c3a7042 -// last-edited: 2026-08-05 +// last-edited: 2026-08-12 //! The agent side of the transport: register, connect, stay connected. //! //! The server half of this lives in `transcodarr-server::session`, and the @@ -426,6 +426,35 @@ impl Shutdown { fn is_stopped(&self) -> bool { self.stopped.load(Ordering::SeqCst) } + + /// Resolves once stopped — **including when it was already stopped before + /// this was called**. + /// + /// Awaiting `notify.notified()` directly is the wrong thing and the reason + /// this method exists. `notify_waiters` wakes only the tasks already parked + /// at the instant it runs and stores no permit, so the signal is lost + /// whenever [`Shutdown::stop`] lands while this task is somewhere else — + /// mid-recovery, mid-register, mid-dispatch. The task then parks on a + /// notification that has already been and gone. That is not a slow + /// shutdown, it is a permanent one: it cost a CI job 26 hours parked on a + /// condvar at zero CPU. + /// + /// The ordering argument: `stop` stores the flag *before* it notifies, and + /// this registers as a waiter *before* it reads the flag. So either the + /// read below sees `true`, or the notify that follows it is guaranteed to + /// reach an already-registered waiter. There is no interleaving in which + /// both are missed. + async fn cancelled(&self) { + let notified = self.notify.notified(); + tokio::pin!(notified); + // Registers interest without awaiting. This line is the fix; moving it + // below the check reopens the race. + notified.as_mut().enable(); + if self.is_stopped() { + return; + } + notified.await; + } } /// Registers, connects, and keeps connected. @@ -497,7 +526,7 @@ impl ConnectClient { tracing::info!(?delay, attempt, "reconnecting"); tokio::select! { _ = tokio::time::sleep(delay) => {} - _ = self.shutdown.notify.notified() => break, + _ = self.shutdown.cancelled() => break, } } tracing::info!("agent client stopped"); @@ -570,7 +599,7 @@ impl ConnectClient { loop { tokio::select! { - _ = self.shutdown.notify.notified() => return Ok(()), + _ = self.shutdown.cancelled() => return Ok(()), message = inbound.message() => { match message { Ok(Some(msg)) => self.dispatch(msg, &link), @@ -814,4 +843,50 @@ mod tests { assert!(p.delay(u32::MAX, "boot-a") <= p.max); assert!(p.delay(u32::MAX, "boot-a") > Duration::ZERO); } + + fn test_shutdown() -> Shutdown { + Shutdown { + stopped: Arc::new(AtomicBool::new(false)), + notify: Arc::new(Notify::new()), + } + } + + /// The lost wakeup, which is the failure that hung a CI job for 26 hours. + /// + /// `stop` runs while nothing is parked yet — exactly what happens when it + /// lands mid-recovery, mid-register or mid-dispatch. `notify_waiters` + /// reaches nobody and leaves no permit behind, so a waiter arriving + /// afterwards has only the flag to go on. + /// + /// Note the timeout. The bug this guards against *hangs* rather than + /// fails, so an assertion alone would reproduce the disease instead of + /// reporting it: without the wrapper, a regression parks this test forever + /// and the suite dies with no failing test to point at. + #[tokio::test] + async fn a_stop_that_precedes_the_waiter_is_still_seen() { + let shutdown = test_shutdown(); + shutdown.stop(); + tokio::time::timeout(Duration::from_secs(5), shutdown.cancelled()) + .await + .expect("cancelled() must resolve when stop() has already happened"); + } + + /// The other direction, kept because the obvious fix breaks it: a + /// `cancelled` that only polled the flag once and never registered would + /// satisfy the test above and leave an already-parked waiter asleep. + #[tokio::test] + async fn a_waiter_already_parked_is_still_woken() { + let shutdown = test_shutdown(); + let waiter = shutdown.clone(); + let parked = tokio::spawn(async move { waiter.cancelled().await }); + // Single-threaded runtime: this hands control to the spawned task and + // does not come back until it has parked. + tokio::task::yield_now().await; + + shutdown.stop(); + tokio::time::timeout(Duration::from_secs(5), parked) + .await + .expect("a waiter parked before stop() must be woken by it") + .expect("the waiting task must not panic"); + } }