fix(client): release transport and feed guards before awaiting

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<AtomicUsize>,
+    }
+
+    #[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<ParkedDisconnect>,
+        async_channel::Receiver<()>,
+        async_channel::Sender<()>,
+        Arc<AtomicUsize>,
+    ) {
+        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());
+    }
 }

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<std::sync::OnceLock<Box<dyn Fn() + Send + Sync>>>);
+
+    struct ReentrantAbortRuntime {
+        handle: tokio::runtime::Handle,
+        hook: AbortHook,
+    }
+
+    impl wacore::runtime::Runtime for ReentrantAbortRuntime {
+        fn spawn(
+            &self,
+            future: std::pin::Pin<Box<dyn Future<Output = ()> + 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<Box<dyn Future<Output = ()> + Send + 'static>>,
+        ) {
+            self.handle.spawn(future);
+        }
+
+        fn sleep(&self, duration: Duration) -> std::pin::Pin<Box<dyn Future<Output = ()> + Send>> {
+            Box::pin(tokio::time::sleep(duration))
+        }
+
+        fn spawn_blocking(
+            &self,
+            f: Box<dyn FnOnce() + Send + 'static>,
+        ) -> std::pin::Pin<Box<dyn Future<Output = ()> + Send>> {
+            Box::pin(async {
+                let _ = tokio::task::spawn_blocking(f).await;
+            })
+        }
+
+        fn yield_now(&self) -> Option<std::pin::Pin<Box<dyn Future<Output = ()> + 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<dyn VideoSource> = Arc::new(src_rx);
+        let sink: Arc<dyn VideoSink> = 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]

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();
         }
     }