diff --git a/src/client/lifecycle.rs b/src/client/lifecycle.rs index 0a8c9fad6..f73075e53 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,11 +1171,16 @@ 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() { - transport.disconnect().await; - } + // 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; + } // 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 @@ -1286,8 +1297,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 +1598,157 @@ 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 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] + 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(); } }