Fix local track publication event ordering

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: alan-george-lk

include/livekit/local_participant.h

@@ -296,6 +296,7 @@ class LIVEKIT_API LocalParticipant : public Participant {
   /// 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 std::mutex published_tracks_mutex_;
   mutable TrackMap published_tracks_by_sid_;
 
   std::unordered_map<std::string, RpcHandler> rpc_handlers_;

src/ffi_client.cpp

@@ -19,6 +19,7 @@
 #include <cassert>
 #include <csignal>
 #include <cstdio>
+#include <exception>
 #include <string>
 #include <type_traits>
 
@@ -694,9 +695,9 @@ std::future<void> FfiClient::simulateScenarioAsync(uintptr_t room_handle, int sc
 }
 
 // Participant APIs Implementation
-std::future<proto::OwnedTrackPublication> FfiClient::publishTrackAsync(std::uint64_t local_participant_handle,
-                                                                       std::uint64_t track_handle,
-                                                                       const TrackPublishOptions& options) {
+std::future<proto::OwnedTrackPublication> FfiClient::publishTrackAsync(
+    std::uint64_t local_participant_handle, std::uint64_t track_handle, const TrackPublishOptions& options,
+    std::function<void(const proto::OwnedTrackPublication&)> on_success) {
   // Generate client-side async_id first
   const AsyncId async_id = generateAsyncId();
 
@@ -708,7 +709,8 @@ std::future<proto::OwnedTrackPublication> FfiClient::publishTrackAsync(std::uint
         return event.has_publish_track() && event.publish_track().async_id() == async_id;
       },
       // Handler: resolve with publication or throw error
-      [](const proto::FfiEvent& event, std::promise<proto::OwnedTrackPublication>& pr) {
+      [on_success = std::move(on_success)](const proto::FfiEvent& event,
+                                           std::promise<proto::OwnedTrackPublication>& pr) {
         const auto& cb = event.publish_track();
 
         // Oneof message { string error = 2; OwnedTrackPublication publication =
@@ -723,7 +725,14 @@ std::future<proto::OwnedTrackPublication> FfiClient::publishTrackAsync(std::uint
         }
 
         const proto::OwnedTrackPublication& pub = cb.publication();
-        pr.set_value(pub);
+        try {
+          if (on_success) {
+            on_success(pub);
+          }
+          pr.set_value(pub);
+        } catch (...) {
+          pr.set_exception(std::current_exception());
+        }
       });
 
   // Build and send the request

src/ffi_client.h

@@ -109,9 +109,9 @@ class LIVEKIT_INTERNAL_API FfiClient {
   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,
-                                                              const TrackPublishOptions& options);
+  std::future<proto::OwnedTrackPublication> publishTrackAsync(
+      std::uint64_t local_participant_handle, std::uint64_t track_handle, const TrackPublishOptions& options,
+      std::function<void(const proto::OwnedTrackPublication&)> on_success = {});
   std::future<void> unpublishTrackAsync(std::uint64_t local_participant_handle, const std::string& track_sid,
                                         bool stop_on_unpublish);
   std::future<void> publishDataAsync(std::uint64_t local_participant_handle, const std::uint8_t* data_ptr,

src/local_participant.cpp

@@ -190,19 +190,18 @@ void LocalParticipant::publishTrack(const std::shared_ptr<Track>& track, const T
   if (track_handle == 0) {
     throw std::runtime_error("LocalParticipant::publishTrack: invalid track FFI handle");
   }
-  auto fut = FfiClient::instance().publishTrackAsync(static_cast<std::uint64_t>(participant_handle),
-                                                     static_cast<std::uint64_t>(track_handle), options);
-
-  // Will throw if the async op fails (error in callback).
-  const proto::OwnedTrackPublication owned_pub = fut.get();
-
-  // Construct a LocalTrackPublication from the proto publication.
-  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);
+  auto fut = FfiClient::instance().publishTrackAsync(
+      static_cast<std::uint64_t>(participant_handle), static_cast<std::uint64_t>(track_handle), options,
+      [this, track](const proto::OwnedTrackPublication& owned_pub) {
+        auto publication = std::make_shared<LocalTrackPublication>(owned_pub);
+        const std::scoped_lock<std::mutex> guard(published_tracks_mutex_);
+        published_tracks_by_sid_[publication->sid()] = track;
+        track->setPublication(publication);
+      });
+
+  // Publication state is installed on the FFI thread before this future is
+  // completed and before Rust can emit LocalTrackPublished.
+  (void)fut.get();
 }
 
 std::shared_ptr<LocalVideoTrack> LocalParticipant::publishVideoTrack(const std::string& name,
@@ -248,6 +247,7 @@ void LocalParticipant::unpublishTrack(const std::string& track_sid) {
 
   fut.get();
 
+  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);
@@ -257,6 +257,7 @@ void LocalParticipant::unpublishTrack(const std::string& track_sid) {
 }
 
 LocalParticipant::PublicationMap LocalParticipant::trackPublications() const {
+  const std::scoped_lock<std::mutex> guard(published_tracks_mutex_);
   PublicationMap out;
   for (auto it = published_tracks_by_sid_.begin(); it != published_tracks_by_sid_.end();) {
     auto t = it->second.lock();
@@ -494,6 +495,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;

src/room.cpp

@@ -63,20 +63,6 @@ std::shared_ptr<LocalTrackPublication> 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<EncryptionType>(info.encryption_type());
-  publication.audio_features_ = convertAudioFeatures(info.audio_features());
-}
-
 std::shared_ptr<livekit::RemoteParticipant> createRemoteParticipant(const proto::OwnedParticipant& owned) {
   const auto& pinfo = owned.info();
   std::unordered_map<std::string, std::string> attrs;
@@ -632,6 +618,7 @@ void Room::onEvent(const FfiEvent& event) {
           const auto& ltr = re.local_track_republished();
           const std::string& previous_sid = ltr.previous_sid();
 
+          const std::scoped_lock<std::mutex> publications_guard(local_participant_->published_tracks_mutex_);
           auto& published = local_participant_->published_tracks_by_sid_;
           auto it = published.find(previous_sid);
           if (it == published.end()) {
@@ -649,7 +636,18 @@ void Room::onEvent(const FfiEvent& event) {
             LK_LOG_WARN("local_track_republished missing publication for sid: {}", previous_sid);
             break;
           }
-          updateLocalTrackPublicationInfo(*publication, ltr.info());
+          const auto& info = ltr.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<EncryptionType>(info.encryption_type());
+          publication->audio_features_ = convertAudioFeatures(info.audio_features());
           published.erase(it);
           published[publication->sid()] = track;
           track->setPublication(publication);

src/tests/integration/test_local_track_publish_sid.cpp

@@ -17,9 +17,12 @@
 #include <gtest/gtest.h>
 
 #include <chrono>
+#include <condition_variable>
 #include <memory>
+#include <mutex>
 #include <string>
 #include <thread>
+#include <unordered_set>
 
 #include "../common/audio_utils.h"
 #include "../common/room_test_access.h"
@@ -31,6 +34,30 @@ namespace {
 
 using namespace std::chrono_literals;
 
+class LocalPublicationDelegate final : public RoomDelegate {
+public:
+  void onLocalTrackPublished(Room&, const LocalTrackPublishedEvent& event) override {
+    if (!event.publication) {
+      return;
+    }
+    {
+      const std::scoped_lock<std::mutex> guard(mutex_);
+      published_sids_.insert(event.publication->sid());
+    }
+    cv_.notify_all();
+  }
+
+  bool waitForPublished(const std::string& sid, std::chrono::milliseconds timeout) {
+    std::unique_lock<std::mutex> lock(mutex_);
+    return cv_.wait_for(lock, timeout, [&] { return published_sids_.count(sid) != 0; });
+  }
+
+private:
+  std::mutex mutex_;
+  std::condition_variable cv_;
+  std::unordered_set<std::string> published_sids_;
+};
+
 void expectTrackSidAssigned(const Track& track, const LocalTrackPublication& publication) {
   const std::string& track_sid = track.sid();
   const std::string& publication_sid = publication.sid();
@@ -58,6 +85,8 @@ TEST_F(LocalTrackPublishSidTest, PublishVideoTrackAssignsSid) {
   failIfNotConfigured();
 
   Room room;
+  LocalPublicationDelegate delegate;
+  room.setDelegate(&delegate);
   const RoomOptions room_options;
   ASSERT_TRUE(room.connect(config_.url, config_.token_a, room_options));
 
@@ -69,13 +98,16 @@ TEST_F(LocalTrackPublishSidTest, PublishVideoTrackAssignsSid) {
   ASSERT_NE(track->publication(), nullptr);
 
   expectTrackSidAssigned(*track, *track->publication());
+  EXPECT_TRUE(delegate.waitForPublished(track->sid(), 5s));
   lockLocalParticipant(room)->unpublishTrack(track->publication()->sid());
 }
 
 TEST_F(LocalTrackPublishSidTest, PublishAudioTrackAssignsSid) {
   failIfNotConfigured();
 
   Room room;
+  LocalPublicationDelegate delegate;
+  room.setDelegate(&delegate);
   const RoomOptions room_options;
   ASSERT_TRUE(room.connect(config_.url, config_.token_a, room_options));
 
@@ -87,6 +119,7 @@ TEST_F(LocalTrackPublishSidTest, PublishAudioTrackAssignsSid) {
   ASSERT_NE(track->publication(), nullptr);
 
   expectTrackSidAssigned(*track, *track->publication());
+  EXPECT_TRUE(delegate.waitForPublished(track->sid(), 5s));
   lockLocalParticipant(room)->unpublishTrack(track->publication()->sid());
 }