Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions include/livekit/local_participant.h
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
9 changes: 8 additions & 1 deletion include/livekit/local_track_publication.h
Original file line number Diff line number Diff line change
Expand Up @@ -23,14 +23,21 @@ namespace livekit {

namespace proto {
class OwnedTrackPublication;
}
class TrackPublicationInfo;
} // namespace proto

/// @brief A track published by a local participant.
class LIVEKIT_API LocalTrackPublication : public TrackPublication {
public:
/// Note, this LocalTrackPublication is constructed internally only;
/// safe to accept proto::OwnedTrackPublication.
explicit LocalTrackPublication(const proto::OwnedTrackPublication& owned);

private:
friend class Room;

/// Refresh publication metadata after a full reconnect republish.
void updateInfo(const proto::TrackPublicationInfo& info);
};

} // namespace livekit
38 changes: 38 additions & 0 deletions src/ffi_client.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -103,6 +103,8 @@ std::optional<FfiClient::AsyncId> 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:
Expand Down Expand Up @@ -655,6 +657,42 @@ std::future<SessionStats> FfiClient::getSessionStatsAsync(uintptr_t room_handle)
return fut;
}

std::future<void> FfiClient::simulateScenarioAsync(uintptr_t room_handle, int scenario) {
const AsyncId async_id = generateAsyncId();

auto fut = registerAsync<void>(
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<void>& 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<proto::SimulateScenarioKind>(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<proto::OwnedTrackPublication> FfiClient::publishTrackAsync(std::uint64_t local_participant_handle,
std::uint64_t track_handle,
Expand Down
2 changes: 2 additions & 0 deletions src/ffi_client.h
Original file line number Diff line number Diff line change
Expand Up @@ -106,6 +106,8 @@ class LIVEKIT_INTERNAL_API FfiClient {

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

std::future<void> simulateScenarioAsync(uintptr_t room_handle, int scenario);

// Participant APIs
std::future<proto::OwnedTrackPublication> publishTrackAsync(std::uint64_t local_participant_handle,
std::uint64_t track_handle,
Expand Down
22 changes: 15 additions & 7 deletions src/local_participant.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@

#include <chrono>
#include <memory>
#include <mutex>
#include <stdexcept>

#include "data_track.pb.h"
Expand Down Expand Up @@ -200,9 +201,11 @@ void LocalParticipant::publishTrack(const std::shared_ptr<Track>& track, const T
auto publication = std::make_shared<LocalTrackPublication>(owned_pub);

const std::string sid = publication->sid();
published_tracks_by_sid_[sid] = std::weak_ptr<Track>(track);

track->setPublication(publication);
{
const std::scoped_lock<std::mutex> guard(published_tracks_mutex_);
published_tracks_by_sid_[sid] = std::weak_ptr<Track>(track);
track->setPublication(publication);
}
}

std::shared_ptr<LocalVideoTrack> LocalParticipant::publishVideoTrack(const std::string& name,
Expand Down Expand Up @@ -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<std::mutex> 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<std::mutex> 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) {
Expand Down Expand Up @@ -494,6 +501,7 @@ void LocalParticipant::handleRpcMethodInvocation(uint64_t invocation_id, const s
}

std::shared_ptr<TrackPublication> LocalParticipant::findTrackPublication(const std::string& sid) const {
const std::scoped_lock<std::mutex> guard(published_tracks_mutex_);
auto it = published_tracks_by_sid_.find(sid);
if (it == published_tracks_by_sid_.end()) {
return nullptr;
Expand Down
15 changes: 15 additions & 0 deletions src/local_track_publication.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@

#include "livekit/local_track_publication.h"

#include "track.pb.h"
#include "track_proto_converter.h"

namespace livekit {
Expand All @@ -27,4 +28,18 @@ LocalTrackPublication::LocalTrackPublication(const proto::OwnedTrackPublication&
static_cast<EncryptionType>(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<EncryptionType>(info.encryption_type());
audio_features_ = convertAudioFeatures(info.audio_features());
}

} // namespace livekit
49 changes: 49 additions & 0 deletions src/room.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand All @@ -47,6 +50,19 @@ using proto::FfiResponse;

namespace {

std::shared_ptr<LocalTrackPublication> localTrackPublication(const std::shared_ptr<Track>& track) {
if (!track) {
return nullptr;
}
if (auto video = std::dynamic_pointer_cast<LocalVideoTrack>(track)) {
return video->publication();
}
if (auto audio = std::dynamic_pointer_cast<LocalAudioTrack>(track)) {
return audio->publication();
}
return nullptr;
}

std::shared_ptr<livekit::RemoteParticipant> createRemoteParticipant(const proto::OwnedParticipant& owned) {
const auto& pinfo = owned.info();
std::unordered_map<std::string, std::string> attrs;
Expand Down Expand Up @@ -593,6 +609,39 @@ void Room::onEvent(const FfiEvent& event) {
}
break;
}
case proto::RoomEvent::kLocalTrackRepublished: {
const std::scoped_lock<std::mutex> guard(lock_);
if (!local_participant_) {
LK_LOG_ERROR("kLocalTrackRepublished: local_participant_ is nullptr");
break;
}
const auto& local_track_repub = re.local_track_republished();
const std::string& previous_sid = local_track_repub.previous_sid();

const std::scoped_lock<std::mutex> 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()) {
LK_LOG_WARN("local_track_republished for unknown previous sid: {}", previous_sid);
break;
}
auto track = it->second.lock();
if (!track) {
published_tracks.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;
}
publication->updateInfo(local_track_repub.info());
published_tracks.erase(it);
published_tracks[publication->sid()] = track;
track->setPublication(publication);
break;
}
case proto::RoomEvent::kLocalTrackSubscribed: {
LocalTrackSubscribedEvent ev;
{
Expand Down
49 changes: 49 additions & 0 deletions src/tests/common/room_test_access.h
Original file line number Diff line number Diff line change
@@ -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 <livekit/livekit.h>

#include <memory>
#include <mutex>
#include <stdexcept>

#include "ffi_client.h"
#include "room.pb.h"

namespace livekit {

struct RoomTestAccess {
static int listenerId(const Room& room) {
const std::scoped_lock<std::mutex> guard(room.lock_);
return room.listener_id_;
}

static void simulateScenario(Room& room, proto::SimulateScenarioKind scenario) {
Comment thread
ladvoc marked this conversation as resolved.
std::shared_ptr<FfiHandle> handle;
{
const std::scoped_lock<std::mutex> 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<int>(scenario)).get();
}
};

} // namespace livekit
47 changes: 47 additions & 0 deletions src/tests/integration/test_local_track_publish_sid.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -16,15 +16,21 @@

#include <gtest/gtest.h>

#include <chrono>
#include <memory>
#include <string>
#include <thread>

#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();
Expand All @@ -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 {};
Expand Down Expand Up @@ -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<EncodedVideoSource>(VideoCodec::H264, 16, 16);
std::shared_ptr<LocalVideoTrack> 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
12 changes: 1 addition & 11 deletions src/tests/integration/test_room.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -29,21 +29,11 @@
#include <thread>
#include <vector>

#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<std::mutex> guard(room.lock_);
return room.listener_id_;
}
};

} // namespace livekit

namespace livekit::test {

// Server-dependent tests - require LIVEKIT_URL and LIVEKIT_TOKEN_A env vars
Expand Down
Loading