diff --git a/docs/README.md b/docs/README.md index 1a4db0c5..3591df0a 100644 --- a/docs/README.md +++ b/docs/README.md @@ -6,6 +6,8 @@ Additional documentation for the SDK. vcpkg, Docker, integration into your CMake project, troubleshooting. - [Authentication](authentication.md) — generating tokens, token source types (initial connect only), and in-session token refresh. +- [Reconnection](reconnection.md) — automatic resume/full reconnect behavior, + callback ordering and threading, object continuity, and application patterns. - [Logging](logging.md) — compile-time vs runtime filtering, log levels, custom sinks (file, JSON, ROS2 `RCLCPP_*` macros). - [Tracing](tracing.md) — Chromium-format performance traces, viewing in diff --git a/docs/authentication.md b/docs/authentication.md index f048aa04..b0fc79fe 100644 --- a/docs/authentication.md +++ b/docs/authentication.md @@ -19,6 +19,9 @@ These are separate mechanisms: the Rust core owns the session JWT. The SDK applies server-pushed token updates internally (for reconnect). Your token endpoint is **not** called again unless **you** invoke `Room::connect` again (for example after a full disconnect). +Automatic resume and full reconnect are both part of the same active room +session; neither invokes a token source. See [Reconnection](reconnection.md) for +the complete lifecycle and terminal-recovery guidance. If you need the latest JWT after a server refresh, implement `RoomDelegate::onTokenRefreshed` and store `TokenRefreshedEvent::token` in your diff --git a/docs/reconnection.md b/docs/reconnection.md new file mode 100644 index 00000000..93d6cf3e --- /dev/null +++ b/docs/reconnection.md @@ -0,0 +1,142 @@ +# Reconnection + +The C++ SDK follows the same reconnect contract as the other LiveKit client +SDKs. After an established room loses its signal or media transport, the +embedded Rust core automatically tries to recover the existing room. The +application observes that process through `RoomDelegate`; it must not start a +second retry loop or call `Room::connect()` while recovery is active. + +## Lifecycle + +A reconnect episode has one of two outcomes: + +1. `Room::connectionState()` changes to `ConnectionState::Reconnecting`. + `onConnectionStateChanged` and then `onReconnecting` report that transition. +2. The Rust core first attempts a lightweight resume of the existing session. +3. If resume cannot succeed, the core escalates to a full reconnect, rebuilds + remote state, and republishes local tracks. +4. Success changes the state to `ConnectionState::Connected`. + `onConnectionStateChanged` and then `onReconnected` run after reconciliation + is complete. +5. Exhausted recovery changes the state to `ConnectionState::Disconnected`. + `onConnectionStateChanged` and then `onDisconnected` report the terminal + outcome exactly once. + +Calling `Room::disconnect()` is also terminal and produces at most one +`onDisconnected` callback. A server-initiated terminal disconnect is followed +by `onRoomEos`, which completes local cleanup. Do not call `connect()` again +until `onRoomEos` has run; alternatively, call `disconnect()` from an +application thread to finish pending cleanup first. + +`RoomOptions::join_retries` controls retries for the initial join. It does not +change the reconnect policy for an established room. Reconnect mode, reason, +attempt count, elapsed time, and retry policy are not currently exposed by the +C++ FFI. + +## Callback threading + +Server-driven room delegate callbacks run on the Rust-managed FFI callback +thread. The callback generated by an explicit `Room::disconnect()` runs on the +calling thread. In either case, a callback must not block, wait for a future, +reconnect the room, or do substantial work. Post a small event to the +application's executor, event loop, or worker queue instead: + +```cpp +class ConnectionDelegate : public livekit::RoomDelegate { +public: + explicit ConnectionDelegate(Executor& executor) : executor_(executor) {} + + void onReconnecting(livekit::Room&, const livekit::ReconnectingEvent&) override { + executor_.post([] { showConnectionBanner("Reconnecting"); }); + } + + void onReconnected(livekit::Room&, const livekit::ReconnectedEvent&) override { + executor_.post([] { hideConnectionBanner(); }); + } + + void onDisconnected(livekit::Room&, const livekit::DisconnectedEvent& event) override { + executor_.post([reason = event.reason] { handleTerminalDisconnect(reason); }); + } + +private: + Executor& executor_; +}; +``` + +The exact executor type belongs to the application. It may be a UI event loop, +an `asio` executor, a service work queue, or another non-FFI thread. + +## Object and registration continuity + +A successful resume preserves the current participant, publication, and track +objects. + +A full reconnect republishes local audio and video tracks. Existing `Track` and +`LocalTrackPublication` objects remain valid and are updated in place before +`onReconnected`. Their SIDs can change, so code should read the current SID from +the object after reconnect instead of retaining a SID string indefinitely. +`LocalParticipant::trackPublications()` is rekeyed to the current SID, and the +current SID can be used with `unpublishTrack`. + +`onLocalTrackRepublished` runs once for each successfully republished local +audio or video track. The event contains the existing publication and track, +the previous SID, and the local participant. Use it to rekey external state +before `onReconnected` reports that reconciliation is complete. + +Publication and local-track properties are updated on the FFI callback thread +and must not be read concurrently from another thread during that update. +Reading them inside `onLocalTrackRepublished` is safe. To consume the refreshed +values elsewhere, copy the values into an application event and transfer that +event through a mutex, thread-safe queue, or executor so the receiving thread +is synchronized with the callback. + +Audio, video, and data callback registrations on `Room` remain installed during +recovery. Reader threads can stop and restart as track events rebuild the +subscription. Register handlers once for the room lifetime; register them again +only after tearing down the room and creating a new connection. + +One FFI limitation remains for data tracks: `LocalDataTrack::info().sid` and +remote data-track SID snapshots can become stale after a full reconnect. +Handle-based operations such as `tryPush`, subscribe, and unpublish continue to +use the current Rust object. Refreshing those snapshots requires a future +Rust/FFI event or query. + +## Operations while reconnecting + +Treat `Reconnecting` as a temporary loss of transport: + +- Pause nonessential media or control operations, or tolerate a transient + operation failure and retry after `onReconnected`. +- Buffer application data only when its ordering, memory bound, and expiry + behavior are explicit. The SDK does not provide an unbounded application + queue. +- Keep application state associated with participant, track, and publication + objects. Do not discard them merely because reconnect started. +- Do not fetch new credentials or call `connect()` during an active episode. + The Rust core uses the latest server-refreshed session token. + +After terminal `onDisconnected`, wait for `onRoomEos` (or explicitly finish +cleanup with `disconnect()`), obtain fresh credentials if needed, and perform +the next `connect()` from an application thread. + +## Application patterns + +Interactive applications commonly disable controls that require the transport, +show a reconnect indicator, and restore controls from `onReconnected`. A +terminal disconnect returns the UI to its join flow. + +Headless services commonly publish lifecycle events into their main work queue +and mark the service degraded while reconnecting. A watchdog may impose an +application-specific availability bound and replace the room after a terminal +disconnect. Its timeout is an operational policy, not the Rust reconnect +deadline. + +Publisher/subscriber processes should keep sources, tracks, publications, and +callback registrations alive through recovery. After `onReconnected`, they can +read current publication SIDs and resume any application-level production that +was paused. + +A ROS bridge is one example of the headless-service pattern: delegate callbacks +enqueue state changes for a ROS executor, while a watchdog decides when the +larger node should be restarted. The same contract applies to conferencing, +agents, streaming, robotics, and embedded applications. diff --git a/include/livekit/local_participant.h b/include/livekit/local_participant.h index 9369a914..6530a076 100644 --- a/include/livekit/local_participant.h +++ b/include/livekit/local_participant.h @@ -42,6 +42,9 @@ struct ParticipantTrackPermission; class FfiClient; class Track; class LocalTrackPublication; +namespace proto { +class TrackPublicationInfo; +} /// @brief Data passed to a registered RPC method handler. struct RpcInvocationData { @@ -241,10 +244,15 @@ class LIVEKIT_API LocalParticipant : public Participant { friend class Room; private: + bool handleTrackRepublished(const std::string& previous_sid, uintptr_t publication_handle, + const proto::TrackPublicationInfo& info, + std::shared_ptr& publication, std::shared_ptr& track); + /// Publication SID → local track (@ref unpublishTrack clears the track’s /// cached publication). @c mutable so @ref trackPublications() const can /// prune expired @c weak_ptr entries. mutable TrackMap published_tracks_by_sid_; + mutable std::mutex publication_mutex_; std::unordered_map rpc_handlers_; diff --git a/include/livekit/local_track_publication.h b/include/livekit/local_track_publication.h index 412b4fb9..2a17f5a3 100644 --- a/include/livekit/local_track_publication.h +++ b/include/livekit/local_track_publication.h @@ -23,7 +23,10 @@ namespace livekit { namespace proto { class OwnedTrackPublication; -} +class TrackPublicationInfo; +} // namespace proto + +class LocalParticipant; /// @brief A track published by a local participant. class LIVEKIT_API LocalTrackPublication : public TrackPublication { @@ -31,6 +34,11 @@ class LIVEKIT_API LocalTrackPublication : public TrackPublication { /// Note, this LocalTrackPublication is constructed internally only; /// safe to accept proto::OwnedTrackPublication. explicit LocalTrackPublication(const proto::OwnedTrackPublication& owned); + +private: + friend class LocalParticipant; + + void updateFromProto(uintptr_t publication_handle, const proto::TrackPublicationInfo& info); }; } // namespace livekit diff --git a/include/livekit/room.h b/include/livekit/room.h index caf4905b..9a48b869 100644 --- a/include/livekit/room.h +++ b/include/livekit/room.h @@ -21,6 +21,7 @@ #include #include #include +#include #include "livekit/data_stream.h" #include "livekit/e2ee.h" @@ -334,6 +335,8 @@ class LIVEKIT_API Room { mutable std::mutex lock_; ConnectionState connection_state_ = ConnectionState::Disconnected; + std::optional last_notified_connection_state_; + bool disconnect_callback_claimed_ = true; RoomDelegate* delegate_ = nullptr; // Not owned RoomInfoData room_info_; std::shared_ptr room_handle_; @@ -355,6 +358,7 @@ class LIVEKIT_API Room { int listener_id_{0}; void onEvent(const proto::FfiEvent& event); + void transitionConnectionState(ConnectionState state, bool notify_delegate); // Shared shutdown path for explicit disconnect, server disconnect, EOS, and destruction. bool shutdown(bool disconnect_ffi, DisconnectReason reason, bool notify_delegate); diff --git a/include/livekit/room_delegate.h b/include/livekit/room_delegate.h index 4dc790e7..c6c453b1 100644 --- a/include/livekit/room_delegate.h +++ b/include/livekit/room_delegate.h @@ -54,6 +54,12 @@ class LIVEKIT_API RoomDelegate { /// Called when a local track is unpublished. virtual void onLocalTrackUnpublished(Room&, const LocalTrackUnpublishedEvent&) {} + /// @brief Called when a local track is republished during a full reconnect. + /// + /// The publication and track have already been updated and rekeyed under the + /// current SID. This callback runs before @ref onReconnected. + virtual void onLocalTrackRepublished(Room&, const LocalTrackRepublishedEvent&) {} + /// Called when a local track gains its first subscriber. virtual void onLocalTrackSubscribed(Room&, const LocalTrackSubscribedEvent&) {} diff --git a/include/livekit/room_event_types.h b/include/livekit/room_event_types.h index 86589681..0975e8df 100644 --- a/include/livekit/room_event_types.h +++ b/include/livekit/room_event_types.h @@ -28,6 +28,7 @@ namespace livekit { // Forward declarations to avoid pulling in heavy headers. class Track; class Participant; +class LocalParticipant; class RemoteParticipant; class RemoteDataTrack; class LocalTrackPublication; @@ -395,6 +396,25 @@ struct LocalTrackUnpublishedEvent { std::shared_ptr publication; }; +/// @brief Fired when the SDK republishes a local track during a full reconnect. +/// +/// The existing publication and track objects are updated in place before this +/// event is delivered. Use @ref previous_sid to reconcile application state +/// keyed by the former server-assigned SID. +struct LocalTrackRepublishedEvent { + /// Existing publication updated with its current SID and metadata. + std::shared_ptr publication; + + /// Existing local track associated with the publication. + std::shared_ptr track; + + /// Local participant that owns the publication (owned by Room). + LocalParticipant* participant = nullptr; + + /// Server-assigned publication SID used before the full reconnect. + std::string previous_sid; +}; + /// Fired when a local track gets its first subscriber. struct LocalTrackSubscribedEvent { /// Subscribed local track. diff --git a/include/livekit/track.h b/include/livekit/track.h index 5fed6fe6..ce845ee4 100644 --- a/include/livekit/track.h +++ b/include/livekit/track.h @@ -31,6 +31,7 @@ namespace livekit { class LocalTrackPublication; +class LocalParticipant; /// @brief Media kind for an audio or video track. enum class TrackKind { @@ -74,6 +75,11 @@ struct ParticipantTrackPermission { }; /// @brief Base class for local and remote media tracks. +/// +/// @warning Local-track properties are not safe to read concurrently with a +/// full reconnect update. Read the refreshed SID from +/// RoomDelegate::onLocalTrackRepublished or after synchronizing application +/// state from RoomDelegate::onReconnected. class LIVEKIT_API Track { public: virtual ~Track() = default; @@ -121,6 +127,10 @@ class LIVEKIT_API Track { std::optional mime_type); private: + friend class LocalParticipant; + + void setSid(const std::string& sid) { sid_ = sid; } + FfiHandle handle_; // Owned std::string sid_; diff --git a/include/livekit/track_publication.h b/include/livekit/track_publication.h index c618cd7d..e12a5aa1 100644 --- a/include/livekit/track_publication.h +++ b/include/livekit/track_publication.h @@ -34,8 +34,14 @@ class RemoteTrack; /// @brief Base class for a track that has been published to a room. /// -/// Wraps the immutable publication info plus an FFI handle, and -/// holds a weak reference to the associated Track (if any). +/// Wraps publication info plus an FFI handle and holds a reference to the +/// associated Track, if any. A local publication's SID and metadata can change +/// in place during a full reconnect. +/// +/// @warning Publication properties are not safe to read concurrently with a +/// reconnect update. Read the refreshed properties from +/// RoomDelegate::onLocalTrackRepublished or after synchronizing application +/// state from RoomDelegate::onReconnected. class LIVEKIT_API TrackPublication { public: virtual ~TrackPublication() = default; diff --git a/scripts/generate-docs.sh b/scripts/generate-docs.sh index 60ba5588..59f34059 100755 --- a/scripts/generate-docs.sh +++ b/scripts/generate-docs.sh @@ -154,7 +154,7 @@ cp "$repo_root/docs/doxygen/Doxyfile" "$doxyfile_tmp" { echo "" echo "# Local override generated by scripts/generate-docs.sh" - echo "INPUT = include/livekit ${mainpage_rel} docs/README.md docs/building.md docs/authentication.md docs/logging.md docs/tracing.md docs/testing.md docs/tools.md" + echo "INPUT = include/livekit ${mainpage_rel} docs/README.md docs/building.md docs/authentication.md docs/reconnection.md docs/logging.md docs/tracing.md docs/testing.md docs/tools.md" echo "USE_MDFILE_AS_MAINPAGE = ${mainpage_rel}" } >>"$doxyfile_tmp" diff --git a/src/ffi_client.cpp b/src/ffi_client.cpp index 89d6b0a8..8d2995df 100644 --- a/src/ffi_client.cpp +++ b/src/ffi_client.cpp @@ -71,6 +71,8 @@ std::optional ExtractAsyncId(const proto::FfiEvent& event) { return event.connect().async_id(); case E::kDisconnect: return event.disconnect().async_id(); + case E::kSimulateScenario: + return event.simulate_scenario().async_id(); case E::kDispose: return event.dispose().async_id(); case E::kPublishTrack: @@ -540,6 +542,42 @@ std::future FfiClient::disconnectAsync(uintptr_t room_handle, DisconnectRe return fut; } +std::future FfiClient::simulateScenarioAsync(uintptr_t room_handle, proto::SimulateScenarioKind scenario) { + const AsyncId async_id = generateAsyncId(); + + auto fut = registerAsync( + async_id, + [async_id](const proto::FfiEvent& event) { + return event.has_simulate_scenario() && event.simulate_scenario().async_id() == async_id; + }, + [](const proto::FfiEvent& event, std::promise& pr) { + const auto& callback = event.simulate_scenario(); + if (callback.has_error() && !callback.error().empty()) { + pr.set_exception(std::make_exception_ptr(std::runtime_error(callback.error()))); + return; + } + pr.set_value(); + }); + + proto::FfiRequest req; + auto* simulate = req.mutable_simulate_scenario(); + simulate->set_room_handle(room_handle); + simulate->set_scenario(scenario); + simulate->set_request_async_id(async_id); + + try { + const proto::FfiResponse resp = sendRequest(req); + if (!resp.has_simulate_scenario()) { + logAndThrow("FfiResponse missing simulate_scenario"); + } + } catch (...) { + cancelPendingByAsyncId(async_id); + throw; + } + + return fut; +} + // Track APIs Implementation std::future> FfiClient::getTrackStatsAsync(uintptr_t track_handle) { // Generate client-side async_id first diff --git a/src/ffi_client.h b/src/ffi_client.h index 9dc840e4..cdcb1a59 100644 --- a/src/ffi_client.h +++ b/src/ffi_client.h @@ -98,6 +98,7 @@ class LIVEKIT_INTERNAL_API FfiClient { const RoomOptions& options); std::future disconnectAsync(uintptr_t room_handle, DisconnectReason reason); + std::future simulateScenarioAsync(uintptr_t room_handle, proto::SimulateScenarioKind scenario); // Track APIs std::future> getTrackStatsAsync(uintptr_t track_handle); diff --git a/src/local_participant.cpp b/src/local_participant.cpp index 7fda68ac..bd806517 100644 --- a/src/local_participant.cpp +++ b/src/local_participant.cpp @@ -197,9 +197,12 @@ void LocalParticipant::publishTrack(const std::shared_ptr& track, const T auto publication = std::make_shared(owned_pub); const std::string sid = publication->sid(); - published_tracks_by_sid_[sid] = std::weak_ptr(track); - + track->setSid(sid); track->setPublication(publication); + { + const std::scoped_lock guard(publication_mutex_); + published_tracks_by_sid_[sid] = std::weak_ptr(track); + } } std::shared_ptr LocalParticipant::publishVideoTrack(const std::string& name, @@ -237,15 +240,21 @@ void LocalParticipant::unpublishTrack(const std::string& track_sid) { fut.get(); - if (auto it = published_tracks_by_sid_.find(track_sid); it != published_tracks_by_sid_.end()) { - if (auto t = it->second.lock()) { - t->setPublication(nullptr); + std::shared_ptr unpublished_track; + { + const std::scoped_lock guard(publication_mutex_); + if (auto it = published_tracks_by_sid_.find(track_sid); it != published_tracks_by_sid_.end()) { + unpublished_track = it->second.lock(); + published_tracks_by_sid_.erase(it); } - published_tracks_by_sid_.erase(it); + } + if (unpublished_track) { + unpublished_track->setPublication(nullptr); } } LocalParticipant::PublicationMap LocalParticipant::trackPublications() const { + const std::scoped_lock guard(publication_mutex_); PublicationMap out; for (auto it = published_tracks_by_sid_.begin(); it != published_tracks_by_sid_.end();) { auto t = it->second.lock(); @@ -261,6 +270,36 @@ LocalParticipant::PublicationMap LocalParticipant::trackPublications() const { return out; } +bool LocalParticipant::handleTrackRepublished(const std::string& previous_sid, uintptr_t publication_handle, + const proto::TrackPublicationInfo& info, + std::shared_ptr& publication, + std::shared_ptr& track) { + const std::scoped_lock guard(publication_mutex_); + const auto it = published_tracks_by_sid_.find(previous_sid); + if (it == published_tracks_by_sid_.end()) { + return false; + } + + track = it->second.lock(); + if (!track) { + published_tracks_by_sid_.erase(it); + return false; + } + + publication = localTrackPublication(track); + if (!publication) { + track.reset(); + return false; + } + + publication->updateFromProto(publication_handle, info); + track->setSid(info.sid()); + + published_tracks_by_sid_.erase(it); + published_tracks_by_sid_[info.sid()] = track; + return true; +} + Result, PublishDataTrackError> LocalParticipant::publishDataTrack( const std::string& name) { auto handle_id = ffiHandleId(); @@ -443,6 +482,7 @@ void LocalParticipant::handleRpcMethodInvocation(uint64_t invocation_id, const s } std::shared_ptr LocalParticipant::findTrackPublication(const std::string& sid) const { + const std::scoped_lock guard(publication_mutex_); auto it = published_tracks_by_sid_.find(sid); if (it == published_tracks_by_sid_.end()) { return nullptr; diff --git a/src/local_track_publication.cpp b/src/local_track_publication.cpp index efe6fea9..8406dfa6 100644 --- a/src/local_track_publication.cpp +++ b/src/local_track_publication.cpp @@ -27,4 +27,21 @@ LocalTrackPublication::LocalTrackPublication(const proto::OwnedTrackPublication& static_cast(owned.info().encryption_type()), convertAudioFeatures(owned.info().audio_features())) {} +void LocalTrackPublication::updateFromProto(uintptr_t publication_handle, const proto::TrackPublicationInfo& info) { + if (handle_.get() != publication_handle) { + handle_.reset(publication_handle); + } + sid_ = info.sid(); + name_ = info.name(); + kind_ = fromProto(info.kind()); + source_ = fromProto(info.source()); + simulcasted_ = info.simulcasted(); + width_ = info.width(); + height_ = info.height(); + mime_type_ = info.mime_type(); + muted_ = info.muted(); + encryption_type_ = static_cast(info.encryption_type()); + audio_features_ = convertAudioFeatures(info.audio_features()); +} + } // namespace livekit diff --git a/src/room.cpp b/src/room.cpp index 8277cf36..07854f2e 100644 --- a/src/room.cpp +++ b/src/room.cpp @@ -99,10 +99,14 @@ bool Room::connect(const std::string& url, const std::string& token, const RoomO { const std::scoped_lock g(lock_); - if (connection_state_ != ConnectionState::Disconnected) { - throw std::runtime_error("already connected"); + const bool cleanup_pending = + listener_id_ != 0 || room_handle_ || local_participant_ || !remote_participants_.empty(); + if (connection_state_ != ConnectionState::Disconnected || cleanup_pending) { + throw std::runtime_error("room is connected or cleanup is pending"); } connection_state_ = ConnectionState::Reconnecting; + last_notified_connection_state_.reset(); + disconnect_callback_claimed_ = false; } FfiClient::ListenerId listenerId = 0; @@ -178,6 +182,7 @@ bool Room::connect(const std::string& url, const std::string& token, const RoomO remote_participants_ = std::move(new_remote_participants); e2ee_manager_ = std::move(new_e2ee_manager); connection_state_ = ConnectionState::Connected; + disconnect_callback_claimed_ = false; } readyForRoomEvent(room_handle_id); @@ -188,6 +193,7 @@ bool Room::connect(const std::string& url, const std::string& token, const RoomO { const std::scoped_lock g(lock_); connection_state_ = ConnectionState::Disconnected; + disconnect_callback_claimed_ = true; if (listener_id_ == listenerId) { listener_to_remove = listener_id_; listener_id_ = 0; @@ -235,10 +241,13 @@ bool Room::shutdown(bool disconnect_ffi, DisconnectReason reason, bool notify_de if (!has_room_state) { return false; } - // The state transition determines which racing path owns the FFI request - // and delegate notification. Remaining room state is still claimed here so - // EOS or destruction can finish local cleanup after a server disconnect. - claimed_disconnect = connection_state_ != ConnectionState::Disconnected; + // State-change and Disconnected events are separate FFI events. Track + // notification ownership independently from connection_state_ so processing + // ConnectionStateChanged(Disconnected) cannot suppress onDisconnected. + claimed_disconnect = !disconnect_callback_claimed_; + if (claimed_disconnect) { + disconnect_callback_claimed_ = true; + } handle = std::move(room_handle_); delegate_snapshot = delegate_; local_participant_to_cleanup = std::move(local_participant_); @@ -308,6 +317,8 @@ bool Room::shutdown(bool disconnect_ffi, DisconnectReason reason, bool notify_de handle.reset(); if (notify_delegate && claimed_disconnect && delegate_snapshot) { + transitionConnectionState(ConnectionState::Disconnected, true); + DisconnectedEvent ev; ev.reason = reason; try { @@ -353,6 +364,26 @@ ConnectionState Room::connectionState() const { return connection_state_; } +void Room::transitionConnectionState(ConnectionState state, bool notify_delegate) { + RoomDelegate* delegate_snapshot = nullptr; + bool should_notify = false; + { + const std::scoped_lock guard(lock_); + connection_state_ = state; + if (notify_delegate && (!last_notified_connection_state_ || *last_notified_connection_state_ != state)) { + last_notified_connection_state_ = state; + delegate_snapshot = delegate_; + should_notify = delegate_snapshot != nullptr; + } + } + + if (should_notify) { + ConnectionStateChangedEvent ev; + ev.state = state; + delegate_snapshot->onConnectionStateChanged(*this, ev); + } +} + std::future Room::getStats() const { std::shared_ptr handle; { @@ -592,6 +623,32 @@ void Room::onEvent(const FfiEvent& event) { } break; } + case proto::RoomEvent::kLocalTrackRepublished: { + bool updated = false; + LocalTrackRepublishedEvent ev; + std::string new_sid; + { + const std::scoped_lock guard(lock_); + if (!local_participant_) { + LK_LOG_ERROR("kLocalTrackRepublished: local_participant_ is nullptr"); + break; + } + const auto& republished = re.local_track_republished(); + ev.previous_sid = republished.previous_sid(); + new_sid = republished.info().sid(); + updated = local_participant_->handleTrackRepublished( + ev.previous_sid, static_cast(republished.publication_handle()), republished.info(), + ev.publication, ev.track); + ev.participant = local_participant_.get(); + } + if (!updated) { + LK_LOG_WARN("local_track_republished for unknown publication sid {} (new sid {})", ev.previous_sid, + new_sid); + } else if (delegate_snapshot) { + delegate_snapshot->onLocalTrackRepublished(*this, ev); + } + break; + } case proto::RoomEvent::kLocalTrackSubscribed: { LocalTrackSubscribedEvent ev; { @@ -1181,32 +1238,19 @@ void Room::onEvent(const FfiEvent& event) { // ------------------------------------------------------------------------ case proto::RoomEvent::kConnectionStateChanged: { - ConnectionStateChangedEvent ev; - { - const std::scoped_lock guard(lock_); - const auto& cs = re.connection_state_changed(); - // TODO, maybe we should update our |connection_state_| - // correspoindingly, but the this kConnectionStateChanged event is never - // triggered in my local test. - LK_LOG_DEBUG("cs.state() is {} connection_state_ is {}", static_cast(cs.state()), - static_cast(connection_state_)); - ev.state = static_cast(cs.state()); - } - if (delegate_snapshot) { - delegate_snapshot->onConnectionStateChanged(*this, ev); - } + transitionConnectionState(toConnectionState(re.connection_state_changed().state()), true); break; } case proto::RoomEvent::kDisconnected: { bool should_notify = false; { const std::scoped_lock guard(lock_); - // Local shutdown marks the state before awaiting the FFI response - // and notifies the delegate itself. Suppress that duplicate while - // passing server-initiated disconnects through unchanged. - should_notify = connection_state_ != ConnectionState::Disconnected; - connection_state_ = ConnectionState::Disconnected; + should_notify = !disconnect_callback_claimed_; + if (should_notify) { + disconnect_callback_claimed_ = true; + } } + transitionConnectionState(ConnectionState::Disconnected, true); if (should_notify && delegate_snapshot) { DisconnectedEvent ev; ev.reason = toDisconnectReason(re.disconnected().reason()); @@ -1215,6 +1259,7 @@ void Room::onEvent(const FfiEvent& event) { break; } case proto::RoomEvent::kReconnecting: { + transitionConnectionState(ConnectionState::Reconnecting, true); const ReconnectingEvent ev; if (delegate_snapshot) { delegate_snapshot->onReconnecting(*this, ev); @@ -1222,6 +1267,7 @@ void Room::onEvent(const FfiEvent& event) { break; } case proto::RoomEvent::kReconnected: { + transitionConnectionState(ConnectionState::Connected, true); const ReconnectedEvent ev; if (delegate_snapshot) { delegate_snapshot->onReconnected(*this, ev); diff --git a/src/tests/integration/test_room.cpp b/src/tests/integration/test_room.cpp index 942d8939..d2d99b61 100644 --- a/src/tests/integration/test_room.cpp +++ b/src/tests/integration/test_room.cpp @@ -27,9 +27,12 @@ #include #include #include +#include #include #include "../common/test_common.h" +#include "ffi_client.h" +#include "room.pb.h" using namespace std::chrono_literals; @@ -40,6 +43,18 @@ struct RoomTestAccess { const std::scoped_lock guard(room.lock_); return room.listener_id_; } + + static void simulateScenario(Room& room, proto::SimulateScenarioKind scenario) { + std::shared_ptr handle; + { + const std::scoped_lock guard(room.lock_); + handle = room.room_handle_; + } + if (!handle || !handle->valid()) { + throw std::runtime_error("cannot simulate reconnect for a disconnected room"); + } + FfiClient::instance().simulateScenarioAsync(handle->get(), scenario).get(); + } }; } // namespace livekit @@ -219,6 +234,235 @@ class TokenRefreshTrackingDelegate : public RoomDelegate { std::string refreshed_token_; }; +class ReconnectTrackingDelegate : public RoomDelegate { +public: + void reset() { + const std::scoped_lock guard(mutex_); + callbacks_.clear(); + observed_states_.clear(); + republished_publication_.reset(); + republished_track_.reset(); + republished_participant_ = nullptr; + previous_sid_.clear(); + reconnecting_count_ = 0; + reconnected_count_ = 0; + republished_count_ = 0; + disconnected_count_ = 0; + } + + void onConnectionStateChanged(Room& room, const ConnectionStateChangedEvent& event) override { + const std::scoped_lock guard(mutex_); + if (event.state == ConnectionState::Reconnecting) { + callbacks_.emplace_back("state:reconnecting"); + } else if (event.state == ConnectionState::Connected) { + callbacks_.emplace_back("state:connected"); + } else { + callbacks_.emplace_back("state:disconnected"); + } + observed_states_.push_back(room.connectionState()); + } + + void onReconnecting(Room& room, const ReconnectingEvent&) override { + const std::scoped_lock guard(mutex_); + callbacks_.emplace_back("reconnecting"); + observed_states_.push_back(room.connectionState()); + ++reconnecting_count_; + cv_.notify_all(); + } + + void onReconnected(Room& room, const ReconnectedEvent&) override { + const std::scoped_lock guard(mutex_); + callbacks_.emplace_back("reconnected"); + observed_states_.push_back(room.connectionState()); + ++reconnected_count_; + cv_.notify_all(); + } + + void onLocalTrackRepublished(Room&, const LocalTrackRepublishedEvent& event) override { + const std::scoped_lock guard(mutex_); + callbacks_.emplace_back("republished"); + republished_publication_ = event.publication; + republished_track_ = event.track; + republished_participant_ = event.participant; + previous_sid_ = event.previous_sid; + ++republished_count_; + cv_.notify_all(); + } + + void onDisconnected(Room& room, const DisconnectedEvent&) override { + const std::scoped_lock guard(mutex_); + callbacks_.emplace_back("disconnected"); + observed_states_.push_back(room.connectionState()); + ++disconnected_count_; + cv_.notify_all(); + } + + bool waitForReconnected(std::chrono::seconds timeout) { + std::unique_lock lock(mutex_); + return cv_.wait_for(lock, timeout, [this] { return reconnected_count_ == 1; }); + } + + bool waitForReconnecting(std::chrono::seconds timeout) { + std::unique_lock lock(mutex_); + return cv_.wait_for(lock, timeout, [this] { return reconnecting_count_ >= 1; }); + } + + std::vector callbacks() const { + const std::scoped_lock guard(mutex_); + return callbacks_; + } + + std::vector observedStates() const { + const std::scoped_lock guard(mutex_); + return observed_states_; + } + + int disconnectedCount() const { + const std::scoped_lock guard(mutex_); + return disconnected_count_; + } + + int republishedCount() const { + const std::scoped_lock guard(mutex_); + return republished_count_; + } + + std::shared_ptr republishedPublication() const { + const std::scoped_lock guard(mutex_); + return republished_publication_; + } + + std::shared_ptr republishedTrack() const { + const std::scoped_lock guard(mutex_); + return republished_track_; + } + + LocalParticipant* republishedParticipant() const { + const std::scoped_lock guard(mutex_); + return republished_participant_; + } + + std::string previousSid() const { + const std::scoped_lock guard(mutex_); + return previous_sid_; + } + +private: + mutable std::mutex mutex_; + std::condition_variable cv_; + std::vector callbacks_; + std::vector observed_states_; + std::shared_ptr republished_publication_; + std::shared_ptr republished_track_; + LocalParticipant* republished_participant_ = nullptr; + std::string previous_sid_; + int reconnecting_count_ = 0; + int reconnected_count_ = 0; + int republished_count_ = 0; + int disconnected_count_ = 0; +}; + +class TrackSubscriptionWaitDelegate : public RoomDelegate { +public: + explicit TrackSubscriptionWaitDelegate(std::string expected_name) : expected_name_(std::move(expected_name)) {} + + void onTrackSubscribed(Room&, const TrackSubscribedEvent& event) override { + if (!event.track || event.track->name() != expected_name_) { + return; + } + { + const std::scoped_lock guard(mutex_); + ++subscription_count_; + } + cv_.notify_all(); + } + + bool wait(std::chrono::seconds timeout) { return waitForCount(1, timeout); } + + bool waitForCount(int expected_count, std::chrono::seconds timeout) { + std::unique_lock lock(mutex_); + return cv_.wait_for(lock, timeout, [this, expected_count] { return subscription_count_ >= expected_count; }); + } + +private: + std::string expected_name_; + std::mutex mutex_; + std::condition_variable cv_; + int subscription_count_ = 0; +}; + +class AudioFrameCounter { +public: + void onFrame(const AudioFrame& frame) { + if (frame.totalSamples() == 0) { + return; + } + { + const std::scoped_lock guard(mutex_); + ++count_; + } + cv_.notify_all(); + } + + int count() const { + const std::scoped_lock guard(mutex_); + return count_; + } + + bool waitForAtLeast(int expected_count, std::chrono::seconds timeout) { + std::unique_lock lock(mutex_); + return cv_.wait_for(lock, timeout, [this, expected_count] { return count_ >= expected_count; }); + } + +private: + mutable std::mutex mutex_; + std::condition_variable cv_; + int count_ = 0; +}; + +class ContinuousAudioCapture { +public: + explicit ContinuousAudioCapture(std::shared_ptr source) : source_(std::move(source)) {} + + ~ContinuousAudioCapture() { stop(); } + + void start() { + running_.store(true, std::memory_order_release); + thread_ = std::thread([this] { + constexpr int kSampleRate = 48'000; + constexpr int kSamplesPerChannel = 480; + AudioFrame frame = AudioFrame::create(kSampleRate, 1, kSamplesPerChannel); + auto& samples = frame.data(); + for (std::size_t i = 0; i < samples.size(); ++i) { + samples[i] = i % 2 == 0 ? 8'000 : -8'000; + } + + auto next_capture = std::chrono::steady_clock::now(); + while (running_.load(std::memory_order_acquire)) { + try { + source_->captureFrame(frame, 1'000); + } catch (const std::exception&) { + // Capture can fail transiently while the transport is rebuilding. + } + next_capture += 10ms; + std::this_thread::sleep_until(next_capture); + } + }); + } + + void stop() { + running_.store(false, std::memory_order_release); + if (thread_.joinable()) { + thread_.join(); + } + } + +private: + std::shared_ptr source_; + std::atomic running_{false}; + std::thread thread_; +}; + } // namespace // livekit-server sends RefreshToken immediately after join, then every ~5 minutes. @@ -436,4 +680,221 @@ TEST_F(RoomLifecycleTest, ParticipantHandlesExpireOnRoomDestruction) { peer.reset(); } +class RoomReconnectTest : public LiveKitTestBase {}; + +TEST_F(RoomReconnectTest, SignalResumePreservesPublicationIdentityAndCallbackOrder) { + if (!config_.available) { + GTEST_SKIP() << "LIVEKIT_URL, LIVEKIT_TOKEN_A, and LIVEKIT_TOKEN_B not set"; + } + + ReconnectTrackingDelegate delegate; + Room room; + room.setDelegate(&delegate); + ASSERT_TRUE(room.connect(config_.url, config_.token_a, RoomOptions{})); + + constexpr char kTrackName[] = "resume-video"; + TrackSubscriptionWaitDelegate receiver_delegate(kTrackName); + Room receiver; + receiver.setDelegate(&receiver_delegate); + ASSERT_TRUE(receiver.connect(config_.url, config_.token_b, RoomOptions{})); + + auto participant = lockLocalParticipant(room); + const std::string sender_identity = participant->identity(); + AudioFrameCounter frame_counter; + receiver.setOnAudioFrameCallback(sender_identity, kTrackName, + [&frame_counter](const AudioFrame& frame) { frame_counter.onFrame(frame); }); + + auto source = std::make_shared(48'000, 1); + auto track = LocalAudioTrack::createLocalAudioTrack(kTrackName, source); + TrackPublishOptions publish_options; + publish_options.source = TrackSource::SOURCE_MICROPHONE; + participant->publishTrack(track, publish_options); + const auto publication = track->publication(); + ASSERT_NE(publication, nullptr); + const std::string initial_sid = publication->sid(); + ContinuousAudioCapture capture(source); + capture.start(); + ASSERT_TRUE(receiver_delegate.wait(15s)); + ASSERT_TRUE(frame_counter.waitForAtLeast(10, 20s)) << "receiver did not get audio before signal resume"; + + delegate.reset(); + RoomTestAccess::simulateScenario(room, proto::SIMULATE_SIGNAL_RECONNECT); + ASSERT_TRUE(delegate.waitForReconnected(30s)); + + EXPECT_EQ(room.connectionState(), ConnectionState::Connected); + EXPECT_EQ(track->publication().get(), publication.get()); + EXPECT_EQ(publication->sid(), initial_sid); + EXPECT_EQ(track->sid(), initial_sid); + const std::vector expected{"state:reconnecting", "reconnecting", "state:connected", "reconnected"}; + EXPECT_EQ(delegate.callbacks(), expected); + + const auto observed_states = delegate.observedStates(); + ASSERT_GE(observed_states.size(), 4u); + EXPECT_EQ(observed_states[0], ConnectionState::Reconnecting); + EXPECT_EQ(observed_states[1], ConnectionState::Reconnecting); + EXPECT_EQ(observed_states[2], ConnectionState::Connected); + EXPECT_EQ(observed_states[3], ConnectionState::Connected); + EXPECT_EQ(delegate.republishedCount(), 0); + + const int frames_at_reconnect = frame_counter.count(); + EXPECT_TRUE(frame_counter.waitForAtLeast(frames_at_reconnect + 10, 20s)) + << "existing audio callback registration did not resume after signal reconnect"; + + capture.stop(); + receiver.clearOnAudioFrameCallback(sender_identity, kTrackName); + participant->unpublishTrack(publication->sid()); +} + +TEST_F(RoomReconnectTest, FullReconnectRekeysPublicationInPlaceBeforeReconnected) { + if (!config_.available) { + GTEST_SKIP() << "LIVEKIT_URL, LIVEKIT_TOKEN_A, and LIVEKIT_TOKEN_B not set"; + } + + ReconnectTrackingDelegate delegate; + Room room; + room.setDelegate(&delegate); + ASSERT_TRUE(room.connect(config_.url, config_.token_a, RoomOptions{})); + + constexpr char kTrackName[] = "full-reconnect-audio"; + TrackSubscriptionWaitDelegate receiver_delegate(kTrackName); + Room receiver; + receiver.setDelegate(&receiver_delegate); + ASSERT_TRUE(receiver.connect(config_.url, config_.token_b, RoomOptions{})); + + auto participant = lockLocalParticipant(room); + const std::string sender_identity = participant->identity(); + AudioFrameCounter frame_counter; + receiver.setOnAudioFrameCallback(sender_identity, kTrackName, + [&frame_counter](const AudioFrame& frame) { frame_counter.onFrame(frame); }); + + auto source = std::make_shared(48'000, 1); + auto track = LocalAudioTrack::createLocalAudioTrack(kTrackName, source); + TrackPublishOptions publish_options; + publish_options.source = TrackSource::SOURCE_MICROPHONE; + participant->publishTrack(track, publish_options); + const auto publication = track->publication(); + ASSERT_NE(publication, nullptr); + const std::string initial_sid = publication->sid(); + ContinuousAudioCapture capture(source); + capture.start(); + ASSERT_TRUE(receiver_delegate.waitForCount(1, 15s)); + ASSERT_TRUE(frame_counter.waitForAtLeast(10, 20s)) << "receiver did not get audio before full reconnect"; + + delegate.reset(); + RoomTestAccess::simulateScenario(room, proto::SIMULATE_FULL_RECONNECT); + ASSERT_TRUE(delegate.waitForReconnected(30s)); + + const std::string current_sid = publication->sid(); + EXPECT_NE(current_sid, initial_sid); + EXPECT_EQ(track->publication().get(), publication.get()); + EXPECT_EQ(track->sid(), current_sid); + + const auto publications = participant->trackPublications(); + EXPECT_EQ(publications.count(initial_sid), 0u); + const auto current = publications.find(current_sid); + ASSERT_NE(current, publications.end()); + EXPECT_EQ(current->second.get(), publication.get()); + + const auto callbacks = delegate.callbacks(); + const std::vector expected_callbacks{"state:reconnecting", "reconnecting", "republished", + "state:connected", "reconnected"}; + EXPECT_EQ(callbacks, expected_callbacks); + EXPECT_EQ(delegate.republishedCount(), 1); + EXPECT_EQ(delegate.previousSid(), initial_sid); + EXPECT_EQ(delegate.republishedPublication().get(), publication.get()); + EXPECT_EQ(delegate.republishedTrack().get(), track.get()); + EXPECT_EQ(delegate.republishedParticipant(), participant.get()); + + ASSERT_TRUE(receiver_delegate.waitForCount(2, 20s)) << "receiver did not resubscribe after full reconnect"; + const int frames_at_reconnect = frame_counter.count(); + ASSERT_TRUE(frame_counter.waitForAtLeast(frames_at_reconnect + 10, 20s)) + << "existing audio callback registration did not resume after full reconnect"; + + const auto remote_participant = lockRemoteParticipant(receiver, sender_identity); + const auto& remote_publications = remote_participant->trackPublications(); + const auto matching_publications = + std::count_if(remote_publications.begin(), remote_publications.end(), [&](const auto& entry) { + return entry.second && entry.second->name() == kTrackName && entry.second->kind() == TrackKind::KIND_AUDIO; + }); + EXPECT_EQ(matching_publications, 1) << "full reconnect must not leave duplicate remote publications"; + + capture.stop(); + receiver.clearOnAudioFrameCallback(sender_identity, kTrackName); + EXPECT_NO_THROW(participant->unpublishTrack(current_sid)); +} + +TEST_F(RoomReconnectTest, FailedResumeEscalatesToFullReconnect) { + if (!config_.available) { + GTEST_SKIP() << "LIVEKIT_URL, LIVEKIT_TOKEN_A, and LIVEKIT_TOKEN_B not set"; + } + + ReconnectTrackingDelegate delegate; + Room room; + room.setDelegate(&delegate); + ASSERT_TRUE(room.connect(config_.url, config_.token_a, RoomOptions{})); + + auto participant = lockLocalParticipant(room); + auto source = std::make_shared(16, 16); + auto track = LocalVideoTrack::createLocalVideoTrack("escalation-video", source); + participant->publishTrack(track, TrackPublishOptions{}); + const auto publication = track->publication(); + ASSERT_NE(publication, nullptr); + const std::string initial_sid = publication->sid(); + + delegate.reset(); + RoomTestAccess::simulateScenario(room, proto::SIMULATE_DISCONNECT_SIGNAL_ON_RESUME); + ASSERT_TRUE(delegate.waitForReconnected(45s)); + EXPECT_NE(publication->sid(), initial_sid); + EXPECT_EQ(track->sid(), publication->sid()); + EXPECT_EQ(delegate.republishedCount(), 1); + EXPECT_EQ(delegate.previousSid(), initial_sid); + EXPECT_EQ(delegate.republishedPublication().get(), publication.get()); + EXPECT_EQ(delegate.republishedTrack().get(), track.get()); + EXPECT_EQ(delegate.republishedParticipant(), participant.get()); + + participant->unpublishTrack(publication->sid()); +} + +TEST_F(RoomReconnectTest, RepeatedSignalReconnectCyclesRemainBounded) { + if (!config_.available) { + GTEST_SKIP() << "LIVEKIT_URL, LIVEKIT_TOKEN_A, and LIVEKIT_TOKEN_B not set"; + } + + ReconnectTrackingDelegate delegate; + Room room; + room.setDelegate(&delegate); + ASSERT_TRUE(room.connect(config_.url, config_.token_a, RoomOptions{})); + + constexpr int kCycles = 3; + for (int cycle = 0; cycle < kCycles; ++cycle) { + delegate.reset(); + const auto started = std::chrono::steady_clock::now(); + RoomTestAccess::simulateScenario(room, proto::SIMULATE_SIGNAL_RECONNECT); + ASSERT_TRUE(delegate.waitForReconnected(30s)) << "cycle " << cycle; + EXPECT_LT(std::chrono::steady_clock::now() - started, 30s); + EXPECT_EQ(room.connectionState(), ConnectionState::Connected); + } +} + +TEST_F(RoomReconnectTest, DisconnectDuringRecoveryNotifiesExactlyOnce) { + if (!config_.available) { + GTEST_SKIP() << "LIVEKIT_URL, LIVEKIT_TOKEN_A, and LIVEKIT_TOKEN_B not set"; + } + + ReconnectTrackingDelegate delegate; + Room room; + room.setDelegate(&delegate); + ASSERT_TRUE(room.connect(config_.url, config_.token_a, RoomOptions{})); + + delegate.reset(); + RoomTestAccess::simulateScenario(room, proto::SIMULATE_SIGNAL_RECONNECT); + ASSERT_TRUE(delegate.waitForReconnecting(10s)); + EXPECT_TRUE(room.disconnect()); + EXPECT_EQ(delegate.disconnectedCount(), 1); + EXPECT_EQ(room.connectionState(), ConnectionState::Disconnected); + + EXPECT_FALSE(room.disconnect()); + EXPECT_EQ(delegate.disconnectedCount(), 1); +} + } // namespace livekit::test diff --git a/src/tests/unit/test_room.cpp b/src/tests/unit/test_room.cpp index 6ea52d0c..9c4ecffc 100644 --- a/src/tests/unit/test_room.cpp +++ b/src/tests/unit/test_room.cpp @@ -20,6 +20,7 @@ #include #include #include +#include #include #include @@ -39,6 +40,7 @@ struct RoomTestAccess { const std::scoped_lock guard(room.lock_); room.connection_state_ = ConnectionState::Connected; + room.disconnect_callback_claimed_ = false; room.room_handle_ = std::make_shared(); room.listener_id_ = listener_id; } @@ -85,6 +87,39 @@ class UnitLifecycleTrackingDelegate : public RoomDelegate { DisconnectReason reason = DisconnectReason::Unknown; }; +class ReconnectLifecycleDelegate : public RoomDelegate { +public: + void onConnectionStateChanged(Room& room, const ConnectionStateChangedEvent& event) override { + const std::scoped_lock guard(mutex_); + callbacks.emplace_back(event.state == ConnectionState::Reconnecting ? "state:reconnecting" + : event.state == ConnectionState::Connected ? "state:connected" + : "state:disconnected"); + observed_states.push_back(room.connectionState()); + } + + void onReconnecting(Room& room, const ReconnectingEvent&) override { + const std::scoped_lock guard(mutex_); + callbacks.emplace_back("reconnecting"); + observed_states.push_back(room.connectionState()); + } + + void onReconnected(Room& room, const ReconnectedEvent&) override { + const std::scoped_lock guard(mutex_); + callbacks.emplace_back("reconnected"); + observed_states.push_back(room.connectionState()); + } + + void onDisconnected(Room& room, const DisconnectedEvent&) override { + const std::scoped_lock guard(mutex_); + callbacks.emplace_back("disconnected"); + observed_states.push_back(room.connectionState()); + } + + std::mutex mutex_; + std::vector callbacks; + std::vector observed_states; +}; + } // namespace TEST_F(RoomTest, ConnectWithoutInitialize) { @@ -401,4 +436,93 @@ TEST_F(RoomTest, DisconnectAfterServerDisconnectCleansUpWithoutDuplicateNotifica EXPECT_EQ(delegate.callbacks.size(), 1) << "onDisconnected must not fire twice"; } +TEST_F(RoomTest, ReconnectEventsUpdateStateBeforeCallbacks) { + ReconnectLifecycleDelegate delegate; + Room room; + std::atomic listener_calls{0}; + room.setDelegate(&delegate); + RoomTestAccess::installConnectedListener(room, listener_calls); + + proto::FfiEvent state_reconnecting; + auto* reconnecting_room_event = state_reconnecting.mutable_room_event(); + reconnecting_room_event->set_room_handle(0); + reconnecting_room_event->mutable_connection_state_changed()->set_state(proto::CONN_RECONNECTING); + emitFfiEvent(state_reconnecting); + + proto::FfiEvent reconnecting; + reconnecting.mutable_room_event()->set_room_handle(0); + reconnecting.mutable_room_event()->mutable_reconnecting(); + emitFfiEvent(reconnecting); + + proto::FfiEvent state_connected; + auto* connected_room_event = state_connected.mutable_room_event(); + connected_room_event->set_room_handle(0); + connected_room_event->mutable_connection_state_changed()->set_state(proto::CONN_CONNECTED); + emitFfiEvent(state_connected); + + proto::FfiEvent reconnected; + reconnected.mutable_room_event()->set_room_handle(0); + reconnected.mutable_room_event()->mutable_reconnected(); + emitFfiEvent(reconnected); + + const std::vector expected{"state:reconnecting", "reconnecting", "state:connected", "reconnected"}; + EXPECT_EQ(delegate.callbacks, expected); + ASSERT_EQ(delegate.observed_states.size(), expected.size()); + EXPECT_EQ(delegate.observed_states[0], ConnectionState::Reconnecting); + EXPECT_EQ(delegate.observed_states[1], ConnectionState::Reconnecting); + EXPECT_EQ(delegate.observed_states[2], ConnectionState::Connected); + EXPECT_EQ(delegate.observed_states[3], ConnectionState::Connected); + EXPECT_EQ(room.connectionState(), ConnectionState::Connected); +} + +TEST_F(RoomTest, ConnectionStateDisconnectedDoesNotSuppressDisconnectedCallback) { + ReconnectLifecycleDelegate delegate; + Room room; + std::atomic listener_calls{0}; + room.setDelegate(&delegate); + RoomTestAccess::installConnectedListener(room, listener_calls); + + proto::FfiEvent state_disconnected; + auto* state_room_event = state_disconnected.mutable_room_event(); + state_room_event->set_room_handle(0); + state_room_event->mutable_connection_state_changed()->set_state(proto::CONN_DISCONNECTED); + emitFfiEvent(state_disconnected); + + proto::FfiEvent disconnected; + auto* disconnected_room_event = disconnected.mutable_room_event(); + disconnected_room_event->set_room_handle(0); + disconnected_room_event->mutable_disconnected()->set_reason(proto::SIGNAL_CLOSE); + emitFfiEvent(disconnected); + + const std::vector expected{"state:disconnected", "disconnected"}; + EXPECT_EQ(delegate.callbacks, expected); + ASSERT_EQ(delegate.observed_states.size(), expected.size()); + EXPECT_EQ(delegate.observed_states[0], ConnectionState::Disconnected); + EXPECT_EQ(delegate.observed_states[1], ConnectionState::Disconnected); +} + +TEST_F(RoomTest, ConnectWaitsForRoomEosAfterTerminalDisconnect) { + Room room; + std::atomic listener_calls{0}; + RoomTestAccess::installConnectedListener(room, listener_calls); + + proto::FfiEvent disconnected; + auto* disconnected_room_event = disconnected.mutable_room_event(); + disconnected_room_event->set_room_handle(0); + disconnected_room_event->mutable_disconnected()->set_reason(proto::SIGNAL_CLOSE); + emitFfiEvent(disconnected); + + EXPECT_EQ(room.connectionState(), ConnectionState::Disconnected); + EXPECT_TRUE(RoomTestAccess::hasRoomHandle(room)); + EXPECT_THROW((void)room.connect("ws://unused", "unused", RoomOptions{}), std::runtime_error); + + proto::FfiEvent eos; + eos.mutable_room_event()->set_room_handle(0); + eos.mutable_room_event()->mutable_eos(); + emitFfiEvent(eos); + + EXPECT_FALSE(RoomTestAccess::hasRoomHandle(room)); + EXPECT_EQ(RoomTestAccess::listenerId(room), 0); +} + } // namespace livekit::test