diff --git a/.changeset/abort_in_flight_ffi_connect.md b/.changeset/abort_in_flight_ffi_connect.md new file mode 100644 index 000000000..de2ccce9b --- /dev/null +++ b/.changeset/abort_in_flight_ffi_connect.md @@ -0,0 +1,15 @@ +--- +livekit-ffi: patch +--- + +# Abort in-flight FFI connect on disconnect + +Cancelling `room.connect()` from a language binding left `Room::connect` running, +and a later ReadyFor timeout sent Panic into the host (python-sdks#785). Connect +now allocates an abortable handle immediately; DisconnectRequest cancels the +handshake, joins the connect task, and `close()`s any Room that already +completed (rust-sdks#1334: dropping a completed Room without close keeps +`engine_task`/`room_task` alive). Abort before `Room::connect` returns Ok does +not enter `SessionInner::close`; that leftover stays in `livekit/`. ReadyFor on +that early handle is accepted during handshake. Missed ReadyFor fails that room +instead of panicking the process. diff --git a/livekit-ffi-node-bindings/proto/room_pb.d.ts b/livekit-ffi-node-bindings/proto/room_pb.d.ts index 069ca8f5c..6e8310a47 100644 --- a/livekit-ffi-node-bindings/proto/room_pb.d.ts +++ b/livekit-ffi-node-bindings/proto/room_pb.d.ts @@ -308,6 +308,18 @@ export declare class ConnectResponse extends Message { */ asyncId?: bigint; + /** + * Allocated as soon as ConnectRequest is received, before Room::connect + * finishes. DisconnectRequest with this handle aborts the in-flight + * handshake and close()s any Room that already completed (rust-sdks#1334: + * dropping a completed Room without close leaks ICE sockets). Old clients + * ignore this proto2 optional field. A missed ReadyFor handshake must fail + * this room only, never Panic the host (python-sdks#785). + * + * @generated from field: optional uint64 room_handle = 2; + */ + roomHandle?: bigint; + constructor(data?: PartialMessage); static readonly runtime: typeof proto2; diff --git a/livekit-ffi-node-bindings/proto/room_pb.js b/livekit-ffi-node-bindings/proto/room_pb.js index 6a6a5ed44..edfbc0dc1 100644 --- a/livekit-ffi-node-bindings/proto/room_pb.js +++ b/livekit-ffi-node-bindings/proto/room_pb.js @@ -164,6 +164,7 @@ const ConnectResponse = /*@__PURE__*/ proto2.makeMessageType( "livekit.proto.ConnectResponse", () => [ { no: 1, name: "async_id", kind: "scalar", T: 4 /* ScalarType.UINT64 */, req: true }, + { no: 2, name: "room_handle", kind: "scalar", T: 4 /* ScalarType.UINT64 */, opt: true }, ], ); diff --git a/livekit-ffi/protocol/room.proto b/livekit-ffi/protocol/room.proto index df08ed5be..79c129b3b 100644 --- a/livekit-ffi/protocol/room.proto +++ b/livekit-ffi/protocol/room.proto @@ -35,6 +35,13 @@ message ConnectRequest { } message ConnectResponse { required uint64 async_id = 1; + // Allocated as soon as ConnectRequest is received, before Room::connect + // finishes. DisconnectRequest with this handle aborts the in-flight + // handshake and close()s any Room that already completed (rust-sdks#1334: + // dropping a completed Room without close leaks ICE sockets). Old clients + // ignore this proto2 optional field. A missed ReadyFor handshake must fail + // this room only, never Panic the host (python-sdks#785). + optional uint64 room_handle = 2; } message ConnectCallback { message ParticipantWithTracks { diff --git a/livekit-ffi/src/server/connect_abort_tests.rs b/livekit-ffi/src/server/connect_abort_tests.rs new file mode 100644 index 000000000..b1dc44ab4 --- /dev/null +++ b/livekit-ffi/src/server/connect_abort_tests.rs @@ -0,0 +1,361 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Cancel-during-handshake tests for issue 1340. +//! +//! A TCP listener that accepts and never speaks WebSocket keeps `Room::connect` +//! parked in the handshake. Disconnect on the handle returned by +//! `ConnectResponse` must abort that future, emit a connect error, and never +//! send `Panic` to the host. + +use std::{net::TcpListener, sync::Arc, time::Duration}; + +use parking_lot::Mutex; + +use crate::{ + proto, + server::{requests, room::FfiConnectingRoom, FfiConfig}, + FFI_SERVER, +}; + +/// `FFI_SERVER` is process-wide; serialize these tests so event sinks do not +/// overwrite each other. +static FFI_TEST_LOCK: Mutex<()> = Mutex::new(()); + +fn start_blackhole() -> (u16, std::thread::JoinHandle<()>) { + let listener = TcpListener::bind("127.0.0.1:0").expect("bind blackhole listener"); + let port = listener.local_addr().expect("local_addr").port(); + let thread = std::thread::spawn(move || { + let mut held = Vec::new(); + for stream in listener.incoming() { + match stream { + Ok(s) => held.push(s), + Err(_) => break, + } + } + drop(held); + }); + (port, thread) +} + +fn install_event_sink() -> tokio::sync::mpsc::UnboundedReceiver { + let (tx, rx) = tokio::sync::mpsc::unbounded_channel(); + FFI_SERVER.setup(FfiConfig { + callback_fn: Arc::new(move |event| { + let _ = tx.send(event); + }), + capture_logs: false, + sdk: "test".into(), + sdk_version: "0".into(), + }); + rx +} + +async fn recv_until( + rx: &mut tokio::sync::mpsc::UnboundedReceiver, + timeout: Duration, + mut pred: impl FnMut(&proto::FfiEvent) -> bool, +) -> proto::FfiEvent { + tokio::time::timeout(timeout, async { + loop { + let event = rx.recv().await.expect("ffi event channel closed"); + if let Some(proto::ffi_event::Message::Panic(panic)) = &event.message { + panic!("host Panic event: {}", panic.message); + } + if pred(&event) { + return event; + } + } + }) + .await + .expect("timed out waiting for expected FFI event") +} + +fn connect_blackhole(port: u16) -> proto::ConnectResponse { + crate::server::room::FfiRoom::connect( + &FFI_SERVER, + proto::ConnectRequest { + url: format!("ws://127.0.0.1:{port}"), + token: "test".into(), + ..Default::default() + }, + ) +} + +fn disconnect_in_flight(room_handle: u64) -> u64 { + let disconnect = requests::handle_request( + &FFI_SERVER, + proto::FfiRequest { + message: Some(proto::ffi_request::Message::Disconnect(proto::DisconnectRequest { + room_handle, + ..Default::default() + })), + }, + ) + .expect("disconnect of an in-flight connect must succeed"); + match disconnect.message { + Some(proto::ffi_response::Message::Disconnect(resp)) => resp.async_id, + other => panic!("expected DisconnectResponse, got {other:?}"), + } +} + +fn assert_connect_cancelled(event: proto::FfiEvent, async_id: u64) { + match event.message { + Some(proto::ffi_event::Message::Connect(cb)) => { + assert_eq!(cb.async_id, async_id); + match cb.message { + Some(proto::connect_callback::Message::Error(err)) => { + assert!( + err.to_lowercase().contains("cancel"), + "expected cancelled connect, got {err}" + ); + } + other => panic!("expected ConnectCallback error, got {other:?}"), + } + } + other => panic!("expected Connect event, got {other:?}"), + } +} + +fn assert_no_live_room(room_handle: u64) { + assert!( + FFI_SERVER.list_rooms().into_iter().all(|room| room.inner.handle_id != room_handle), + "aborted connect must not leave a live FfiRoom handle" + ); + assert!( + FFI_SERVER.retrieve_handle::(room_handle).is_err(), + "aborted connect must not leave an FfiConnectingRoom handle" + ); +} + +#[test] +fn disconnect_aborts_in_flight_connect() { + let _lock = FFI_TEST_LOCK.lock(); + let (port, _blackhole) = start_blackhole(); + let mut events = install_event_sink(); + + let response = connect_blackhole(port); + let room_handle = + response.room_handle.expect("ConnectResponse must allocate a room_handle immediately"); + + let _ = disconnect_in_flight(room_handle); + + let connect_cb = FFI_SERVER.async_runtime.block_on(recv_until( + &mut events, + Duration::from_secs(15), + |event| { + matches!( + &event.message, + Some(proto::ffi_event::Message::Connect(cb)) + if cb.async_id == response.async_id + ) + }, + )); + + assert_connect_cancelled(connect_cb, response.async_id); + assert_no_live_room(room_handle); +} + +#[test] +fn drop_handle_aborts_in_flight_connect() { + let _lock = FFI_TEST_LOCK.lock(); + let (port, _blackhole) = start_blackhole(); + let mut events = install_event_sink(); + + let response = connect_blackhole(port); + let room_handle = + response.room_handle.expect("ConnectResponse must allocate a room_handle immediately"); + + assert!(FFI_SERVER.drop_handle(room_handle), "drop_handle must find the connecting handle"); + + let connect_cb = FFI_SERVER.async_runtime.block_on(recv_until( + &mut events, + Duration::from_secs(15), + |event| { + matches!( + &event.message, + Some(proto::ffi_event::Message::Connect(cb)) + if cb.async_id == response.async_id + ) + }, + )); + + assert_connect_cancelled(connect_cb, response.async_id); + assert_no_live_room(room_handle); +} + +#[test] +fn disconnect_callback_waits_until_connect_settles() { + let _lock = FFI_TEST_LOCK.lock(); + let (port, _blackhole) = start_blackhole(); + let mut events = install_event_sink(); + + let response = connect_blackhole(port); + let room_handle = + response.room_handle.expect("ConnectResponse must allocate a room_handle immediately"); + let disconnect_async_id = disconnect_in_flight(room_handle); + + let mut saw_connect_error = false; + FFI_SERVER.async_runtime.block_on(recv_until(&mut events, Duration::from_secs(15), |event| { + if let Some(proto::ffi_event::Message::Connect(cb)) = &event.message { + if cb.async_id == response.async_id { + match &cb.message { + Some(proto::connect_callback::Message::Error(err)) => { + assert!( + err.to_lowercase().contains("cancel"), + "expected cancelled connect, got {err}" + ); + saw_connect_error = true; + } + other => panic!("expected ConnectCallback error, got {other:?}"), + } + } + } + if let Some(proto::ffi_event::Message::Disconnect(cb)) = &event.message { + if cb.async_id == disconnect_async_id { + assert!( + saw_connect_error, + "DisconnectCallback must arrive after ConnectCallback error \ + (handshake actually aborted)" + ); + assert_no_live_room(room_handle); + return true; + } + } + false + })); +} + +#[test] +fn connect_after_aborted_connect_gets_a_fresh_handle() { + let _lock = FFI_TEST_LOCK.lock(); + let (port, _blackhole) = start_blackhole(); + let mut events = install_event_sink(); + + let first = connect_blackhole(port); + let first_handle = + first.room_handle.expect("ConnectResponse must allocate a room_handle immediately"); + let _ = disconnect_in_flight(first_handle); + + let first_cb = FFI_SERVER.async_runtime.block_on(recv_until( + &mut events, + Duration::from_secs(15), + |event| { + matches!( + &event.message, + Some(proto::ffi_event::Message::Connect(cb)) + if cb.async_id == first.async_id + ) + }, + )); + assert_connect_cancelled(first_cb, first.async_id); + assert_no_live_room(first_handle); + + let second = connect_blackhole(port); + let second_handle = + second.room_handle.expect("second ConnectResponse must allocate a room_handle"); + assert_ne!( + first_handle, second_handle, + "a new connect after abort must get a different room_handle" + ); + + let _ = disconnect_in_flight(second_handle); + let second_cb = FFI_SERVER.async_runtime.block_on(recv_until( + &mut events, + Duration::from_secs(15), + |event| { + matches!( + &event.message, + Some(proto::ffi_event::Message::Connect(cb)) + if cb.async_id == second.async_id + ) + }, + )); + assert_connect_cancelled(second_cb, second.async_id); + assert_no_live_room(second_handle); +} + +#[test] +fn dispose_cancels_in_flight_connect() { + let _lock = FFI_TEST_LOCK.lock(); + let (port, _blackhole) = start_blackhole(); + let mut events = install_event_sink(); + + let response = connect_blackhole(port); + let room_handle = + response.room_handle.expect("ConnectResponse must allocate a room_handle immediately"); + + // Full `dispose()` clears FFI config and would break later tests on this + // process-wide server. `dispose` calls `cancel_connecting_rooms`; test that. + FFI_SERVER.async_runtime.block_on(async { + tokio::time::timeout(Duration::from_secs(15), FFI_SERVER.cancel_connecting_rooms()) + .await + .expect("cancel_connecting_rooms timed out") + }); + + let connect_cb = FFI_SERVER.async_runtime.block_on(recv_until( + &mut events, + Duration::from_secs(15), + |event| { + matches!( + &event.message, + Some(proto::ffi_event::Message::Connect(cb)) + if cb.async_id == response.async_id + ) + }, + )); + assert_connect_cancelled(connect_cb, response.async_id); + assert_no_live_room(room_handle); +} + +#[test] +fn ready_for_connecting_handle_is_accepted() { + let _lock = FFI_TEST_LOCK.lock(); + let (port, _blackhole) = start_blackhole(); + let mut events = install_event_sink(); + + let response = connect_blackhole(port); + let room_handle = + response.room_handle.expect("ConnectResponse must allocate a room_handle immediately"); + + let ready = requests::handle_request( + &FFI_SERVER, + proto::FfiRequest { + message: Some(proto::ffi_request::Message::ReadyForRoomEvent( + proto::ReadyForRoomEventRequest { room_handle, ..Default::default() }, + )), + }, + ) + .expect("ReadyForRoomEvent on a connecting handle must succeed"); + assert!( + matches!(ready.message, Some(proto::ffi_response::Message::ReadyForRoomEvent(_))), + "expected ReadyForRoomEventResponse, got {:?}", + ready.message + ); + + let _ = disconnect_in_flight(room_handle); + let connect_cb = FFI_SERVER.async_runtime.block_on(recv_until( + &mut events, + Duration::from_secs(15), + |event| { + matches!( + &event.message, + Some(proto::ffi_event::Message::Connect(cb)) + if cb.async_id == response.async_id + ) + }, + )); + assert_connect_cancelled(connect_cb, response.async_id); + assert_no_live_room(room_handle); +} diff --git a/livekit-ffi/src/server/mod.rs b/livekit-ffi/src/server/mod.rs index a4256789a..b9411c8c7 100644 --- a/livekit-ffi/src/server/mod.rs +++ b/livekit-ffi/src/server/mod.rs @@ -54,6 +54,8 @@ pub mod video_stream; #[cfg(test)] mod audio_filter_tests; +#[cfg(test)] +mod connect_abort_tests; #[derive(Clone)] pub struct FfiConfig { @@ -170,9 +172,36 @@ impl FfiServer { .collect() } + /// Snapshot of in-flight connects (handles that are still `FfiConnectingRoom`). + pub fn list_connecting_rooms(&self) -> Vec { + self.ffi_handles + .iter() + .filter_map(|h| h.value().downcast_ref::().cloned()) + .collect() + } + + /// Cancel every in-flight `Room::connect` and wait until those tasks settle. + /// rust-sdks#1340: dispose used to only close live `FfiRoom`s, so a handshake + /// in progress kept running after the server was torn down. `close()` any + /// Room that already completed; abort before connect returns Ok still does + /// not enter `SessionInner::close`. + pub async fn cancel_connecting_rooms(&'static self) { + let connecting = self.list_connecting_rooms(); + let mut finished_flags = Vec::with_capacity(connecting.len()); + for room in connecting { + finished_flags.push(room.finished_flag()); + room.cancel(); + } + for flag in finished_flags { + room::wait_until_flag(&flag).await; + } + } + pub async fn dispose(&'static self) { log::debug!("disposing ffi server"); + self.cancel_connecting_rooms().await; + // Close all rooms let rooms = self.list_rooms(); @@ -260,6 +289,11 @@ impl FfiServer { } pub fn drop_handle(&self, id: FfiHandleId) -> bool { + { + if let Ok(connecting) = self.retrieve_handle::(id) { + connecting.cancel(); + } + } let existed = self.ffi_handles.remove(&id).is_some(); self.handle_dropped_txs.remove(&id); if !existed { diff --git a/livekit-ffi/src/server/requests.rs b/livekit-ffi/src/server/requests.rs index 8ee569885..71a32eb58 100644 --- a/livekit-ffi/src/server/requests.rs +++ b/livekit-ffi/src/server/requests.rs @@ -71,7 +71,24 @@ fn on_disconnect( .map(DisconnectReason::from) .unwrap_or(DisconnectReason::ClientInitiated); + if let Ok(connecting) = + server.retrieve_handle::(disconnect.room_handle) + { + let finished = connecting.finished_flag(); + connecting.cancel(); + let handle = server.async_runtime.spawn(async move { + // Wait until the connect task has finished. If Room::connect already + // returned Ok, settle_aborted_connect close()s that Room. Abort during + // wait_pc_connection still has no Room to close. + room::wait_until_flag(&finished).await; + let _ = server.send_event(proto::DisconnectCallback { async_id }.into()); + }); + server.watch_panic(handle); + return Ok(proto::DisconnectResponse { async_id }); + } + let ffi_room = server.retrieve_handle::(disconnect.room_handle)?.clone(); + ffi_room.cancel(); let handle = server.async_runtime.spawn(async move { ffi_room.close(server, reason).await; @@ -123,8 +140,15 @@ fn on_ready_for_room_event( server: &'static FfiServer, request: proto::ReadyForRoomEventRequest, ) -> FfiResult { - let ffi_room = server.retrieve_handle::(request.room_handle)?.clone(); - ffi_room.ready_for_room_event(); + if let Ok(ffi_room) = server.retrieve_handle::(request.room_handle) { + ffi_room.ready_for_room_event(); + return Ok(proto::ReadyForRoomEventResponse::default()); + } + // ConnectResponse.room_handle is valid during handshake. Queue the permit + // on FfiConnectingRoom so a client that readies as soon as it has the + // handle does not type-error and then time out after ConnectCallback. + let connecting = server.retrieve_handle::(request.room_handle)?; + connecting.ready_for_room_event(); Ok(proto::ReadyForRoomEventResponse::default()) } diff --git a/livekit-ffi/src/server/room.rs b/livekit-ffi/src/server/room.rs index cc92de40e..b1148a20b 100644 --- a/livekit-ffi/src/server/room.rs +++ b/livekit-ffi/src/server/room.rs @@ -20,7 +20,7 @@ use livekit::{prelude::*, registered_audio_filter_plugins, PluginError}; use livekit::{ChatMessage, StreamReader}; use livekit_protocol as lk_proto; use parking_lot::Mutex; -use tokio::sync::{broadcast, mpsc, oneshot, Mutex as AsyncMutex, Notify}; +use tokio::sync::{broadcast, mpsc, oneshot, watch, Mutex as AsyncMutex, Notify}; use tokio::task::JoinHandle; use super::FfiDataBuffer; @@ -49,6 +49,39 @@ pub struct FfiTrack { impl FfiHandle for FfiTrack {} impl FfiHandle for FfiPublication {} impl FfiHandle for FfiRoom {} +impl FfiHandle for FfiConnectingRoom {} + +/// In-flight connect. Stored under `room_handle` from the moment +/// `ConnectRequest` is received so `DisconnectRequest` can abort +/// `Room::connect` before a `Room` exists. +#[derive(Clone)] +pub struct FfiConnectingRoom { + pub handle_id: FfiHandleId, + abort: watch::Sender, + finished: watch::Sender, + /// Created at ConnectRequest so a ReadyForRoomEventRequest on the early + /// `room_handle` can store a permit before `FfiRoom` exists. + room_event_ready_notify: Arc, +} + +impl FfiConnectingRoom { + pub fn cancel(&self) { + self.abort.send_replace(true); + self.room_event_ready_notify.notify_one(); + } + + pub fn is_cancelled(&self) -> bool { + *self.abort.borrow() + } + + pub fn finished_flag(&self) -> watch::Sender { + self.finished.clone() + } + + pub fn ready_for_room_event(&self) { + self.room_event_ready_notify.notify_one(); + } +} #[derive(Clone)] pub struct FfiRoom { @@ -60,6 +93,7 @@ pub struct FfiRoom { /// event-forwarding tasks once it fires, ensuring no room events are /// emitted before the client is ready to receive them. room_event_ready_notify: Arc, + abort: watch::Sender, } pub struct RoomInner { @@ -127,15 +161,102 @@ struct FfiSipDtmfPacket { async_id: u64, } +pub(crate) async fn wait_until_flag(flag: &watch::Sender) { + let mut rx = flag.subscribe(); + loop { + if *rx.borrow() { + return; + } + if rx.changed().await.is_err() { + return; + } + } +} + +fn abort_requested(abort: &watch::Sender) -> bool { + *abort.borrow() +} + +async fn send_connect_error(server: &'static FfiServer, async_id: u64, error: String) { + let _ = server.send_event( + proto::ConnectCallback { + async_id, + message: Some(proto::connect_callback::Message::Error(error)), + ..Default::default() + } + .into(), + ); +} + +enum ConnectOutcome { + Connected(Room, mpsc::UnboundedReceiver), + Failed(String), + Cancelled, +} + +/// rust-sdks#1334: aborting a JoinHandle without awaiting it drops a completed +/// `Room` without `close()`; `engine_task`/`room_task` keep ICE sockets alive +/// (~13 UDP/cycle). Always join after abort and close if connect already returned Ok. +async fn settle_aborted_connect( + connect_join: JoinHandle)>>, + user_aborted: bool, +) -> ConnectOutcome { + connect_join.abort(); + match connect_join.await { + Ok(Ok((room, events))) => { + let _ = room.close_with_reason(DisconnectReason::ClientInitiated.into()).await; + drop(events); + ConnectOutcome::Cancelled + } + Ok(Err(e)) => { + if user_aborted { + ConnectOutcome::Cancelled + } else { + ConnectOutcome::Failed(e.to_string()) + } + } + Err(join) if join.is_cancelled() => ConnectOutcome::Cancelled, + Err(join) if join.is_panic() => { + // python-sdks#785: surface as a connect error; do not send_panic. + ConnectOutcome::Failed("Room::connect panicked".into()) + } + Err(_) => ConnectOutcome::Cancelled, + } +} + +/// Marks the in-flight connect finished on every exit, including panic unwind. +/// `send_replace` (not `send`): `watch::Sender::send` no-ops if no receiver exists yet. +struct FinishOnDrop(watch::Sender); + +impl Drop for FinishOnDrop { + fn drop(&mut self) { + self.0.send_replace(true); + } +} + impl FfiRoom { pub fn connect( server: &'static FfiServer, connect: proto::ConnectRequest, ) -> proto::ConnectResponse { let async_id = server.resolve_async_id(connect.request_async_id); + let handle_id = server.next_id(); + let (abort_tx, _) = watch::channel(false); + let (finished_tx, _) = watch::channel(false); + let ready_notify = Arc::new(Notify::new()); + + server.store_handle( + handle_id, + FfiConnectingRoom { + handle_id, + abort: abort_tx.clone(), + finished: finished_tx.clone(), + room_event_ready_notify: ready_notify.clone(), + }, + ); let req = connect.clone(); - let mut options: RoomOptions = connect.options.into(); + let mut options: RoomOptions = connect.options.clone().into(); { let config = server.config.lock(); @@ -145,211 +266,311 @@ impl FfiRoom { } } - let connect = async move { - match Room::connect(&connect.url, &connect.token, options.clone()).await { - Ok((room, mut events)) => { - // initialize audio filters - let result = server - .async_runtime - .spawn_blocking(move || { - for filter in registered_audio_filter_plugins().into_iter() { - filter.on_load(&req.url, &req.token)?; - } - Ok::<(), PluginError>(()) - }) - .await; + let abort = abort_tx.clone(); + let connect_req = connect; + let connect_task = async move { + let _finish = FinishOnDrop(finished_tx); + let connect_url = connect_req.url.clone(); + let connect_token = connect_req.token.clone(); + let connect_options = options.clone(); + let mut connect_join = server.async_runtime.spawn(async move { + Room::connect(&connect_url, &connect_token, connect_options).await + }); + + let join_result = tokio::select! { + biased; + _ = wait_until_flag(&abort) => None, + join = &mut connect_join => Some(join), + }; + let outcome = match join_result { + None => settle_aborted_connect(connect_join, true).await, + Some(Ok(Ok((room, events)))) => ConnectOutcome::Connected(room, events), + Some(Ok(Err(e))) => ConnectOutcome::Failed(e.to_string()), + Some(Err(join)) if join.is_panic() => { + // python-sdks#785: do not send_panic from the connect path. + ConnectOutcome::Failed("Room::connect panicked".into()) + } + Some(Err(_)) => ConnectOutcome::Cancelled, + }; + + let (room, mut events) = match outcome { + ConnectOutcome::Cancelled => { + send_connect_error(server, async_id, "connect cancelled".into()).await; + server.drop_handle(handle_id); + return; + } + ConnectOutcome::Failed(e) => { + log::error!("error while connecting to a room: {}", e); + send_connect_error(server, async_id, e).await; + server.drop_handle(handle_id); + return; + } + ConnectOutcome::Connected(room, events) => (room, events), + }; + + if abort_requested(&abort) { + let _ = room.close_with_reason(DisconnectReason::ClientInitiated.into()).await; + send_connect_error(server, async_id, "connect cancelled".into()).await; + server.drop_handle(handle_id); + return; + } + + // initialize audio filters + let filter_req = req.clone(); + let filter_join = server.async_runtime.spawn_blocking(move || { + for filter in registered_audio_filter_plugins().into_iter() { + filter.on_load(&filter_req.url, &filter_req.token)?; + } + Ok::<(), PluginError>(()) + }); + + let result = tokio::select! { + biased; + _ = wait_until_flag(&abort) => None, + result = filter_join => Some(result), + }; + + match result { + None => { + // rust-sdks#1334: abort during filter wait must close the Room. + let _ = room.close_with_reason(DisconnectReason::ClientInitiated.into()).await; + send_connect_error(server, async_id, "connect cancelled".into()).await; + server.drop_handle(handle_id); + return; + } + Some(Ok(Ok(()))) => (), + Some(Ok(Err(e))) => { // Filter failures are non-fatal: keep the RTC session alive, just // without the filter enabled. - match result { - Ok(Ok(())) => (), - Ok(Err(e)) => { - let hint = match &e { - PluginError::OnLoad(_) => " — ensure you are connecting to LiveKit Cloud and that the filter is configured correctly", - PluginError::Library(_) => " — the filter dylib could not be loaded", - PluginError::NotImplemented(_) => " — the filter dylib is missing a required entry point", - }; - log::error!("audio filter disabled, continuing without it: {e}{hint}"); - } - Err(join_err) => { - log::error!("audio filter disabled, on_load task panicked: {join_err}"); - } + let hint = match &e { + PluginError::OnLoad(_) => " — ensure you are connecting to LiveKit Cloud and that the filter is configured correctly", + PluginError::Library(_) => " — the filter dylib could not be loaded", + PluginError::NotImplemented(_) => " — the filter dylib is missing a required entry point", }; + log::error!("audio filter disabled, continuing without it: {e}{hint}"); + } + Some(Err(join_err)) => { + log::error!("audio filter disabled, on_load task panicked: {join_err}"); + } + } - // Successfully connected to the room - // Forward the initial state for the FfiClient - let Some(RoomEvent::Connected { participants_with_tracks }) = - events.recv().await - else { - unreachable!("Connected event should always be the first event"); - }; + if abort_requested(&abort) { + let _ = room.close_with_reason(DisconnectReason::ClientInitiated.into()).await; + send_connect_error(server, async_id, "connect cancelled".into()).await; + server.drop_handle(handle_id); + return; + } - let (data_tx, data_rx) = mpsc::unbounded_channel(); - let (transcription_tx, transcription_rx) = mpsc::unbounded_channel(); - let (dtmf_tx, dtmf_rx) = mpsc::unbounded_channel(); - let (close_tx, close_rx) = broadcast::channel(1); + // Successfully connected to the room + // Forward the initial state for the FfiClient + let connected = tokio::select! { + biased; + _ = wait_until_flag(&abort) => None, + ev = events.recv() => ev, + }; + let Some(RoomEvent::Connected { participants_with_tracks }) = connected else { + if abort_requested(&abort) { + let _ = room.close_with_reason(DisconnectReason::ClientInitiated.into()).await; + send_connect_error(server, async_id, "connect cancelled".into()).await; + server.drop_handle(handle_id); + return; + } + // Do not unreachable!/watch_panic the host if the first event is wrong. + log::error!("first room event was not Connected"); + let _ = room.close_with_reason(DisconnectReason::ClientInitiated.into()).await; + send_connect_error(server, async_id, "first room event was not Connected".into()) + .await; + server.drop_handle(handle_id); + return; + }; - let handle_id = server.next_id(); - let inner = Arc::new(RoomInner { - room, - handle_id, - data_tx, - transcription_tx, - dtmf_tx, - pending_published_tracks: Default::default(), - pending_unpublished_tracks: Default::default(), - track_handle_lookup: Default::default(), - local_publication_lookup: Default::default(), - rpc_method_invocation_waiters: Default::default(), - url: connect.url, - }); - - let (local_info, remote_infos) = - build_initial_states(server, &inner, participants_with_tracks); - - // Send callback - let ffi_room = Self { - inner: inner.clone(), - handle: Default::default(), - room_event_ready_notify: Arc::new(Notify::new()), - }; - server.store_handle(ffi_room.inner.handle_id, ffi_room.clone()); - - // Keep the lock until the handle is "Some" (So it is OK for the client to - // request a disconnect quickly after connecting) - // (When requesting a disconnect, the handle will still be locked and the - // disconnect will wait for the lock to be released and gracefully close the - // room) - let mut handle = ffi_room.handle.lock().await; - let room_info = proto::RoomInfo::from(&ffi_room); - - // Send the async response to the FfiClient *before* starting the tasks. - // Ensure no events are sent before the callback - let _ = server.send_event( - proto::ConnectCallback { - async_id, - message: Some(proto::connect_callback::Message::Result( - proto::connect_callback::Result { - room: proto::OwnedRoom { - handle: proto::FfiOwnedHandle { id: handle_id }, - info: room_info, - }, - local_participant: local_info, - participants: remote_infos, - }, - )), - } - .into(), - ); + if abort_requested(&abort) { + let _ = room.close_with_reason(DisconnectReason::ClientInitiated.into()).await; + send_connect_error(server, async_id, "connect cancelled".into()).await; + server.drop_handle(handle_id); + return; + } - // Wait for the FFI client to install its event listener and - // send a ReadyForRoomEventRequest before forwarding any room - // events. This avoids a race where events emitted between - // the ConnectCallback and the listener registration are - // dropped. - if tokio::time::timeout( - ROOM_EVENT_READY_TIMEOUT, - ffi_room.room_event_ready_notify.notified(), - ) - .await - .is_err() - { - let msg = format!( - "timed out waiting for ReadyForRoomEventRequest after ConnectCallback \ - (room_handle={handle_id})" - ); - log::error!("{}", msg); - drop(handle); - ffi_room.close(server, DisconnectReason::ConnectionTimeout).await; - server.drop_handle(handle_id); - server.send_panic(Box::new(FfiError::InvalidRequest(msg.into()))); - return; - } + let (data_tx, data_rx) = mpsc::unbounded_channel(); + let (transcription_tx, transcription_rx) = mpsc::unbounded_channel(); + let (dtmf_tx, dtmf_rx) = mpsc::unbounded_channel(); + let (close_tx, close_rx) = broadcast::channel(1); + + let inner = Arc::new(RoomInner { + room, + handle_id, + data_tx, + transcription_tx, + dtmf_tx, + pending_published_tracks: Default::default(), + pending_unpublished_tracks: Default::default(), + track_handle_lookup: Default::default(), + local_publication_lookup: Default::default(), + rpc_method_invocation_waiters: Default::default(), + url: connect_req.url, + }); + + let (local_info, remote_infos) = + build_initial_states(server, &inner, participants_with_tracks); - // Update Room SID on promise resolve. Spawned after the - // ready handshake so the RoomSidChanged event is never - // delivered before the client is ready to receive it. - let room_handle = inner.handle_id.clone(); - server.async_runtime.spawn(async move { - let _ = server.send_event( - proto::RoomEvent { - room_handle, - message: Some( - proto::RoomSidChanged { - sid: ffi_room.inner.room.sid().await.into(), - } - .into(), - ), - } - .into(), - ); - }); - - // Forward events - let event_handle = server.watch_panic({ - let close_rx = close_rx.resubscribe(); - server.async_runtime.spawn(room_task( - server, - inner.clone(), - events, - close_rx, - )) - }); - - let data_handle = server.watch_panic({ - let close_rx = close_rx.resubscribe(); - server.async_runtime.spawn(data_task( - server, - inner.clone(), - data_rx, - close_rx, - )) - }); // Publish data - - let transcription_handle = server.watch_panic({ - let close_rx = close_rx.resubscribe(); - server.async_runtime.spawn(transcription_task( - server, - inner.clone(), - transcription_rx, - close_rx, - )) - }); // Publish transcription - - let sip_dtmf_handle = - server.watch_panic(server.async_runtime.spawn(sip_dtmf_task( - server, - inner.clone(), - dtmf_rx, - close_rx, - ))); - - *handle = Some(Handle { - event_handle, - data_handle, - transcription_handle, - sip_dtmf_handle, - close_tx, - }); + let ffi_room = Self { + inner: inner.clone(), + handle: Default::default(), + room_event_ready_notify: ready_notify, + abort: abort.clone(), + }; + + // Overwrite the connecting handle in place. take+store left the id + // missing for one DashMap beat; DisconnectRequest then 404'd. + server.store_handle(handle_id, ffi_room.clone()); + if abort_requested(&abort) { + // Client never received ConnectCallback.Result, so it does not + // own these ids. Drop them or the handle table leaks. + drop_initial_state_handles(server, &local_info, &remote_infos); + ffi_room.close(server, DisconnectReason::ClientInitiated).await; + send_connect_error(server, async_id, "connect cancelled".into()).await; + server.drop_handle(handle_id); + return; + } + + // Keep the lock until the handle is "Some" (So it is OK for the client to + // request a disconnect quickly after connecting) + // (When requesting a disconnect, the handle will still be locked and the + // disconnect will wait for the lock to be released and gracefully close the + // room) + let mut handle = ffi_room.handle.lock().await; + let room_info = proto::RoomInfo::from(&ffi_room); + + // Send the async response to the FfiClient *before* starting the tasks. + // Ensure no events are sent before the callback + let _ = server.send_event( + proto::ConnectCallback { + async_id, + message: Some(proto::connect_callback::Message::Result( + proto::connect_callback::Result { + room: proto::OwnedRoom { + handle: proto::FfiOwnedHandle { id: handle_id }, + info: room_info, + }, + local_participant: local_info, + participants: remote_infos, + }, + )), } - Err(e) => { - // Failed to connect to the room, send an error message to the FfiClient - // TODO(theomonnom): Typed errors? - log::error!("error while connecting to a room: {}", e); - let _ = server.send_event( - proto::ConnectCallback { - async_id, - message: Some(proto::connect_callback::Message::Error(e.to_string())), - ..Default::default() - } - .into(), + .into(), + ); + + // Wait for the FFI client to install its event listener and + // send a ReadyForRoomEventRequest before forwarding any room + // events. This avoids a race where events emitted between + // the ConnectCallback and the listener registration are + // dropped. + // + // A missed ReadyFor handshake fails this room only. It must + // never panic the host process. Disconnect/cancel unblocks + // this wait immediately. + let ready = tokio::select! { + biased; + _ = wait_until_flag(&abort) => Err(()), + ready = tokio::time::timeout( + ROOM_EVENT_READY_TIMEOUT, + ffi_room.room_event_ready_notify.notified(), + ) => ready.map_err(|_| ()), + }; + + if ready.is_err() { + if abort_requested(&abort) { + log::info!( + "connect aborted while waiting for ReadyForRoomEventRequest \ + (room_handle={handle_id})" + ); + } else { + // python-sdks#785: missed ReadyFor is not a panic. Result was + // already sent; close the room and return. Do not send_panic + // and do not send a second ConnectCallback. + log::error!( + "timed out waiting for ReadyForRoomEventRequest after ConnectCallback \ + (room_handle={handle_id})" ); } - }; + drop(handle); + // Result was already delivered. End the event stream so the + // language object does not sit on a room that will never emit. + let _ = server.send_event( + proto::RoomEvent { + room_handle: handle_id, + message: Some(proto::RoomEos {}.into()), + } + .into(), + ); + ffi_room.close(server, DisconnectReason::ConnectionTimeout).await; + server.drop_handle(handle_id); + return; + } + + // Update Room SID on promise resolve. Spawned after the + // ready handshake so the RoomSidChanged event is never + // delivered before the client is ready to receive it. + let room_handle = inner.handle_id; + let sid_room = ffi_room.clone(); + server.async_runtime.spawn(async move { + let _ = server.send_event( + proto::RoomEvent { + room_handle, + message: Some( + proto::RoomSidChanged { sid: sid_room.inner.room.sid().await.into() } + .into(), + ), + } + .into(), + ); + }); + + // Forward events + let event_handle = server.watch_panic({ + let close_rx = close_rx.resubscribe(); + server.async_runtime.spawn(room_task(server, inner.clone(), events, close_rx)) + }); + + let data_handle = server.watch_panic({ + let close_rx = close_rx.resubscribe(); + server.async_runtime.spawn(data_task(server, inner.clone(), data_rx, close_rx)) + }); // Publish data + + let transcription_handle = server.watch_panic({ + let close_rx = close_rx.resubscribe(); + server.async_runtime.spawn(transcription_task( + server, + inner.clone(), + transcription_rx, + close_rx, + )) + }); // Publish transcription + + let sip_dtmf_handle = server.watch_panic(server.async_runtime.spawn(sip_dtmf_task( + server, + inner.clone(), + dtmf_rx, + close_rx, + ))); + + *handle = Some(Handle { + event_handle, + data_handle, + transcription_handle, + sip_dtmf_handle, + close_tx, + }); }; - server.watch_panic(server.async_runtime.spawn(connect)); - proto::ConnectResponse { async_id } + server.watch_panic(server.async_runtime.spawn(connect_task)); + proto::ConnectResponse { async_id, room_handle: Some(handle_id), ..Default::default() } + } + + pub fn cancel(&self) { + self.abort.send_replace(true); + self.room_event_ready_notify.notify_one(); } /// Release the connect task's wait point so room event forwarding can @@ -361,6 +582,9 @@ impl FfiRoom { /// Close the room and stop the tasks pub async fn close(&self, server: &'static FfiServer, reason: DisconnectReason) { + let _ = self.abort.send_replace(true); + self.room_event_ready_notify.notify_one(); + // drop associated track handles for (_, &handle) in self.inner.track_handle_lookup.lock().iter() { if server.drop_handle(handle) { @@ -1592,6 +1816,20 @@ async fn forward_event( }; } +fn drop_initial_state_handles( + server: &'static FfiServer, + local: &proto::OwnedParticipant, + remotes: &[proto::connect_callback::ParticipantWithTracks], +) { + let _ = server.drop_handle(local.handle.id); + for remote in remotes { + let _ = server.drop_handle(remote.participant.handle.id); + for publication in &remote.publications { + let _ = server.drop_handle(publication.handle.id); + } + } +} + fn build_initial_states( server: &'static FfiServer, inner: &Arc,