From cccc108ef51034d558a06d20fdb163417b2e561f Mon Sep 17 00:00:00 2001 From: Allen R Chan Date: Fri, 16 Oct 2026 23:27:46 -0700 Subject: [PATCH 1/5] fix(signaling): retain unacknowledged queueable requests --- ...n-queueable-signals-after-write-failure.md | 4 + livekit-signaling/src/lib.rs | 157 +++++++++++++++++- 2 files changed, 156 insertions(+), 5 deletions(-) create mode 100644 .changeset/retain-queueable-signals-after-write-failure.md diff --git a/.changeset/retain-queueable-signals-after-write-failure.md b/.changeset/retain-queueable-signals-after-write-failure.md new file mode 100644 index 000000000..b306614c3 --- /dev/null +++ b/.changeset/retain-queueable-signals-after-write-failure.md @@ -0,0 +1,4 @@ +--- +"livekit-signaling": patch +--- +Retain failed and unattempted queueable signals in FIFO order until a transport acknowledges them, without replaying a successfully written prefix after reconnecting. diff --git a/livekit-signaling/src/lib.rs b/livekit-signaling/src/lib.rs index 226ad1372..a1743ad72 100644 --- a/livekit-signaling/src/lib.rs +++ b/livekit-signaling/src/lib.rs @@ -16,6 +16,7 @@ use std::{ borrow::Cow, + collections::VecDeque, fmt::Debug, io::Write, sync::{ @@ -314,7 +315,7 @@ struct SignalInner { /// Session-scoped signals held while no confirmed transport can carry them. Always taken /// under `state`, so the lock order is fixed, and held across a drain so a concurrent /// sender cannot slip a newer signal between two held ones. - queue: AsyncMutex>, + queue: AsyncMutex>, url: String, options: SignalOptions, join_response: proto::JoinResponse, @@ -761,6 +762,12 @@ impl SignalInner { let state = self.state.read().await; let stream = match &*state { SignalState::Connected(stream) => { + if !is_pass_through(&signal) { + let mut queue = self.queue.lock().await; + queue.push_back(signal); + Self::flush_pending(&mut queue, stream).await; + return; + } self.flush_queue(stream).await; Some(stream) } @@ -786,7 +793,7 @@ impl SignalInner { if is_pass_through(&signal) { log::warn!("dropping pass-through signal: {why}"); } else { - self.queue.lock().await.push(signal); + self.queue.lock().await.push_back(signal); } } @@ -794,10 +801,20 @@ impl SignalInner { /// for the whole drain: a concurrent `Connected` send waits here instead of overtaking. async fn flush_queue(&self, stream: &SignalStream) { let mut queue = self.queue.lock().await; - for signal in queue.drain(..) { - if let Err(err) = stream.send(signal).await { - log::error!("failed to send queued signal: {}", err); // Lost message + Self::flush_pending(&mut queue, stream).await; + } + + /// Remove only acknowledged writes, retaining the unsent FIFO on failure or cancellation. + async fn flush_pending( + queue: &mut VecDeque, + stream: &SignalStream, + ) { + while let Some(signal) = queue.front() { + if let Err(err) = stream.send(signal.clone()).await { + log::error!("failed to send queued signal; retaining unsent requests: {}", err); + break; } + queue.pop_front(); } } @@ -1515,6 +1532,136 @@ mod tests { .collect() } + /// Fail one selected queueable write after recording its successful prefix. + struct FailingConn { + fail_sid: &'static str, + attempts: Mutex>, + } + + #[async_trait::async_trait] + impl livekit_net::WsConnection for FailingConn { + async fn send(&self, frame: Vec) -> Result<(), livekit_net::TransportError> { + let request = proto::SignalRequest::decode(frame.as_slice()) + .expect("the signal stream must encode a valid request"); + if let Some(proto::signal_request::Message::Mute(request)) = request.message { + let failed = request.sid == self.fail_sid; + self.attempts.lock().push((request.sid, failed)); + if failed { + return Err(livekit_net::TransportError::Connection( + "test write failed".into(), + )); + } + } + Ok(()) + } + + async fn recv(&self) -> Result>, livekit_net::TransportError> { + std::future::pending().await + } + + async fn close(&self) {} + } + + #[tokio::test] + async fn failed_flush_retains_unsent_signals_for_the_next_connection() { + let inner = make_stub_inner(); + inner.send(mute("successful-prefix")).await; + inner.send(mute("failed-head")).await; + inner.send(mute("unsent-tail")).await; + let failed_conn = + Arc::new(FailingConn { fail_sid: "failed-head", attempts: Mutex::new(Vec::new()) }); + let (stream, _events) = SignalStream::spawn(failed_conn.clone()); + *inner.state.write().await = SignalState::Reconnecting(Some(stream)); + + inner.set_reconnected().await; + let after_failure = queued_sids(&inner).await; + inner.transport_failed().await; + + inner.send(mute("later-request")).await; + let after_later_send = queued_sids(&inner).await; + let recovered_conn = Arc::new(RecordingConn { sent: Mutex::new(Vec::new()) }); + let (stream, _events) = SignalStream::spawn(recovered_conn.clone()); + *inner.state.write().await = SignalState::Reconnecting(Some(stream)); + inner.set_reconnected().await; + let after_recovery = queued_sids(&inner).await; + inner.close(true).await; + + assert_eq!( + *failed_conn.attempts.lock(), + vec![("successful-prefix".into(), false), ("failed-head".into(), true)], + "the external transport must fail the selected write after the successful prefix" + ); + assert_eq!( + after_failure, + vec!["failed-head", "unsent-tail"], + "a failed flush must retain the failed head and every unattempted signal" + ); + assert_eq!(after_later_send, vec!["failed-head", "unsent-tail", "later-request"]); + assert_eq!( + sent_sids(&recovered_conn), + vec!["failed-head", "unsent-tail", "later-request"], + "recovery must preserve FIFO without replaying the successful prefix" + ); + assert!(after_recovery.is_empty()); + assert_eq!(Arc::strong_count(&failed_conn), 1, "the failed stream must release its tasks"); + assert_eq!( + Arc::strong_count(&recovered_conn), + 1, + "the recovered stream must release its tasks" + ); + } + + /// Reports a started write but never acknowledges it. + struct StalledWriteConn { + started: tokio::sync::Notify, + } + + #[async_trait::async_trait] + impl livekit_net::WsConnection for StalledWriteConn { + async fn send(&self, _frame: Vec) -> Result<(), livekit_net::TransportError> { + self.started.notify_one(); + std::future::pending().await + } + + async fn recv(&self) -> Result>, livekit_net::TransportError> { + std::future::pending().await + } + + async fn close(&self) {} + } + + #[tokio::test] + async fn cancelled_send_retains_the_unacknowledged_queueable_signal() { + let conn = Arc::new(StalledWriteConn { started: tokio::sync::Notify::new() }); + let (stream, _events) = SignalStream::spawn(conn.clone()); + let inner = make_stub_inner(); + *inner.state.write().await = SignalState::Connected(stream); + let sending = tokio::spawn({ + let inner = inner.clone(); + async move { inner.send(mute("unacknowledged-request")).await } + }); + tokio::time::timeout(Duration::from_secs(1), conn.started.notified()) + .await + .expect("the external write must have started before cancellation"); + sending.abort(); + assert!(sending.await.expect_err("send task must be cancelled").is_cancelled()); + let after_cancellation = queued_sids(&inner).await; + inner.transport_failed().await; + + let recovered_conn = Arc::new(RecordingConn { sent: Mutex::new(Vec::new()) }); + let (stream, _events) = SignalStream::spawn(recovered_conn.clone()); + *inner.state.write().await = SignalState::Reconnecting(Some(stream)); + inner.set_reconnected().await; + let after_recovery = queued_sids(&inner).await; + inner.close(true).await; + + assert_eq!(after_cancellation, vec!["unacknowledged-request"]); + assert_eq!(sent_sids(&recovered_conn), vec!["unacknowledged-request"]); + assert!(after_recovery.is_empty()); + assert_eq!(Arc::strong_count(&conn), 1, "the stalled writer must be joined"); + assert_eq!(Arc::strong_count(&recovered_conn), 1, "the recovered writer must be joined"); + } + /// Held signals are older than anything a concurrent sender has, so they must all reach /// the wire before it. The read lock is shared, so only the queue lock can enforce that. #[tokio::test] From 7058c60b8f366b9f2be59cad5281443a0f9e6c92 Mon Sep 17 00:00:00 2001 From: Allen R Chan Date: Fri, 16 Oct 2026 23:32:05 -0700 Subject: [PATCH 2/5] docs: include downstream signaling patch releases --- .changeset/retain-queueable-signals-after-write-failure.md | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/.changeset/retain-queueable-signals-after-write-failure.md b/.changeset/retain-queueable-signals-after-write-failure.md index b306614c3..f3978a517 100644 --- a/.changeset/retain-queueable-signals-after-write-failure.md +++ b/.changeset/retain-queueable-signals-after-write-failure.md @@ -1,4 +1,8 @@ --- -"livekit-signaling": patch +livekit: patch +livekit-api: patch +livekit-capture: patch +livekit-ffi: patch +livekit-signaling: patch --- Retain failed and unattempted queueable signals in FIFO order until a transport acknowledges them, without replaying a successfully written prefix after reconnecting. From 0373740338804d3975cfb0c9c33a07ccfe863149 Mon Sep 17 00:00:00 2001 From: Allen R Chan Date: Sat, 17 Oct 2026 00:21:24 -0700 Subject: [PATCH 3/5] fix(signaling): retain writer receipts across caller cancellation --- ...n-queueable-signals-after-write-failure.md | 2 +- livekit-signaling/src/lib.rs | 173 ++++++++++++++++-- livekit-signaling/src/signal_stream.rs | 36 +++- 3 files changed, 195 insertions(+), 16 deletions(-) diff --git a/.changeset/retain-queueable-signals-after-write-failure.md b/.changeset/retain-queueable-signals-after-write-failure.md index f3978a517..2a22da7cf 100644 --- a/.changeset/retain-queueable-signals-after-write-failure.md +++ b/.changeset/retain-queueable-signals-after-write-failure.md @@ -5,4 +5,4 @@ livekit-capture: patch livekit-ffi: patch livekit-signaling: patch --- -Retain failed and unattempted queueable signals in FIFO order until a transport acknowledges them, without replaying a successfully written prefix after reconnecting. +Retain failed and unattempted queueable signals in FIFO order until a transport acknowledges them, preserving acknowledgements across caller cancellation so successful writes are not replayed on the same connection or after reconnecting. diff --git a/livekit-signaling/src/lib.rs b/livekit-signaling/src/lib.rs index a1743ad72..6784a5646 100644 --- a/livekit-signaling/src/lib.rs +++ b/livekit-signaling/src/lib.rs @@ -39,7 +39,7 @@ use tokio::{ time::{interval, sleep, Instant}, }; -use crate::signal_stream::SignalStream; +use crate::signal_stream::{SignalStream, SignalWrite}; use livekit_net::HttpClientExt; mod region_url_provider; @@ -308,6 +308,38 @@ impl SignalState { } } +/// Retain both the request and any admitted write until its result is observed. +struct QueuedSignal { + signal: proto::signal_request::Message, + write: Option, +} + +impl From for QueuedSignal { + fn from(signal: proto::signal_request::Message) -> Self { + Self { signal, write: None } + } +} + +impl QueuedSignal { + /// Settle an admitted write before removing or replaying the request. + async fn send(&mut self, stream: &SignalStream) -> SignalResult<()> { + loop { + if let Some(write) = self.write.as_mut() { + let same_stream = write.belongs_to(stream); + let result = write.complete().await; + self.write = None; + if same_stream || result.is_ok() { + return result; + } + // Only an unsuccessful old-stream attempt may be retried here. + // A successful write is not replayed, even after reconnecting. + } else { + self.write = Some(stream.begin_send(self.signal.clone()).await?); + } + } + } +} + struct SignalInner { // The write lock is held across reconnects to avoid transportless gaps state: AsyncRwLock, @@ -315,7 +347,7 @@ struct SignalInner { /// Session-scoped signals held while no confirmed transport can carry them. Always taken /// under `state`, so the lock order is fixed, and held across a drain so a concurrent /// sender cannot slip a newer signal between two held ones. - queue: AsyncMutex>, + queue: AsyncMutex>, url: String, options: SignalOptions, join_response: proto::JoinResponse, @@ -764,7 +796,7 @@ impl SignalInner { SignalState::Connected(stream) => { if !is_pass_through(&signal) { let mut queue = self.queue.lock().await; - queue.push_back(signal); + queue.push_back(signal.into()); Self::flush_pending(&mut queue, stream).await; return; } @@ -793,7 +825,7 @@ impl SignalInner { if is_pass_through(&signal) { log::warn!("dropping pass-through signal: {why}"); } else { - self.queue.lock().await.push_back(signal); + self.queue.lock().await.push_back(signal.into()); } } @@ -805,12 +837,9 @@ impl SignalInner { } /// Remove only acknowledged writes, retaining the unsent FIFO on failure or cancellation. - async fn flush_pending( - queue: &mut VecDeque, - stream: &SignalStream, - ) { - while let Some(signal) = queue.front() { - if let Err(err) = stream.send(signal.clone()).await { + async fn flush_pending(queue: &mut VecDeque, stream: &SignalStream) { + while let Some(signal) = queue.front_mut() { + if let Err(err) = signal.send(stream).await { log::error!("failed to send queued signal; retaining unsent requests: {}", err); break; } @@ -1403,7 +1432,7 @@ mod tests { .lock() .await .iter() - .filter_map(|signal| match signal { + .filter_map(|queued| match &queued.signal { proto::signal_request::Message::Mute(m) => Some(m.sid.clone()), _ => None, }) @@ -1662,6 +1691,126 @@ mod tests { assert_eq!(Arc::strong_count(&recovered_conn), 1, "the recovered writer must be joined"); } + /// Hold the first external write until the test explicitly permits it to succeed. + struct DelayedWriteConn { + attempts: Mutex>, + started: tokio::sync::Notify, + release: tokio::sync::Notify, + completed: tokio::sync::Notify, + } + + #[async_trait::async_trait] + impl livekit_net::WsConnection for DelayedWriteConn { + async fn send(&self, frame: Vec) -> Result<(), livekit_net::TransportError> { + let request = proto::SignalRequest::decode(frame.as_slice()) + .expect("the signal stream must encode a valid request"); + let Some(proto::signal_request::Message::Mute(request)) = request.message else { + return Ok(()); + }; + let first = { + let mut attempts = self.attempts.lock(); + let first = attempts.is_empty(); + attempts.push(request.sid); + first + }; + if first { + self.started.notify_one(); + self.release.notified().await; + self.completed.notify_one(); + } + Ok(()) + } + + async fn recv(&self) -> Result>, livekit_net::TransportError> { + std::future::pending().await + } + + async fn close(&self) {} + } + + #[tokio::test] + async fn cancelled_write_that_succeeds_is_not_replayed_on_the_same_connection() { + let conn = Arc::new(DelayedWriteConn { + attempts: Mutex::new(Vec::new()), + started: tokio::sync::Notify::new(), + release: tokio::sync::Notify::new(), + completed: tokio::sync::Notify::new(), + }); + let (stream, _events) = SignalStream::spawn(conn.clone()); + let inner = make_stub_inner(); + *inner.state.write().await = SignalState::Connected(stream); + let sending = tokio::spawn({ + let inner = inner.clone(); + async move { inner.send(mute("track-a")).await } + }); + tokio::time::timeout(Duration::from_secs(1), conn.started.notified()) + .await + .expect("the first external write must start before cancelling its caller"); + sending.abort(); + assert!(sending.await.expect_err("send task must be cancelled").is_cancelled()); + conn.release.notify_one(); + tokio::time::timeout(Duration::from_secs(1), conn.completed.notified()) + .await + .expect("the original writer must finish successfully before the next send"); + + inner.send(mute("track-b")).await; + let queued = queued_sids(&inner).await; + inner.close(true).await; + + assert_eq!( + *conn.attempts.lock(), + vec!["track-a", "track-b"], + "caller cancellation must not duplicate an acknowledged write on the same socket" + ); + assert!(queued.is_empty()); + assert_eq!(Arc::strong_count(&conn), 1, "the stream must release its reader and writer"); + } + + #[tokio::test] + async fn cancelled_write_that_succeeds_is_not_replayed_after_reconnect() { + let conn = Arc::new(DelayedWriteConn { + attempts: Mutex::new(Vec::new()), + started: tokio::sync::Notify::new(), + release: tokio::sync::Notify::new(), + completed: tokio::sync::Notify::new(), + }); + let (stream, _events) = SignalStream::spawn(conn.clone()); + let inner = make_stub_inner(); + *inner.state.write().await = SignalState::Connected(stream); + let sending = tokio::spawn({ + let inner = inner.clone(); + async move { inner.send(mute("track-a")).await } + }); + tokio::time::timeout(Duration::from_secs(1), conn.started.notified()) + .await + .expect("the first external write must start before cancelling its caller"); + sending.abort(); + assert!(sending.await.expect_err("send task must be cancelled").is_cancelled()); + conn.release.notify_one(); + tokio::time::timeout(Duration::from_secs(1), conn.completed.notified()) + .await + .expect("the original writer must finish successfully before reconnecting"); + inner.transport_failed().await; + + let recovered_conn = Arc::new(RecordingConn { sent: Mutex::new(Vec::new()) }); + let (stream, _events) = SignalStream::spawn(recovered_conn.clone()); + *inner.state.write().await = SignalState::Reconnecting(Some(stream)); + inner.set_reconnected().await; + inner.send(mute("track-b")).await; + let queued = queued_sids(&inner).await; + inner.close(true).await; + + assert_eq!(*conn.attempts.lock(), vec!["track-a"]); + assert_eq!( + sent_sids(&recovered_conn), + vec!["track-b"], + "a successful old-stream acknowledgement must prevent replay after reconnecting" + ); + assert!(queued.is_empty()); + assert_eq!(Arc::strong_count(&conn), 1, "the original writer must be joined"); + assert_eq!(Arc::strong_count(&recovered_conn), 1, "the recovered writer must be joined"); + } + /// Held signals are older than anything a concurrent sender has, so they must all reach /// the wire before it. The read lock is shared, so only the queue lock can enforce that. #[tokio::test] @@ -1670,7 +1819,7 @@ mod tests { let (stream, _events) = SignalStream::spawn(conn.clone()); let inner = make_stub_inner(); *inner.state.write().await = SignalState::Connected(stream); - inner.queue.lock().await.extend([mute("held-1"), mute("held-2")]); + inner.queue.lock().await.extend([mute("held-1"), mute("held-2")].map(QueuedSignal::from)); let a = tokio::spawn({ let inner = inner.clone(); diff --git a/livekit-signaling/src/signal_stream.rs b/livekit-signaling/src/signal_stream.rs index faabb8f29..35c45898d 100644 --- a/livekit-signaling/src/signal_stream.rs +++ b/livekit-signaling/src/signal_stream.rs @@ -43,6 +43,25 @@ pub(super) struct SignalStream { write_handle: JoinHandle<()>, } +/// A receipt for an admitted write that can outlive the sending caller. +pub(super) struct SignalWrite { + response: oneshot::Receiver>, + // A queue entry must not keep an old stream's writer alive during close. + stream: mpsc::WeakSender, +} + +impl SignalWrite { + /// Identify the writer without retaining a strong channel ownership edge. + pub(super) fn belongs_to(&self, stream: &SignalStream) -> bool { + self.stream.upgrade().is_some_and(|sender| sender.same_channel(&stream.internal_tx)) + } + + /// Await by mutable reference so cancellation preserves the acknowledgement. + pub(super) async fn complete(&mut self) -> SignalResult<()> { + (&mut self.response).await.map_err(|_| SignalError::SendError)? + } +} + /// Terse: this appears inside `SignalState` in logs and tests. impl std::fmt::Debug for SignalStream { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { @@ -126,10 +145,21 @@ impl SignalStream { /// Send a SignalRequest to the websocket. /// It also waits for the message to be sent. pub async fn send(&self, signal: proto::signal_request::Message) -> SignalResult<()> { + self.begin_send(signal).await?.complete().await + } + + /// Admit a write and return its receipt without another cancellation point. + pub(super) async fn begin_send( + &self, + signal: proto::signal_request::Message, + ) -> SignalResult { + let permit = self.internal_tx.reserve().await.map_err(|_| SignalError::SendError)?; let (send, recv) = oneshot::channel(); - let msg = InternalMessage::Signal { signal, response_chn: send }; - let _ = self.internal_tx.send(msg).await; - recv.await.map_err(|_| SignalError::SendError)? + // Once admitted, delivery and returning the receipt are synchronous. A + // cancelled capacity wait has not sent anything; a cancelled completion + // wait leaves the receipt owned by the queue entry. + permit.send(InternalMessage::Signal { signal, response_chn: send }); + Ok(SignalWrite { response: recv, stream: self.internal_tx.downgrade() }) } /// This task is used to send messages to the websocket. From 20b79f79b01ce02211813b128a5b7d9f5e649ab9 Mon Sep 17 00:00:00 2001 From: Allen R Chan Date: Sat, 17 Oct 2026 03:40:26 -0700 Subject: [PATCH 4/5] ci: retain macOS reconnection crash reports --- .github/workflows/tests.yml | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index d3eb0d98c..05ab989e0 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -166,6 +166,7 @@ jobs: # TODO: Remove once this is enabled by default. enable_participant_data_blob: true - name: Test + id: tests env: RUST_LOG: info shell: bash @@ -193,3 +194,16 @@ jobs: run: | cargo build --verbose --target ${{ matrix.target }} -p livekit-net --features uniffi cargo build --verbose --target ${{ matrix.target }} -p livekit-net --features uniffi,native + + # Preserve the failing test and collect only this native test process's + # OS crash reports, never a process-memory core or unrelated host logs. + - name: Upload macOS reconnection crash reports + if: ${{ failure() && runner.os == 'macOS' && steps.tests.outcome == 'failure' }} + uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4.6.2 + with: + name: reconnection-crash-reports-${{ matrix.target }} + path: | + ~/Library/Logs/DiagnosticReports/reconnection_test*.ips + ~/Library/Logs/DiagnosticReports/reconnection_test*.crash + if-no-files-found: warn + retention-days: 7 From 79868965a55a07c5973eedec0a32c8ed896b70f3 Mon Sep 17 00:00:00 2001 From: Allen R Chan Date: Sat, 17 Oct 2026 05:02:36 -0700 Subject: [PATCH 5/5] ci: observe later test targets after an earlier failure --- .github/workflows/tests.yml | 3 +++ 1 file changed, 3 insertions(+) diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 05ab989e0..c1e2ba6a8 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -171,7 +171,10 @@ jobs: RUST_LOG: info shell: bash run: | + # Keep the job failed, but observe later test binaries too. An earlier + # data-track failure must not hide native reconnection crash reports. cargo test --verbose \ + --no-fail-fast \ --target ${{ matrix.target }} \ --features __lk-e2e-test \ -- \