From 562fa8dc19d30ca516cd776dcd82ef7adefdd7a6 Mon Sep 17 00:00:00 2001 From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com> Date: Tue, 6 Oct 2026 15:30:32 -0700 Subject: [PATCH 1/4] Handle track republish after full-reconnect --- include/livekit/local_track_publication.h | 3 + src/ffi_client.cpp | 38 ++++++++++++ src/ffi_client.h | 2 + src/room.cpp | 62 +++++++++++++++++++ src/tests/common/room_test_access.h | 49 +++++++++++++++ .../test_local_track_publish_sid.cpp | 47 ++++++++++++++ src/tests/integration/test_room.cpp | 12 +--- 7 files changed, 202 insertions(+), 11 deletions(-) create mode 100644 src/tests/common/room_test_access.h diff --git a/include/livekit/local_track_publication.h b/include/livekit/local_track_publication.h index 412b4fb9..bae23d94 100644 --- a/include/livekit/local_track_publication.h +++ b/include/livekit/local_track_publication.h @@ -31,6 +31,9 @@ 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 Room; }; } // namespace livekit diff --git a/src/ffi_client.cpp b/src/ffi_client.cpp index 6c924560..867eba94 100644 --- a/src/ffi_client.cpp +++ b/src/ffi_client.cpp @@ -103,6 +103,8 @@ std::optional ExtractAsyncId(const proto::FfiEvent& event) { return event.get_stats().async_id(); case E::kGetSessionStats: return event.get_session_stats().async_id(); + case E::kSimulateScenario: + return event.simulate_scenario().async_id(); case E::kPublishSipDtmf: return event.publish_sip_dtmf().async_id(); case E::kChatMessage: @@ -655,6 +657,42 @@ std::future FfiClient::getSessionStatsAsync(uintptr_t room_handle) return fut; } +std::future FfiClient::simulateScenarioAsync(uintptr_t room_handle, int 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& cb = event.simulate_scenario(); + if (cb.has_error() && !cb.error().empty()) { + pr.set_exception(std::make_exception_ptr(std::runtime_error(cb.error()))); + return; + } + pr.set_value(); + }); + + proto::FfiRequest req; + auto* msg = req.mutable_simulate_scenario(); + msg->set_room_handle(room_handle); + msg->set_scenario(static_cast(scenario)); + msg->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; +} + // Participant APIs Implementation std::future FfiClient::publishTrackAsync(std::uint64_t local_participant_handle, std::uint64_t track_handle, diff --git a/src/ffi_client.h b/src/ffi_client.h index c0835027..0104de76 100644 --- a/src/ffi_client.h +++ b/src/ffi_client.h @@ -106,6 +106,8 @@ class LIVEKIT_INTERNAL_API FfiClient { std::future getSessionStatsAsync(uintptr_t room_handle); + std::future simulateScenarioAsync(uintptr_t room_handle, int scenario); + // Participant APIs std::future publishTrackAsync(std::uint64_t local_participant_handle, std::uint64_t track_handle, diff --git a/src/room.cpp b/src/room.cpp index f37c5642..ff1dcd19 100644 --- a/src/room.cpp +++ b/src/room.cpp @@ -20,7 +20,10 @@ #include "ffi_client.h" #include "livekit/audio_stream.h" #include "livekit/e2ee.h" +#include "livekit/local_audio_track.h" #include "livekit/local_participant.h" +#include "livekit/local_track_publication.h" +#include "livekit/local_video_track.h" #include "livekit/remote_audio_track.h" #include "livekit/remote_data_track.h" #include "livekit/remote_participant.h" @@ -47,6 +50,33 @@ using proto::FfiResponse; namespace { +std::shared_ptr localTrackPublication(const std::shared_ptr& track) { + if (!track) { + return nullptr; + } + if (auto video = std::dynamic_pointer_cast(track)) { + return video->publication(); + } + if (auto audio = std::dynamic_pointer_cast(track)) { + return audio->publication(); + } + return nullptr; +} + +void updateLocalTrackPublicationInfo(LocalTrackPublication& publication, const proto::TrackPublicationInfo& info) { + publication.sid_ = info.sid(); + publication.name_ = info.name(); + publication.kind_ = fromProto(info.kind()); + publication.source_ = fromProto(info.source()); + publication.simulcasted_ = info.simulcasted(); + publication.width_ = info.width(); + publication.height_ = info.height(); + publication.mime_type_ = info.mime_type(); + publication.muted_ = info.muted(); + publication.encryption_type_ = static_cast(info.encryption_type()); + publication.audio_features_ = convertAudioFeatures(info.audio_features()); +} + std::shared_ptr createRemoteParticipant(const proto::OwnedParticipant& owned) { const auto& pinfo = owned.info(); std::unordered_map attrs; @@ -593,6 +623,38 @@ void Room::onEvent(const FfiEvent& event) { } break; } + case proto::RoomEvent::kLocalTrackRepublished: { + const std::scoped_lock guard(lock_); + if (!local_participant_) { + LK_LOG_ERROR("kLocalTrackRepublished: local_participant_ is nullptr"); + break; + } + const auto& ltr = re.local_track_republished(); + const std::string& previous_sid = ltr.previous_sid(); + + auto& published = local_participant_->published_tracks_by_sid_; + auto it = published.find(previous_sid); + if (it == published.end()) { + LK_LOG_WARN("local_track_republished for unknown previous sid: {}", previous_sid); + break; + } + auto track = it->second.lock(); + if (!track) { + published.erase(it); + LK_LOG_WARN("local_track_republished for expired previous sid: {}", previous_sid); + break; + } + auto publication = localTrackPublication(track); + if (!publication) { + LK_LOG_WARN("local_track_republished missing publication for sid: {}", previous_sid); + break; + } + updateLocalTrackPublicationInfo(*publication, ltr.info()); + published.erase(it); + published[publication->sid()] = track; + track->setPublication(publication); + break; + } case proto::RoomEvent::kLocalTrackSubscribed: { LocalTrackSubscribedEvent ev; { diff --git a/src/tests/common/room_test_access.h b/src/tests/common/room_test_access.h new file mode 100644 index 00000000..9509b00c --- /dev/null +++ b/src/tests/common/room_test_access.h @@ -0,0 +1,49 @@ +/* + * Copyright 2026 LiveKit + * + * 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. + */ + +#pragma once + +#include + +#include +#include +#include + +#include "ffi_client.h" +#include "room.pb.h" + +namespace livekit { + +struct RoomTestAccess { + static int listenerId(const Room& room) { + 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) { + throw std::runtime_error("cannot simulate scenario for a disconnected room"); + } + FfiClient::instance().simulateScenarioAsync(handle->get(), static_cast(scenario)).get(); + } +}; + +} // namespace livekit diff --git a/src/tests/integration/test_local_track_publish_sid.cpp b/src/tests/integration/test_local_track_publish_sid.cpp index 91fbf33b..ee767450 100644 --- a/src/tests/integration/test_local_track_publish_sid.cpp +++ b/src/tests/integration/test_local_track_publish_sid.cpp @@ -16,15 +16,21 @@ #include +#include #include #include +#include #include "../common/audio_utils.h" +#include "../common/room_test_access.h" #include "../common/test_common.h" +#include "room.pb.h" namespace livekit::test { namespace { +using namespace std::chrono_literals; + void expectTrackSidAssigned(const Track& track, const LocalTrackPublication& publication) { const std::string& track_sid = track.sid(); const std::string& publication_sid = publication.sid(); @@ -33,6 +39,17 @@ void expectTrackSidAssigned(const Track& track, const LocalTrackPublication& pub EXPECT_EQ(track_sid, publication_sid); } +bool waitForSidChange(const Track& track, const std::string& previous_sid, std::chrono::milliseconds timeout) { + const auto deadline = std::chrono::steady_clock::now() + timeout; + while (std::chrono::steady_clock::now() < deadline) { + if (track.sid() != previous_sid && track.sid() != "TR_unknown") { + return true; + } + std::this_thread::sleep_for(50ms); + } + return track.sid() != previous_sid && track.sid() != "TR_unknown"; +} + } // namespace class LocalTrackPublishSidTest : public LiveKitTestBase {}; @@ -73,4 +90,34 @@ TEST_F(LocalTrackPublishSidTest, PublishAudioTrackAssignsSid) { lockLocalParticipant(room)->unpublishTrack(track->publication()->sid()); } +TEST_F(LocalTrackPublishSidTest, FullReconnectUpdatesPublishedSid) { + failIfNotConfigured(); + + Room room; + const RoomOptions room_options; + ASSERT_TRUE(room.connect(config_.url, config_.token_a, room_options)); + + auto source = std::make_shared(VideoCodec::H264, 16, 16); + std::shared_ptr track; + ASSERT_NO_THROW( + track = lockLocalParticipant(room)->publishVideoTrack("republish-sid-check", source, TrackSource::SOURCE_CAMERA)); + ASSERT_NE(track, nullptr); + ASSERT_NE(track->publication(), nullptr); + expectTrackSidAssigned(*track, *track->publication()); + + const std::string previous_sid = track->sid(); + ASSERT_NO_THROW(RoomTestAccess::simulateScenario(room, proto::SIMULATE_FULL_RECONNECT)); + ASSERT_TRUE(waitForSidChange(*track, previous_sid, 30s)) << "Timed out waiting for republished track SID"; + + ASSERT_NE(track->publication(), nullptr); + expectTrackSidAssigned(*track, *track->publication()); + EXPECT_NE(track->sid(), previous_sid); + + const auto pubs = lockLocalParticipant(room)->trackPublications(); + EXPECT_EQ(pubs.count(previous_sid), 0u); + EXPECT_EQ(pubs.count(track->sid()), 1u); + + lockLocalParticipant(room)->unpublishTrack(track->sid()); +} + } // namespace livekit::test diff --git a/src/tests/integration/test_room.cpp b/src/tests/integration/test_room.cpp index 942d8939..b6f3c80a 100644 --- a/src/tests/integration/test_room.cpp +++ b/src/tests/integration/test_room.cpp @@ -29,21 +29,11 @@ #include #include +#include "../common/room_test_access.h" #include "../common/test_common.h" using namespace std::chrono_literals; -namespace livekit { - -struct RoomTestAccess { - static int listenerId(const Room& room) { - const std::scoped_lock guard(room.lock_); - return room.listener_id_; - } -}; - -} // namespace livekit - namespace livekit::test { // Server-dependent tests - require LIVEKIT_URL and LIVEKIT_TOKEN_A env vars From ee9762ed1443b13aec1e46cb89366601e8fafc74 Mon Sep 17 00:00:00 2001 From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com> Date: Mon, 12 Oct 2026 16:27:19 -0700 Subject: [PATCH 2/4] Address comments --- src/room.cpp | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) diff --git a/src/room.cpp b/src/room.cpp index ff1dcd19..2b2334e2 100644 --- a/src/room.cpp +++ b/src/room.cpp @@ -629,18 +629,18 @@ void Room::onEvent(const FfiEvent& event) { LK_LOG_ERROR("kLocalTrackRepublished: local_participant_ is nullptr"); break; } - const auto& ltr = re.local_track_republished(); - const std::string& previous_sid = ltr.previous_sid(); + const auto& local_track_repub = re.local_track_republished(); + const std::string& previous_sid = local_track_repub.previous_sid(); - auto& published = local_participant_->published_tracks_by_sid_; - auto it = published.find(previous_sid); - if (it == published.end()) { + auto& published_tracks = local_participant_->published_tracks_by_sid_; + auto it = published_tracks.find(previous_sid); + if (it == published_tracks.end()) { LK_LOG_WARN("local_track_republished for unknown previous sid: {}", previous_sid); break; } auto track = it->second.lock(); if (!track) { - published.erase(it); + published_tracks.erase(it); LK_LOG_WARN("local_track_republished for expired previous sid: {}", previous_sid); break; } @@ -649,9 +649,9 @@ void Room::onEvent(const FfiEvent& event) { LK_LOG_WARN("local_track_republished missing publication for sid: {}", previous_sid); break; } - updateLocalTrackPublicationInfo(*publication, ltr.info()); - published.erase(it); - published[publication->sid()] = track; + updateLocalTrackPublicationInfo(*publication, local_track_repub.info()); + published_tracks.erase(it); + published_tracks[publication->sid()] = track; track->setPublication(publication); break; } From 679996f07125287ce387934e216cdfdf13629110 Mon Sep 17 00:00:00 2001 From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com> Date: Wed, 14 Oct 2026 11:21:50 -0700 Subject: [PATCH 3/4] Adress build error and sync issue --- include/livekit/local_participant.h | 4 ++++ include/livekit/local_track_publication.h | 4 ++++ src/local_participant.cpp | 22 +++++++++++++++------- src/local_track_publication.cpp | 15 +++++++++++++++ src/room.cpp | 17 ++--------------- 5 files changed, 40 insertions(+), 22 deletions(-) diff --git a/include/livekit/local_participant.h b/include/livekit/local_participant.h index 28ba83ef..557205b0 100644 --- a/include/livekit/local_participant.h +++ b/include/livekit/local_participant.h @@ -293,6 +293,10 @@ class LIVEKIT_API LocalParticipant : public Participant { friend class Room; private: + /// Guards published_tracks_by_sid_ and coordinated publication/track updates. + /// Never held across blocking FFI requests. When both this and Room::lock_ + /// are needed, acquire Room::lock_ first. + mutable std::mutex published_tracks_mutex_; /// 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. diff --git a/include/livekit/local_track_publication.h b/include/livekit/local_track_publication.h index bae23d94..bea1a6d8 100644 --- a/include/livekit/local_track_publication.h +++ b/include/livekit/local_track_publication.h @@ -23,6 +23,7 @@ namespace livekit { namespace proto { class OwnedTrackPublication; +class TrackPublicationInfo; } /// @brief A track published by a local participant. @@ -34,6 +35,9 @@ class LIVEKIT_API LocalTrackPublication : public TrackPublication { private: friend class Room; + + /// Refresh publication metadata after a full reconnect republish. + void updateInfo(const proto::TrackPublicationInfo& info); }; } // namespace livekit diff --git a/src/local_participant.cpp b/src/local_participant.cpp index 5203ca83..8dbc1fdc 100644 --- a/src/local_participant.cpp +++ b/src/local_participant.cpp @@ -18,6 +18,7 @@ #include #include +#include #include #include "data_track.pb.h" @@ -200,9 +201,11 @@ 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->setPublication(publication); + { + const std::scoped_lock guard(published_tracks_mutex_); + published_tracks_by_sid_[sid] = std::weak_ptr(track); + track->setPublication(publication); + } } std::shared_ptr LocalParticipant::publishVideoTrack(const std::string& name, @@ -248,16 +251,20 @@ 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); + { + const std::scoped_lock guard(published_tracks_mutex_); + 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); + } + published_tracks_by_sid_.erase(it); } - published_tracks_by_sid_.erase(it); } } LocalParticipant::PublicationMap LocalParticipant::trackPublications() const { PublicationMap out; + const std::scoped_lock guard(published_tracks_mutex_); for (auto it = published_tracks_by_sid_.begin(); it != published_tracks_by_sid_.end();) { auto t = it->second.lock(); if (!t) { @@ -494,6 +501,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(published_tracks_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..26979f70 100644 --- a/src/local_track_publication.cpp +++ b/src/local_track_publication.cpp @@ -16,6 +16,7 @@ #include "livekit/local_track_publication.h" +#include "track.pb.h" #include "track_proto_converter.h" namespace livekit { @@ -27,4 +28,18 @@ LocalTrackPublication::LocalTrackPublication(const proto::OwnedTrackPublication& static_cast(owned.info().encryption_type()), convertAudioFeatures(owned.info().audio_features())) {} +void LocalTrackPublication::updateInfo(const proto::TrackPublicationInfo& info) { + 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 2b2334e2..99b39b97 100644 --- a/src/room.cpp +++ b/src/room.cpp @@ -63,20 +63,6 @@ std::shared_ptr localTrackPublication(const std::shared_p return nullptr; } -void updateLocalTrackPublicationInfo(LocalTrackPublication& publication, const proto::TrackPublicationInfo& info) { - publication.sid_ = info.sid(); - publication.name_ = info.name(); - publication.kind_ = fromProto(info.kind()); - publication.source_ = fromProto(info.source()); - publication.simulcasted_ = info.simulcasted(); - publication.width_ = info.width(); - publication.height_ = info.height(); - publication.mime_type_ = info.mime_type(); - publication.muted_ = info.muted(); - publication.encryption_type_ = static_cast(info.encryption_type()); - publication.audio_features_ = convertAudioFeatures(info.audio_features()); -} - std::shared_ptr createRemoteParticipant(const proto::OwnedParticipant& owned) { const auto& pinfo = owned.info(); std::unordered_map attrs; @@ -632,6 +618,7 @@ void Room::onEvent(const FfiEvent& event) { const auto& local_track_repub = re.local_track_republished(); const std::string& previous_sid = local_track_repub.previous_sid(); + const std::scoped_lock tracks_guard(local_participant_->published_tracks_mutex_); auto& published_tracks = local_participant_->published_tracks_by_sid_; auto it = published_tracks.find(previous_sid); if (it == published_tracks.end()) { @@ -649,7 +636,7 @@ void Room::onEvent(const FfiEvent& event) { LK_LOG_WARN("local_track_republished missing publication for sid: {}", previous_sid); break; } - updateLocalTrackPublicationInfo(*publication, local_track_repub.info()); + publication->updateInfo(local_track_repub.info()); published_tracks.erase(it); published_tracks[publication->sid()] = track; track->setPublication(publication); From 14e7d25f778016b46880387c63590171e5c34e73 Mon Sep 17 00:00:00 2001 From: Alan George Date: Wed, 14 Oct 2026 16:29:58 -0600 Subject: [PATCH 4/4] clang-format fixes --- include/livekit/local_track_publication.h | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/include/livekit/local_track_publication.h b/include/livekit/local_track_publication.h index bea1a6d8..a0ccbf3b 100644 --- a/include/livekit/local_track_publication.h +++ b/include/livekit/local_track_publication.h @@ -24,7 +24,7 @@ namespace livekit { namespace proto { class OwnedTrackPublication; class TrackPublicationInfo; -} +} // namespace proto /// @brief A track published by a local participant. class LIVEKIT_API LocalTrackPublication : public TrackPublication {