Skip to content

Commit 80c5591

Browse files
ladvocalan-george-lk
authored andcommitted
Handle track republish after full-reconnect
1 parent 95ab708 commit 80c5591

7 files changed

Lines changed: 202 additions & 11 deletions

File tree

‎include/livekit/local_track_publication.h‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,9 @@ class LIVEKIT_API LocalTrackPublication : public TrackPublication {
3131
/// Note, this LocalTrackPublication is constructed internally only;
3232
/// safe to accept proto::OwnedTrackPublication.
3333
explicit LocalTrackPublication(const proto::OwnedTrackPublication& owned);
34+
35+
private:
36+
friend class Room;
3437
};
3538

3639
} // namespace livekit

‎src/ffi_client.cpp‎

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -103,6 +103,8 @@ std::optional<FfiClient::AsyncId> ExtractAsyncId(const proto::FfiEvent& event) {
103103
return event.get_stats().async_id();
104104
case E::kGetSessionStats:
105105
return event.get_session_stats().async_id();
106+
case E::kSimulateScenario:
107+
return event.simulate_scenario().async_id();
106108
case E::kPublishSipDtmf:
107109
return event.publish_sip_dtmf().async_id();
108110
case E::kChatMessage:
@@ -655,6 +657,42 @@ std::future<SessionStats> FfiClient::getSessionStatsAsync(uintptr_t room_handle)
655657
return fut;
656658
}
657659

660+
std::future<void> FfiClient::simulateScenarioAsync(uintptr_t room_handle, int scenario) {
661+
const AsyncId async_id = generateAsyncId();
662+
663+
auto fut = registerAsync<void>(
664+
async_id,
665+
[async_id](const proto::FfiEvent& event) {
666+
return event.has_simulate_scenario() && event.simulate_scenario().async_id() == async_id;
667+
},
668+
[](const proto::FfiEvent& event, std::promise<void>& pr) {
669+
const auto& cb = event.simulate_scenario();
670+
if (cb.has_error() && !cb.error().empty()) {
671+
pr.set_exception(std::make_exception_ptr(std::runtime_error(cb.error())));
672+
return;
673+
}
674+
pr.set_value();
675+
});
676+
677+
proto::FfiRequest req;
678+
auto* msg = req.mutable_simulate_scenario();
679+
msg->set_room_handle(room_handle);
680+
msg->set_scenario(static_cast<proto::SimulateScenarioKind>(scenario));
681+
msg->set_request_async_id(async_id);
682+
683+
try {
684+
const proto::FfiResponse resp = sendRequest(req);
685+
if (!resp.has_simulate_scenario()) {
686+
logAndThrow("FfiResponse missing simulate_scenario");
687+
}
688+
} catch (...) {
689+
cancelPendingByAsyncId(async_id);
690+
throw;
691+
}
692+
693+
return fut;
694+
}
695+
658696
// Participant APIs Implementation
659697
std::future<proto::OwnedTrackPublication> FfiClient::publishTrackAsync(std::uint64_t local_participant_handle,
660698
std::uint64_t track_handle,

‎src/ffi_client.h‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -106,6 +106,8 @@ class LIVEKIT_INTERNAL_API FfiClient {
106106

107107
std::future<SessionStats> getSessionStatsAsync(uintptr_t room_handle);
108108

109+
std::future<void> simulateScenarioAsync(uintptr_t room_handle, int scenario);
110+
109111
// Participant APIs
110112
std::future<proto::OwnedTrackPublication> publishTrackAsync(std::uint64_t local_participant_handle,
111113
std::uint64_t track_handle,

‎src/room.cpp‎

Lines changed: 62 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,10 @@
2020
#include "ffi_client.h"
2121
#include "livekit/audio_stream.h"
2222
#include "livekit/e2ee.h"
23+
#include "livekit/local_audio_track.h"
2324
#include "livekit/local_participant.h"
25+
#include "livekit/local_track_publication.h"
26+
#include "livekit/local_video_track.h"
2427
#include "livekit/remote_audio_track.h"
2528
#include "livekit/remote_data_track.h"
2629
#include "livekit/remote_participant.h"
@@ -47,6 +50,33 @@ using proto::FfiResponse;
4750

4851
namespace {
4952

53+
std::shared_ptr<LocalTrackPublication> localTrackPublication(const std::shared_ptr<Track>& track) {
54+
if (!track) {
55+
return nullptr;
56+
}
57+
if (auto video = std::dynamic_pointer_cast<LocalVideoTrack>(track)) {
58+
return video->publication();
59+
}
60+
if (auto audio = std::dynamic_pointer_cast<LocalAudioTrack>(track)) {
61+
return audio->publication();
62+
}
63+
return nullptr;
64+
}
65+
66+
void updateLocalTrackPublicationInfo(LocalTrackPublication& publication, const proto::TrackPublicationInfo& info) {
67+
publication.sid_ = info.sid();
68+
publication.name_ = info.name();
69+
publication.kind_ = fromProto(info.kind());
70+
publication.source_ = fromProto(info.source());
71+
publication.simulcasted_ = info.simulcasted();
72+
publication.width_ = info.width();
73+
publication.height_ = info.height();
74+
publication.mime_type_ = info.mime_type();
75+
publication.muted_ = info.muted();
76+
publication.encryption_type_ = static_cast<EncryptionType>(info.encryption_type());
77+
publication.audio_features_ = convertAudioFeatures(info.audio_features());
78+
}
79+
5080
std::shared_ptr<livekit::RemoteParticipant> createRemoteParticipant(const proto::OwnedParticipant& owned) {
5181
const auto& pinfo = owned.info();
5282
std::unordered_map<std::string, std::string> attrs;
@@ -593,6 +623,38 @@ void Room::onEvent(const FfiEvent& event) {
593623
}
594624
break;
595625
}
626+
case proto::RoomEvent::kLocalTrackRepublished: {
627+
const std::scoped_lock<std::mutex> guard(lock_);
628+
if (!local_participant_) {
629+
LK_LOG_ERROR("kLocalTrackRepublished: local_participant_ is nullptr");
630+
break;
631+
}
632+
const auto& ltr = re.local_track_republished();
633+
const std::string& previous_sid = ltr.previous_sid();
634+
635+
auto& published = local_participant_->published_tracks_by_sid_;
636+
auto it = published.find(previous_sid);
637+
if (it == published.end()) {
638+
LK_LOG_WARN("local_track_republished for unknown previous sid: {}", previous_sid);
639+
break;
640+
}
641+
auto track = it->second.lock();
642+
if (!track) {
643+
published.erase(it);
644+
LK_LOG_WARN("local_track_republished for expired previous sid: {}", previous_sid);
645+
break;
646+
}
647+
auto publication = localTrackPublication(track);
648+
if (!publication) {
649+
LK_LOG_WARN("local_track_republished missing publication for sid: {}", previous_sid);
650+
break;
651+
}
652+
updateLocalTrackPublicationInfo(*publication, ltr.info());
653+
published.erase(it);
654+
published[publication->sid()] = track;
655+
track->setPublication(publication);
656+
break;
657+
}
596658
case proto::RoomEvent::kLocalTrackSubscribed: {
597659
LocalTrackSubscribedEvent ev;
598660
{
Lines changed: 49 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,49 @@
1+
/*
2+
* Copyright 2026 LiveKit
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
#pragma once
18+
19+
#include <livekit/livekit.h>
20+
21+
#include <memory>
22+
#include <mutex>
23+
#include <stdexcept>
24+
25+
#include "ffi_client.h"
26+
#include "room.pb.h"
27+
28+
namespace livekit {
29+
30+
struct RoomTestAccess {
31+
static int listenerId(const Room& room) {
32+
const std::scoped_lock<std::mutex> guard(room.lock_);
33+
return room.listener_id_;
34+
}
35+
36+
static void simulateScenario(Room& room, proto::SimulateScenarioKind scenario) {
37+
std::shared_ptr<FfiHandle> handle;
38+
{
39+
const std::scoped_lock<std::mutex> guard(room.lock_);
40+
handle = room.room_handle_;
41+
}
42+
if (!handle) {
43+
throw std::runtime_error("cannot simulate scenario for a disconnected room");
44+
}
45+
FfiClient::instance().simulateScenarioAsync(handle->get(), static_cast<int>(scenario)).get();
46+
}
47+
};
48+
49+
} // namespace livekit

‎src/tests/integration/test_local_track_publish_sid.cpp‎

Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,15 +16,21 @@
1616

1717
#include <gtest/gtest.h>
1818

19+
#include <chrono>
1920
#include <memory>
2021
#include <string>
22+
#include <thread>
2123

2224
#include "../common/audio_utils.h"
25+
#include "../common/room_test_access.h"
2326
#include "../common/test_common.h"
27+
#include "room.pb.h"
2428

2529
namespace livekit::test {
2630
namespace {
2731

32+
using namespace std::chrono_literals;
33+
2834
void expectTrackSidAssigned(const Track& track, const LocalTrackPublication& publication) {
2935
const std::string& track_sid = track.sid();
3036
const std::string& publication_sid = publication.sid();
@@ -33,6 +39,17 @@ void expectTrackSidAssigned(const Track& track, const LocalTrackPublication& pub
3339
EXPECT_EQ(track_sid, publication_sid);
3440
}
3541

42+
bool waitForSidChange(const Track& track, const std::string& previous_sid, std::chrono::milliseconds timeout) {
43+
const auto deadline = std::chrono::steady_clock::now() + timeout;
44+
while (std::chrono::steady_clock::now() < deadline) {
45+
if (track.sid() != previous_sid && track.sid() != "TR_unknown") {
46+
return true;
47+
}
48+
std::this_thread::sleep_for(50ms);
49+
}
50+
return track.sid() != previous_sid && track.sid() != "TR_unknown";
51+
}
52+
3653
} // namespace
3754

3855
class LocalTrackPublishSidTest : public LiveKitTestBase {};
@@ -73,4 +90,34 @@ TEST_F(LocalTrackPublishSidTest, PublishAudioTrackAssignsSid) {
7390
lockLocalParticipant(room)->unpublishTrack(track->publication()->sid());
7491
}
7592

93+
TEST_F(LocalTrackPublishSidTest, FullReconnectUpdatesPublishedSid) {
94+
failIfNotConfigured();
95+
96+
Room room;
97+
const RoomOptions room_options;
98+
ASSERT_TRUE(room.connect(config_.url, config_.token_a, room_options));
99+
100+
auto source = std::make_shared<EncodedVideoSource>(VideoCodec::H264, 16, 16);
101+
std::shared_ptr<LocalVideoTrack> track;
102+
ASSERT_NO_THROW(
103+
track = lockLocalParticipant(room)->publishVideoTrack("republish-sid-check", source, TrackSource::SOURCE_CAMERA));
104+
ASSERT_NE(track, nullptr);
105+
ASSERT_NE(track->publication(), nullptr);
106+
expectTrackSidAssigned(*track, *track->publication());
107+
108+
const std::string previous_sid = track->sid();
109+
ASSERT_NO_THROW(RoomTestAccess::simulateScenario(room, proto::SIMULATE_FULL_RECONNECT));
110+
ASSERT_TRUE(waitForSidChange(*track, previous_sid, 30s)) << "Timed out waiting for republished track SID";
111+
112+
ASSERT_NE(track->publication(), nullptr);
113+
expectTrackSidAssigned(*track, *track->publication());
114+
EXPECT_NE(track->sid(), previous_sid);
115+
116+
const auto pubs = lockLocalParticipant(room)->trackPublications();
117+
EXPECT_EQ(pubs.count(previous_sid), 0u);
118+
EXPECT_EQ(pubs.count(track->sid()), 1u);
119+
120+
lockLocalParticipant(room)->unpublishTrack(track->sid());
121+
}
122+
76123
} // namespace livekit::test

‎src/tests/integration/test_room.cpp‎

Lines changed: 1 addition & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -29,21 +29,11 @@
2929
#include <thread>
3030
#include <vector>
3131

32+
#include "../common/room_test_access.h"
3233
#include "../common/test_common.h"
3334

3435
using namespace std::chrono_literals;
3536

36-
namespace livekit {
37-
38-
struct RoomTestAccess {
39-
static int listenerId(const Room& room) {
40-
const std::scoped_lock<std::mutex> guard(room.lock_);
41-
return room.listener_id_;
42-
}
43-
};
44-
45-
} // namespace livekit
46-
4737
namespace livekit::test {
4838

4939
// Server-dependent tests - require LIVEKIT_URL and LIVEKIT_TOKEN_A env vars

0 commit comments

Comments
 (0)