`Room::close()` does not return when the signal link stops delivering data

[gasmith]

Summary

(Disclaimer: This issue report was prepared with AI.)

A signal WebSocket can stop delivering data while the socket stays open. There is no FIN, no RST, and no ICMP error. This is the usual behavior of a lost cellular or Wi-Fi uplink.

In this condition Room::close() does not return. The reconnect loop stops at the same time. The client does not recover, and the client does not leave the room.

We saw this in production. A device stayed in a room for hours after our SDK reported a disconnect. We had a 5s timeout around Room::close() and dropped the Room when the timer expired, but the room stayed active.

Verified on main (90a9e45b).

Root cause

SignalStream::close waits for read_handle (livekit-signaling/src/signal_stream.rs:86):

pub async fn close(self, notify_close: bool) {
    if notify_close {
        let _ = self.internal_tx.send(InternalMessage::Close).await;
    }
    let _ = self.write_handle.await;
    let _ = self.read_handle.await;   // does not return
}

read_task waits in conn.recv(). It exits only on Ok(None) or Err(..) (signal_stream.rs:139). Nothing else can stop it.

write_task calls conn.close() before it exits, but this does not stop read_task. NativeConnection::close locks only the writer and closes the SplitSink (livekit-net/src/native/connection.rs:83). The reader is a separate SplitStream that continues to poll the socket. On a link that delivers no data, recv() never completes.

SignalStream has no Drop implementation. A dropped SignalStream detaches its tasks, so the tasks continue to run.

Every caller that awaits a SignalStream::close therefore stops permanently.

Effect

On a ping timeout, signal_task exits its loop and calls inner.close(true) (lib.rs:756). That call holds the stream write lock for the full shutdown, because the workspace uses edition 2021 (Cargo.toml:47) and the RwLockWriteGuard lives to the end of the if let block (lib.rs:617). The lock is therefore never released. Then:

• SignalInner::restart waits on self.stream.write().await, so the reconnect never reaches SignalStream::connect. The engine stays in Reconnecting and logs no connect attempt. restart is also affected on its own path: it calls old_stream.close(false).await while it holds its own write guard (lib.rs:562-564).
• SessionInner::close sends Leave, which is a pass-through signal (lib.rs:772) and therefore waits on self.stream.read().await. Room::close() does not return from this await.

#1335 examined these same two awaits and called them unbounded (livekit/src/rtc_engine/rtc_session.rs:1997). The mechanism it identified is a different one, and it is finite: conn.send() on a half-open socket blocks the write task until TCP gives up, and the queued InternalMessage::Close waits behind it in a capacity-8 channel.

The read path has no such limit. A socket that receives no data never triggers a retransmission timeout, so nothing in the kernel or in this crate ever wakes read_task. SignalStream::close joins read_handle as well as write_handle, so it does not return at all. A caller timeout is therefore not a sufficient workaround, and #1335 already treats a caller timeout as the expected response to this API.

The CLOSE_DRAIN_TIMEOUT in the fix below also bounds the channel backpressure case that #1335 describes.

Two further points make the state unrecoverable:

• Room has no Drop implementation. A caller that abandons close() cannot drop the room instead. The engine tasks hold their own Arcs, and the socket stays open.
• RoomSession::close takes its task handles on entry (livekit/src/room/mod.rs:1168). A cancelled close() future cannot be retried. The second call returns AlreadyClosed and shuts down nothing.

Reproduction

Add this to livekit-signaling/src/signal_stream.rs. It needs no server. It builds a SignalStream the same way connect does, over a transport that never delivers data and whose close affects only the writer.

#[cfg(test)]
mod blackhole_tests {
    use super::*;
    use livekit_net::TransportError;
    use std::sync::atomic::{AtomicBool, Ordering};

    /// A transport that delivers no data and never closes the read side.
    /// `close` only records the call, as `NativeConnection::close` closes only
    /// the writer.
    struct StalledConn {
        closed: AtomicBool,
    }

    #[async_trait::async_trait]
    impl livekit_net::WsConnection for StalledConn {
        async fn send(&self, _frame: Vec<u8>) -> Result<(), TransportError> {
            Ok(())
        }
        async fn recv(&self) -> Result<Option<Vec<u8>>, TransportError> {
            std::future::pending().await
        }
        async fn close(&self) {
            self.closed.store(true, Ordering::SeqCst);
        }
    }

    fn stalled_stream() -> (SignalStream, Arc<StalledConn>) {
        let conn = Arc::new(StalledConn { closed: AtomicBool::new(false) });
        let dyn_conn: Arc<dyn livekit_net::WsConnection> = conn.clone();
        let (emitter, _events) = mpsc::unbounded_channel();
        let (internal_tx, internal_rx) = mpsc::channel::<InternalMessage>(8);
        let write_handle = tokio::spawn(SignalStream::write_task(internal_rx, dyn_conn.clone()));
        let read_handle =
            tokio::spawn(SignalStream::read_task(internal_tx.clone(), dyn_conn, emitter));
        (SignalStream { internal_tx, read_handle, write_handle }, conn)
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn close_returns_on_a_dead_link() {
        let (stream, conn) = stalled_stream();
        let result = tokio::time::timeout(Duration::from_secs(10), stream.close(true)).await;
        assert!(result.is_ok(), "SignalStream::close() did not return on a dead link");
        assert!(conn.closed.load(Ordering::SeqCst), "conn.close() was never called");
    }
}

Run it:

cargo test -p livekit-signaling --lib blackhole

Result on main:

test signal_stream::blackhole_tests::close_returns_on_a_dead_link ... FAILED
panicked at: SignalStream::close() did not return on a dead link
test result: FAILED. 0 passed; 1 failed; ... finished in 10.00s

With the fix below, the test passes. Add the shutdown_rx argument to the read_task call in stalled_stream when you apply the fix.

We also have an end-to-end test that puts a TCP proxy in front of the signal WebSocket only, and then stops the proxy from forwarding data while it holds both sockets open. It reproduces the same failure against livekit-server --dev. We can include it in the PR if you want it.

Suggested fix

1. Give read_task a oneshot::Receiver to select on, so that SignalStream::close can stop it. A dropped sender is sufficient, so a dropped SignalStream also stops its reader.
2. Add a CLOSE_DRAIN_TIMEOUT backstop on the joins, so that close stays bounded if a transport fails in some other way.

The patch also takes the stream out of the slot before it closes it, so that SignalInner::close does not hold the write lock across a network operation. This is not required for the fix, and edition 2024 would change the guard scope anyway. We include it because the terminal teardown is the one path where the lock protects nothing. Say the word and we will drop that hunk.

diff --git a/livekit-signaling/src/lib.rs b/livekit-signaling/src/lib.rs
index 7771b94a..0f392d1c 100644
--- a/livekit-signaling/src/lib.rs
+++ b/livekit-signaling/src/lib.rs
@@ -613,9 +613,14 @@ impl SignalInner {
         self.flush_queue().await;
     }
 
-    /// Close the connection
+    /// Close the connection.
+    ///
+    /// Take the stream out of the slot before closing it, so the write lock is
+    /// not held across the shutdown. `restart` and pass-through sends would
+    /// otherwise queue behind a network operation.
     pub async fn close(&self, notify_close: bool) {
-        if let Some(stream) = self.stream.write().await.take() {
+        let stream = self.stream.write().await.take();
+        if let Some(stream) = stream {
             stream.close(notify_close).await;
         }
     }
diff --git a/livekit-signaling/src/signal_stream.rs b/livekit-signaling/src/signal_stream.rs
index 4efd7505..0e0f0c2c 100644
--- a/livekit-signaling/src/signal_stream.rs
+++ b/livekit-signaling/src/signal_stream.rs
@@ -30,12 +30,20 @@ enum InternalMessage {
     Close,
 }
 
+/// Grace period for the read and write tasks to stop during [`SignalStream::close`].
+/// A link that dies without FIN or RST can leave a task parked in the connection.
+/// `close` must stay bounded in that case.
+const CLOSE_DRAIN_TIMEOUT: Duration = Duration::from_secs(2);
+
 /// SignalStream holds the WebSocket connection (via `WsConnection`).
 ///
 /// It is replaced by [SignalClient] at each reconnection.
 #[derive(Debug)]
 pub(super) struct SignalStream {
     internal_tx: mpsc::Sender<InternalMessage>,
+    /// Dropped to stop `read_task`, which is otherwise parked in a `recv()`
+    /// that never returns.
+    shutdown_tx: oneshot::Sender<()>,
     read_handle: JoinHandle<()>,
     write_handle: JoinHandle<()>,
 }
@@ -75,20 +83,43 @@ impl SignalStream {
 
         let (emitter, events) = mpsc::unbounded_channel();
         let (internal_tx, internal_rx) = mpsc::channel::<InternalMessage>(8);
+        let (shutdown_tx, shutdown_rx) = oneshot::channel();
         let write_handle = tokio::spawn(Self::write_task(internal_rx, conn.clone()));
-        let read_handle = tokio::spawn(Self::read_task(internal_tx.clone(), conn, emitter));
+        let read_handle =
+            tokio::spawn(Self::read_task(internal_tx.clone(), conn, emitter, shutdown_rx));
 
-        Ok((Self { internal_tx, read_handle, write_handle }, events))
+        Ok((Self { internal_tx, shutdown_tx, read_handle, write_handle }, events))
     }
 
     /// Close the websocket.
     /// It sends a Close message before closing.
+    ///
+    /// Bounded by [`CLOSE_DRAIN_TIMEOUT`]. Callers hold the stream lock across
+    /// this call, and the room shutdown goes through it, so it must not depend
+    /// on the peer. A link that dies without FIN or RST never delivers the close
+    /// that the read task waits for, so stop that task instead.
     pub async fn close(self, notify_close: bool) {
+        let Self { internal_tx, shutdown_tx, read_handle, write_handle } = self;
+
         if notify_close {
-            let _ = self.internal_tx.send(InternalMessage::Close).await;
+            // Best effort: the peer can already be unreachable.
+            let _ =
+                tokio::time::timeout(CLOSE_DRAIN_TIMEOUT, internal_tx.send(InternalMessage::Close))
+                    .await;
+        }
+
+        // Stop the read task first. It holds a clone of `internal_tx`, so the
+        // write task's channel closes only after the read task exits.
+        drop(shutdown_tx);
+        drop(internal_tx);
+
+        let drain = async {
+            let _ = read_handle.await;
+            let _ = write_handle.await;
+        };
+        if tokio::time::timeout(CLOSE_DRAIN_TIMEOUT, drain).await.is_err() {
+            log::warn!("signal stream did not shut down within {CLOSE_DRAIN_TIMEOUT:?}");
         }
-        let _ = self.write_handle.await;
-        let _ = self.read_handle.await;
     }
 
     /// Send a SignalRequest to the websocket.
@@ -134,9 +165,17 @@ impl SignalStream {
         internal_tx: mpsc::Sender<InternalMessage>,
         conn: Arc<dyn livekit_net::WsConnection>,
         emitter: mpsc::UnboundedSender<Box<proto::signal_response::Message>>,
+        mut shutdown_rx: oneshot::Receiver<()>,
     ) {
         loop {
-            match conn.recv().await {
+            // `recv` never returns on a link that delivers no data: no data, no
+            // FIN, no RST. Without this arm the task outlives the SignalStream
+            // and blocks every caller that awaits its JoinHandle.
+            let received = tokio::select! {
+                result = conn.recv() => result,
+                _ = &mut shutdown_rx => break,
+            };
+            match received {
                 Ok(Some(bytes)) => {
                     match proto::SignalResponse::decode(bytes.as_slice()) {
                         Ok(res) => {

Notes

• read_task holds a clone of internal_tx (signal_stream.rs:79), so drop(internal_tx) alone does not close the write task's channel. The drain above therefore awaits read_handle first. The CLOSE_DRAIN_TIMEOUT bounds the wait in either order.
• After the fix, a Room::close() that races an in-flight reconnect can still take about 14 s. restart holds the stream write lock across the reconnect on purpose, so that pass-through sends go to the new stream. The Leave waits for CLOSE_DRAIN_TIMEOUT (2 s), SIGNAL_CONNECT_TIMEOUT (5 s) and JOIN_RESPONSE_TIMEOUT (5 s) on the read lock. The final signal_client.close() adds 2 s more. This looks acceptable, but callers cannot assume that close() is fast. It would be nice to document the upper bound on how long close() can take.
• The unrecoverable state described in Effect is a separate problem, and it is already known: Close peer connections before awaiting signal teardown #1335 left the Room-dropped path untouched on purpose, because it needs the spawned tasks stopped before the graph can drop. We did not change it either. We note it only because it is the reason a slow close() became a room that the device could never leave.