From 03770574f8e6ca63b957c0d69a00b3c1fe1ef448 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 06:50:58 +0000 Subject: [PATCH 1/2] fix(client): release transport and feed guards before awaiting Edition 2024 shortened the life of an `if let` scrutinee temporary only for the `else` branch. In the arm that matches, the temporary is still alive for the whole body, because the binding may borrow from it. So in `if let Some(x) = mutex.lock()....take() { x.something().await }` the guard is held for the entire body, and the same is true of a `match` scrutinee. That bites four teardown paths in the client: `disconnect`, `reconnect`, `reconnect_immediately` and `cleanup_connection_state` all closed the socket with the `transport` guard still held. `WsTransport::disconnect` writes a real TLS close frame with no timeout, and first waits on the sink mutex the sender task may be holding, while `connect_internal` installs the next connection's transport through that same `transport` mutex. A close that stalls therefore parks the whole reconnect. Hoisting the value out of the guard mirrors what the stream-error path in `node_io` already does. `cleanup_connection_state` had one more of the same shape: the `app_state_processor` guard was held across `clear_key_cache().await`, which takes the processor's own cache lock, so every `get_app_state_processor()` waited behind it. In the video facade, `attach_endpoints` and `detach_endpoints` aborted the previous feed under the `feed` mutex. `AbortHandle::abort` runs a closure the `Runtime` supplied; a runtime that cancels synchronously drops the task inline, and anything in that drop touching `feed` re-enters a non-reentrant `std::sync::Mutex`. Not reachable with the current tokio runtime, whose abort is just a flag, but the trait admits runtimes where it is. `AbortHandle::abort` itself called the closure under its own `abort_fn` lock for the same reason and now takes it out first. Both fixes come with a regression test that fails on the previous code: one proves the transport slot stays installable while a close is in flight, one drives attach/replace/detach against a runtime whose abort re-enters the feed lock. --- src/client/lifecycle.rs | 134 ++++++++++++++++++++++++++++++++++++++-- src/voip/facade.rs | 120 +++++++++++++++++++++++++++++++++-- wacore/src/runtime.rs | 6 +- 3 files changed, 247 insertions(+), 13 deletions(-) diff --git a/src/client/lifecycle.rs b/src/client/lifecycle.rs index 0a8c9fad6..25fbc2daf 100644 --- a/src/client/lifecycle.rs +++ b/src/client/lifecycle.rs @@ -937,7 +937,11 @@ impl Client { } // Close after flush; cleanup may also win this race on the run loop. - if let Some(transport) = self.transport.lock().await.as_ref() { + // Guard dropped first: edition 2024 keeps an `if let` scrutinee temporary alive + // for the whole matching arm, so holding it across `disconnect()` (an untimed + // socket write) would park `connect_internal`, which installs through this mutex. + let transport = self.transport.lock().await.clone(); + if let Some(transport) = transport { transport.disconnect().await; } self.cleanup_connection_state().await; @@ -1003,7 +1007,8 @@ impl Client { .await; self.notify_connection_shutdown(); - if let Some(transport) = self.transport.lock().await.as_ref() { + let transport = self.transport.lock().await.clone(); + if let Some(transport) = transport { transport.disconnect().await; } } @@ -1038,7 +1043,8 @@ impl Client { .await; self.notify_connection_shutdown(); - if let Some(transport) = self.transport.lock().await.as_ref() { + let transport = self.transport.lock().await.clone(); + if let Some(transport) = transport { transport.disconnect().await; } } @@ -1165,7 +1171,8 @@ impl Client { // `Client::disconnect()`). Transport impls make `disconnect()` // idempotent, so the redundant call from `Client::disconnect()` is // safe. - if let Some(transport) = self.transport.lock().await.take() { + let transport = self.transport.lock().await.take(); + if let Some(transport) = transport { transport.disconnect().await; } *self.transport_events.lock().await = None; @@ -1286,8 +1293,9 @@ impl Client { *self.media_conn.write().await = None; // Clear app state key cache — keys will be re-fetched from DB on demand - if let Some(proc) = self.app_state_processor.lock().await.as_ref() { - proc.clear_key_cache().await; + let processor = self.app_state_processor.lock().await.clone(); + if let Some(processor) = processor { + processor.clear_key_cache().await; } #[cfg(feature = "client-lifecycle")] drop(scope_close); @@ -1586,4 +1594,118 @@ mod tests { .expect("run() must still return on a terminal shutdown") .expect("the run task must not panic"); } + + /// A transport whose `disconnect()` parks until released, standing in for the real + /// close-frame write: a TLS write with no timeout, behind the sender's own mutex. + struct ParkedDisconnect { + entered: async_channel::Sender<()>, + release: async_channel::Receiver<()>, + closes: Arc, + } + + #[async_trait::async_trait] + impl crate::transport::Transport for ParkedDisconnect { + async fn send(&self, _data: bytes::Bytes) -> Result<()> { + Ok(()) + } + + async fn disconnect(&self) { + self.closes.fetch_add(1, Ordering::SeqCst); + let _ = self.entered.send(()).await; + let _ = self.release.recv().await; + } + } + + fn parked_transport() -> ( + Arc, + async_channel::Receiver<()>, + async_channel::Sender<()>, + Arc, + ) { + let (entered_tx, entered_rx) = async_channel::bounded(4); + let (release_tx, release_rx) = async_channel::bounded(1); + let closes = Arc::new(AtomicUsize::new(0)); + let transport = Arc::new(ParkedDisconnect { + entered: entered_tx, + release: release_rx, + closes: closes.clone(), + }); + (transport, entered_rx, release_tx, closes) + } + + /// The teardown paths must not hold the `transport` mutex across the socket close: + /// `connect_internal` installs the next connection's transport through that same mutex, + /// so a close that never returns would take the whole reconnect down with it. + #[tokio::test] + async fn an_in_flight_socket_close_does_not_park_the_transport_slot() { + let client = crate::test_utils::create_test_client().await; + let (transport, entered_rx, release_tx, _closes) = parked_transport(); + *client.transport.lock().await = Some(transport); + + let cleanup = tokio::spawn({ + let client = Arc::clone(&client); + async move { client.cleanup_connection_state().await } + }); + tokio::time::timeout(Duration::from_secs(5), entered_rx.recv()) + .await + .expect("cleanup must reach the socket close") + .expect("the observer channel must stay open"); + + // Exactly what `connect_internal` does once the next socket is up. + tokio::time::timeout(Duration::from_secs(5), async { + *client.transport.lock().await = Some(Arc::new(crate::transport::mock::MockTransport)); + }) + .await + .expect("installing the next transport must not wait on the in-flight close"); + + drop(release_tx); + tokio::time::timeout(Duration::from_secs(5), cleanup) + .await + .expect("cleanup must finish once the close returns") + .expect("cleanup must not panic"); + } + + /// The happy path the fix must preserve: cleanup still closes the socket it owns and + /// still leaves the slot empty for the next connection. + #[tokio::test] + async fn cleanup_closes_the_transport_and_clears_the_slot() { + let client = crate::test_utils::create_test_client().await; + let (transport, _entered_rx, release_tx, closes) = parked_transport(); + drop(release_tx); // never park: this is the ordinary, prompt close + *client.transport.lock().await = Some(transport); + + tokio::time::timeout(Duration::from_secs(5), client.cleanup_connection_state()) + .await + .expect("cleanup must not block"); + + assert_eq!( + closes.load(Ordering::SeqCst), + 1, + "the socket must be closed" + ); + assert!( + client.transport.lock().await.is_none(), + "cleanup owns the teardown and must leave the slot free" + ); + } + + /// Same for the user-facing path: `disconnect()` closes the socket and hands a cleared + /// slot back, so nothing from the dead connection survives into the next one. + #[tokio::test] + async fn disconnect_closes_the_transport_and_clears_the_slot() { + let client = crate::test_utils::create_test_client().await; + let (transport, _entered_rx, release_tx, closes) = parked_transport(); + drop(release_tx); + *client.transport.lock().await = Some(transport); + + tokio::time::timeout(Duration::from_secs(10), client.disconnect()) + .await + .expect("disconnect must not block"); + + assert!( + closes.load(Ordering::SeqCst) >= 1, + "the socket must be closed" + ); + assert!(client.transport.lock().await.is_none()); + } } diff --git a/src/voip/facade.rs b/src/voip/facade.rs index 196cefa6d..c688b64f8 100644 --- a/src/voip/facade.rs +++ b/src/voip/facade.rs @@ -2851,12 +2851,15 @@ impl VideoShared { ended, }; let handle = client.runtime.spawn(Box::pin(feed.run())); - if let Some(old) = self + // Abort outside the guard: edition 2024 keeps an `if let` scrutinee temporary + // alive for the whole matching arm, and a `Runtime` that cancels synchronously + // would drop the task inline and re-enter this non-reentrant mutex. + let old = self .feed .lock() .unwrap_or_else(|e| e.into_inner()) - .replace(handle) - { + .replace(handle); + if let Some(old) = old { old.abort(); } } @@ -2865,7 +2868,8 @@ impl VideoShared { /// dropped, and the drive loop's video plane is disabled so it stops emitting/decoding video. fn detach_endpoints(&self) { *self.sink_slot.lock().unwrap_or_else(|e| e.into_inner()) = None; - if let Some(feed) = self.feed.lock().unwrap_or_else(|e| e.into_inner()).take() { + let feed = self.feed.lock().unwrap_or_else(|e| e.into_inner()).take(); + if let Some(feed) = feed { feed.abort(); } self.send_control(VideoControl::Disable); @@ -8676,6 +8680,114 @@ mod tests { ); } + /// Runs a caller-supplied hook inline on `abort()`, standing in for a `Runtime` that + /// finishes cancellation synchronously and therefore drops the task on the spot. + #[derive(Clone, Default)] + struct AbortHook(Arc>>); + + struct ReentrantAbortRuntime { + handle: tokio::runtime::Handle, + hook: AbortHook, + } + + impl wacore::runtime::Runtime for ReentrantAbortRuntime { + fn spawn( + &self, + future: std::pin::Pin + Send + 'static>>, + ) -> wacore::runtime::AbortHandle { + let task = self.handle.spawn(future); + let hook = self.hook.clone(); + wacore::runtime::AbortHandle::new(move || { + task.abort(); + if let Some(reenter) = hook.0.get() { + reenter(); + } + }) + } + + fn spawn_detached( + &self, + future: std::pin::Pin + Send + 'static>>, + ) { + self.handle.spawn(future); + } + + fn sleep(&self, duration: Duration) -> std::pin::Pin + Send>> { + Box::pin(tokio::time::sleep(duration)) + } + + fn spawn_blocking( + &self, + f: Box, + ) -> std::pin::Pin + Send>> { + Box::pin(async { + let _ = tokio::task::spawn_blocking(f).await; + }) + } + + fn yield_now(&self) -> Option + Send>>> { + None + } + } + + /// The old feed must be aborted with the `feed` mutex already released: cancellation runs + /// runtime-supplied code, and on a runtime that cancels synchronously that code runs the + /// task's drop inline, re-entering this non-reentrant mutex. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn feed_swap_does_not_abort_while_holding_the_feed_lock() { + let hook = AbortHook::default(); + let persistence_manager = Arc::new( + PersistenceManager::new(create_test_backend().await) + .await + .expect("persistence manager"), + ); + let client = Client::builder() + .with_runtime_arc(Arc::new(ReentrantAbortRuntime { + handle: tokio::runtime::Handle::current(), + hook: hook.clone(), + })) + .with_persistence_manager(persistence_manager) + .with_transport_factory(crate::transport::mock::MockTransportFactory::new()) + .with_http_client(MockHttpClient) + .build() + .await + .expect("client build") + .into_client(); + + let shared = Arc::new(VideoShared::new()); + let _receivers = shared.take_receivers(); + hook.0 + .set({ + let shared = Arc::clone(&shared); + Box::new(move || { + let _guard = shared.feed.lock().unwrap_or_else(|e| e.into_inner()); + }) + }) + .ok() + .expect("the hook is installed once"); + + let (src_rx, vout_tx) = video_endpoints(); + let source: Arc = Arc::new(src_rx); + let sink: Arc = Arc::new(vout_tx); + let ended = Arc::new(EndedFlag::default()); + + // A re-entrant lock parks its thread for good, so drive the sequence on a thread of its + // own and let the timeout report the deadlock instead of hanging the suite. + let (done_tx, done_rx) = std::sync::mpsc::sync_channel(1); + std::thread::spawn(move || { + shared.attach_endpoints(&client, &source, &sink, ended.clone()); + // Replacing an existing feed aborts it... + shared.attach_endpoints(&client, &source, &sink, ended.clone()); + // ...and so does releasing the endpoints. + shared.detach_endpoints(); + let _ = done_tx.send(()); + }); + + done_rx + .recv_timeout(Duration::from_secs(5)) + .expect("attach/detach must not hold the feed lock while aborting the old feed"); + } + // A second take is defensive: it must yield closed channels (their driver arms self-disable), // never panic or hand out a duplicate of the live receivers. #[test] diff --git a/wacore/src/runtime.rs b/wacore/src/runtime.rs index 929b468cc..05b26e33f 100644 --- a/wacore/src/runtime.rs +++ b/wacore/src/runtime.rs @@ -124,12 +124,12 @@ impl AbortHandle { /// Explicitly abort the spawned task without waiting for drop. pub fn abort(&self) { - if let Some(f) = self + let abort_fn = self .abort_fn .lock() .unwrap_or_else(|e| e.into_inner()) - .take() - { + .take(); + if let Some(f) = abort_fn { f(); } } From 13ddd6f333532306449a2e6d757d64de6f6d3a48 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 07:14:35 +0000 Subject: [PATCH 2/2] fix(client): clear connection slots before awaiting the socket close Releasing the transport guard removed an accidental barrier: while cleanup held it across the close, a concurrent `connect()` could not finish publishing its transport, event receiver and Noise socket, because the first of those three installs blocked on the same mutex. Without the guard it can publish all three while the old close is still in flight, and cleanup then cleared the replacement's `transport_events` and `noise_socket` instead of the ones it was tearing down, leaving the new connection with no read loop. Clear all three slots in the same pass that takes the transport, before the close is awaited, so cleanup only ever retires state that existed when it started. Ordering between shutdown, disconnect and cleanup is unchanged; this moves two lines within cleanup itself. --- src/client/lifecycle.rs | 47 +++++++++++++++++++++++++++++++++++++++-- 1 file changed, 45 insertions(+), 2 deletions(-) diff --git a/src/client/lifecycle.rs b/src/client/lifecycle.rs index 25fbc2daf..f73075e53 100644 --- a/src/client/lifecycle.rs +++ b/src/client/lifecycle.rs @@ -1171,12 +1171,16 @@ impl Client { // `Client::disconnect()`). Transport impls make `disconnect()` // idempotent, so the redundant call from `Client::disconnect()` is // safe. + // All three slots are cleared before the close is awaited, not after. The guard on + // `transport` no longer spans that await, so a `connect()` racing this teardown can + // publish its own transport, events and socket while the close is in flight; clearing + // afterwards would strip the replacement connection instead of the one being torn down. let transport = self.transport.lock().await.take(); + *self.transport_events.lock().await = None; + *self.noise_socket.lock().await = None; if let Some(transport) = transport { transport.disconnect().await; } - *self.transport_events.lock().await = None; - *self.noise_socket.lock().await = None; // Authoritative point for the gauge: every disconnect (intentional or a // run-loop drop/reconnect) funnels through here, so disconnect()'s early // set is just a prompt redundant signal. (`is_connected` was already cleared above, before @@ -1665,6 +1669,45 @@ mod tests { .expect("cleanup must not panic"); } + /// The other half of releasing the guard: now that a `connect()` can publish its own + /// connection while the old close is still in flight, cleanup must not come back and + /// strip the replacement's state. Everything cleanup clears, it clears before the close. + #[tokio::test] + async fn a_connection_published_during_cleanup_is_not_stripped_by_it() { + let client = crate::test_utils::create_test_client().await; + let (transport, entered_rx, release_tx, _closes) = parked_transport(); + *client.transport.lock().await = Some(transport); + *client.transport_events.lock().await = Some(async_channel::bounded(1).1); + + let cleanup = tokio::spawn({ + let client = Arc::clone(&client); + async move { client.cleanup_connection_state().await } + }); + tokio::time::timeout(Duration::from_secs(5), entered_rx.recv()) + .await + .expect("cleanup must reach the socket close") + .expect("the observer channel must stay open"); + + // The publish half of `connect_internal`, minus the handshake. + *client.transport.lock().await = Some(Arc::new(crate::transport::mock::MockTransport)); + *client.transport_events.lock().await = Some(async_channel::bounded(1).1); + + drop(release_tx); + tokio::time::timeout(Duration::from_secs(5), cleanup) + .await + .expect("cleanup must finish once the close returns") + .expect("cleanup must not panic"); + + assert!( + client.transport.lock().await.is_some(), + "the replacement transport must survive the teardown it did not belong to" + ); + assert!( + client.transport_events.lock().await.is_some(), + "the replacement's event receiver must survive too, or its read loop never starts" + ); + } + /// The happy path the fix must preserve: cleanup still closes the socket it owns and /// still leaves the slot empty for the next connection. #[tokio::test]