From 045ac22ab738bdcde8871b607b9f0f54dac62292 Mon Sep 17 00:00:00 2001 From: Alan George Date: Thu, 20 Aug 2026 10:46:10 -0600 Subject: [PATCH 1/6] Add tryPush for pointer/size to avoid a copy --- include/livekit/local_data_track.h | 18 ++++ src/local_data_track.cpp | 34 ++++--- src/tests/integration/test_data_track.cpp | 110 ++++++++++++++++++++++ 3 files changed, 151 insertions(+), 11 deletions(-) diff --git a/include/livekit/local_data_track.h b/include/livekit/local_data_track.h index 18c2eb77..5563c2ab 100644 --- a/include/livekit/local_data_track.h +++ b/include/livekit/local_data_track.h @@ -78,6 +78,24 @@ class LIVEKIT_API LocalDataTrack { Result tryPush(std::vector&& payload, std::optional user_timestamp = std::nullopt); + /// Try to push a frame from a borrowed byte buffer. + /// + /// Copies @p size bytes from @p data into an FFI request before returning; + /// the SDK does not retain the buffer. This avoids an intermediate + /// DataTrackFrame or std::vector copy when the caller already owns a byte + /// buffer, but it is not a zero-copy send. C++17-friendly equivalent of a + /// span overload. + /// + /// @param data Pointer to @p size payload bytes. Must be non-null. + /// @param size Number of bytes at @p data. Must be non-zero. + /// @param user_timestamp Optional application-defined timestamp. The unit is + /// caller-defined; SDK examples use microseconds since the Unix epoch. + /// @return success on delivery acceptance, or a typed error describing why + /// the frame could not be queued. A null @p data or zero @p size + /// returns @ref LocalDataTrackTryPushErrorCode::INTERNAL. + Result tryPush(const std::uint8_t* data, std::size_t size, + std::optional user_timestamp = std::nullopt); + /// Whether the track is still published in the room. bool isPublished() const; diff --git a/src/local_data_track.cpp b/src/local_data_track.cpp index 5986e283..9f3055f8 100644 --- a/src/local_data_track.cpp +++ b/src/local_data_track.cpp @@ -29,19 +29,39 @@ LocalDataTrack::LocalDataTrack(const proto::OwnedLocalDataTrack& owned) : handle_(static_cast(owned.handle().id())), info_(fromProto(owned.info())) {} Result LocalDataTrack::tryPush(const DataTrackFrame& frame) { + return tryPush(frame.payload.data(), frame.payload.size(), frame.user_timestamp); +} + +Result LocalDataTrack::tryPush(std::vector&& payload, + std::optional user_timestamp) { + const DataTrackFrame frame(std::move(payload), user_timestamp); + return tryPush(frame); +} + +Result LocalDataTrack::tryPush(const std::uint8_t* data, std::size_t size, + std::optional user_timestamp) { if (!handle_.valid()) { return Result::failure(LocalDataTrackTryPushError{ LocalDataTrackTryPushErrorCode::INVALID_HANDLE, "LocalDataTrack::tryPush: invalid FFI handle"}); } + if (size == 0) { + return Result::failure( + LocalDataTrackTryPushError{LocalDataTrackTryPushErrorCode::INTERNAL, "LocalDataTrack::tryPush: empty size"}); + + } else if (data == nullptr) { + return Result::failure(LocalDataTrackTryPushError{ + LocalDataTrackTryPushErrorCode::INTERNAL, "LocalDataTrack::tryPush: payload pointer is null"}); + } try { proto::FfiRequest req; auto* msg = req.mutable_local_data_track_try_push(); msg->set_track_handle(static_cast(handle_.get())); auto* pf = msg->mutable_frame(); - pf->set_payload(frame.payload.data(), frame.payload.size()); - if (frame.user_timestamp.has_value()) { - pf->set_user_timestamp(frame.user_timestamp.value()); + // Size and data are checked above + pf->set_payload(data, size); + if (user_timestamp.has_value()) { + pf->set_user_timestamp(user_timestamp.value()); } const proto::FfiResponse resp = FfiClient::instance().sendRequest(req); @@ -56,14 +76,6 @@ Result LocalDataTrack::tryPush(const DataTrack } } -Result LocalDataTrack::tryPush(std::vector&& payload, - std::optional user_timestamp) { - DataTrackFrame frame; - frame.payload = std::move(payload); - frame.user_timestamp = user_timestamp; - return tryPush(frame); -} - bool LocalDataTrack::isPublished() const { if (!handle_.valid()) { return false; diff --git a/src/tests/integration/test_data_track.cpp b/src/tests/integration/test_data_track.cpp index 89690ab9..6034b2c8 100644 --- a/src/tests/integration/test_data_track.cpp +++ b/src/tests/integration/test_data_track.cpp @@ -962,6 +962,116 @@ TEST_F(DataTrackE2ETest, PreservesUserTimestampEndToEnd) { EXPECT_EQ(frame.user_timestamp.value(), sent_timestamp); } +TEST_F(DataTrackE2ETest, RejectsNullBorrowedPayloadWithNonZeroSize) { + const auto track_name = makeTrackName("null_borrowed_payload"); + auto rooms = testRooms(1); + auto local_track = requirePublishedTrack(rooms[0]->localParticipant(), track_name); + + const auto push_result = local_track->tryPush(static_cast(nullptr), 4); + ASSERT_FALSE(push_result); + EXPECT_EQ(push_result.error().code, LocalDataTrackTryPushErrorCode::INTERNAL); + EXPECT_FALSE(push_result.error().message.empty()); + + local_track->unpublishDataTrack(); +} + +TEST_F(DataTrackE2ETest, RejectsEmptyBorrowedPayload) { + const auto track_name = makeTrackName("empty_borrowed_payload"); + auto rooms = testRooms(1); + auto local_track = requirePublishedTrack(rooms[0]->localParticipant(), track_name); + + const std::uint8_t payload = 0; + const auto push_result = local_track->tryPush(&payload, 0); + ASSERT_FALSE(push_result); + EXPECT_EQ(push_result.error().code, LocalDataTrackTryPushErrorCode::INTERNAL); + EXPECT_FALSE(push_result.error().message.empty()); + + local_track->unpublishDataTrack(); +} + +TEST_F(DataTrackE2ETest, CopiesBorrowedPayloadBeforeTryPushReturns) { + const auto track_name = makeTrackName("borrowed_payload"); + const auto sent_timestamp = getTimestampUs(); + constexpr std::uint8_t kOriginal = 0xA5; + constexpr std::uint8_t kMutated = 0x5A; + constexpr std::size_t kPayloadSize = 64; + + DataTrackPublishedDelegate subscriber_delegate; + std::vector room_configs(2); + room_configs[1].delegate = &subscriber_delegate; + + auto rooms = testRooms(room_configs); + auto& publisher_room = rooms[0]; + + auto publish_result = lockLocalParticipant(*publisher_room)->publishDataTrack(track_name); + if (!publish_result) { + FAIL() << describeDataTrackError(publish_result.error()); + } + auto local_track = publish_result.value(); + ASSERT_TRUE(local_track->isPublished()); + + auto remote_track = subscriber_delegate.waitForTrack(kTrackWaitTimeout); + ASSERT_NE(remote_track, nullptr) << "Timed out waiting for remote data track"; + + auto subscribe_result = remote_track->subscribe(); + if (!subscribe_result) { + FAIL() << describeDataTrackError(subscribe_result.error()); + } + auto subscription = subscribe_result.value(); + + std::promise frame_promise; + auto frame_future = frame_promise.get_future(); + std::thread reader([&]() { + try { + DataTrackFrame frame; + if (!subscription->read(frame)) { + throw std::runtime_error("Subscription ended before borrowed-payload frame arrived"); + } + frame_promise.set_value(std::move(frame)); + } catch (...) { + frame_promise.set_exception(std::current_exception()); + } + }); + + bool pushed = false; + for (int attempt = 0; attempt < kTimestampFrameAttempts; ++attempt) { + std::array payload{}; + payload.fill(kOriginal); + auto push_result = local_track->tryPush(payload.data(), payload.size(), sent_timestamp); + payload.fill(kMutated); + pushed = static_cast(push_result) || pushed; + if (frame_future.wait_for(25ms) == std::future_status::ready) { + break; + } + } + const auto frame_status = frame_future.wait_for(5s); + + if (frame_status != std::future_status::ready) { + subscription->close(); + } + + subscription->close(); + reader.join(); + local_track->unpublishDataTrack(); + + ASSERT_TRUE(pushed) << "Failed to push borrowed data frame"; + ASSERT_EQ(frame_status, std::future_status::ready) << "Timed out waiting for borrowed-payload frame"; + + DataTrackFrame frame; + try { + frame = frame_future.get(); + } catch (const std::exception& e) { + FAIL() << e.what(); + } + + ASSERT_EQ(frame.payload.size(), kPayloadSize); + EXPECT_TRUE(std::all_of(frame.payload.begin(), frame.payload.end(), [](std::uint8_t byte) { + return byte == kOriginal; + })) << "Received payload reflects caller mutation after tryPush returned"; + ASSERT_TRUE(frame.user_timestamp.has_value()); + EXPECT_EQ(frame.user_timestamp.value(), sent_timestamp); +} + TEST_F(DataTrackE2ETest, PublishesAndReceivesEncryptedFramesEndToEnd) { runEncryptedDataTrackRoundTrip(kDefaultKeyDerivationFunction, "e2ee_transport"); } From f3e3856bc9e9511db45dfd82a8976f9f6286e132 Mon Sep 17 00:00:00 2001 From: Alan George Date: Thu, 20 Aug 2026 10:59:12 -0600 Subject: [PATCH 2/6] Remove ref --- include/livekit/local_data_track.h | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/include/livekit/local_data_track.h b/include/livekit/local_data_track.h index 5563c2ab..e00078e7 100644 --- a/include/livekit/local_data_track.h +++ b/include/livekit/local_data_track.h @@ -91,8 +91,7 @@ class LIVEKIT_API LocalDataTrack { /// @param user_timestamp Optional application-defined timestamp. The unit is /// caller-defined; SDK examples use microseconds since the Unix epoch. /// @return success on delivery acceptance, or a typed error describing why - /// the frame could not be queued. A null @p data or zero @p size - /// returns @ref LocalDataTrackTryPushErrorCode::INTERNAL. + /// the frame could not be queued. Result tryPush(const std::uint8_t* data, std::size_t size, std::optional user_timestamp = std::nullopt); From b52066b3c1d340cd26162393fdb7fba43e899574 Mon Sep 17 00:00:00 2001 From: Alan George Date: Thu, 20 Aug 2026 15:40:24 -0600 Subject: [PATCH 3/6] Maybe fix windows build --- src/tests/integration/test_data_track.cpp | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/tests/integration/test_data_track.cpp b/src/tests/integration/test_data_track.cpp index 6034b2c8..0514cdae 100644 --- a/src/tests/integration/test_data_track.cpp +++ b/src/tests/integration/test_data_track.cpp @@ -1065,8 +1065,8 @@ TEST_F(DataTrackE2ETest, CopiesBorrowedPayloadBeforeTryPushReturns) { } ASSERT_EQ(frame.payload.size(), kPayloadSize); - EXPECT_TRUE(std::all_of(frame.payload.begin(), frame.payload.end(), [](std::uint8_t byte) { - return byte == kOriginal; + EXPECT_TRUE(std::all_of(frame.payload.begin(), frame.payload.end(), [expected = kOriginal](std::uint8_t byte) { + return byte == expected; })) << "Received payload reflects caller mutation after tryPush returned"; ASSERT_TRUE(frame.user_timestamp.has_value()); EXPECT_EQ(frame.user_timestamp.value(), sent_timestamp); From 30b147c3aa5b34e4b8fc6949db3e6ce6f827100c Mon Sep 17 00:00:00 2001 From: Alan George Date: Fri, 21 Aug 2026 11:35:15 -0600 Subject: [PATCH 4/6] Update comments --- include/livekit/local_data_track.h | 20 ++++++++++++-------- 1 file changed, 12 insertions(+), 8 deletions(-) diff --git a/include/livekit/local_data_track.h b/include/livekit/local_data_track.h index e00078e7..d6029499 100644 --- a/include/livekit/local_data_track.h +++ b/include/livekit/local_data_track.h @@ -67,13 +67,17 @@ class LIVEKIT_API LocalDataTrack { /// Try to push a frame to all subscribers of this track. /// - /// @return success on delivery acceptance, or a typed error describing why + /// @param frame The frame to push. + /// @return Success on delivery acceptance, or a typed error describing why /// the frame could not be queued. Result tryPush(const DataTrackFrame& frame); /// Try to push a frame to all subscribers of this track. /// - /// @return success on delivery acceptance, or a typed error describing why + /// @param payload The payload to push. + /// @param user_timestamp Optional application-defined timestamp. The unit is + /// caller-defined. + /// @return Success on delivery acceptance, or a typed error describing why /// the frame could not be queued. Result tryPush(std::vector&& payload, std::optional user_timestamp = std::nullopt); @@ -81,16 +85,16 @@ class LIVEKIT_API LocalDataTrack { /// Try to push a frame from a borrowed byte buffer. /// /// Copies @p size bytes from @p data into an FFI request before returning; - /// the SDK does not retain the buffer. This avoids an intermediate - /// DataTrackFrame or std::vector copy when the caller already owns a byte - /// buffer, but it is not a zero-copy send. C++17-friendly equivalent of a - /// span overload. + /// the SDK does not retain the buffer. + /// + /// @note This avoids an intermediate copy when the caller already owns a byte + /// buffer, but it is not a zero-copy send. /// /// @param data Pointer to @p size payload bytes. Must be non-null. /// @param size Number of bytes at @p data. Must be non-zero. /// @param user_timestamp Optional application-defined timestamp. The unit is - /// caller-defined; SDK examples use microseconds since the Unix epoch. - /// @return success on delivery acceptance, or a typed error describing why + /// caller-defined. + /// @return Success on delivery acceptance, or a typed error describing why /// the frame could not be queued. Result tryPush(const std::uint8_t* data, std::size_t size, std::optional user_timestamp = std::nullopt); From 440e5efd96b6b3cd315091fedcecf4fa84353dab Mon Sep 17 00:00:00 2001 From: Alan George Date: Tue, 25 Aug 2026 20:09:52 -0600 Subject: [PATCH 5/6] Now accepts empty payload --- include/livekit/local_data_track.h | 5 +++-- src/local_data_track.cpp | 7 +------ src/tests/integration/test_data_track.cpp | 9 +++------ 3 files changed, 7 insertions(+), 14 deletions(-) diff --git a/include/livekit/local_data_track.h b/include/livekit/local_data_track.h index d6029499..8c57aeb1 100644 --- a/include/livekit/local_data_track.h +++ b/include/livekit/local_data_track.h @@ -90,8 +90,9 @@ class LIVEKIT_API LocalDataTrack { /// @note This avoids an intermediate copy when the caller already owns a byte /// buffer, but it is not a zero-copy send. /// - /// @param data Pointer to @p size payload bytes. Must be non-null. - /// @param size Number of bytes at @p data. Must be non-zero. + /// @param data Pointer to @p size payload bytes. May be null when @p size is + /// zero. + /// @param size Number of bytes at @p data. May be zero. /// @param user_timestamp Optional application-defined timestamp. The unit is /// caller-defined. /// @return Success on delivery acceptance, or a typed error describing why diff --git a/src/local_data_track.cpp b/src/local_data_track.cpp index 9f3055f8..5e619d5f 100644 --- a/src/local_data_track.cpp +++ b/src/local_data_track.cpp @@ -44,11 +44,7 @@ Result LocalDataTrack::tryPush(const std::uint return Result::failure(LocalDataTrackTryPushError{ LocalDataTrackTryPushErrorCode::INVALID_HANDLE, "LocalDataTrack::tryPush: invalid FFI handle"}); } - if (size == 0) { - return Result::failure( - LocalDataTrackTryPushError{LocalDataTrackTryPushErrorCode::INTERNAL, "LocalDataTrack::tryPush: empty size"}); - - } else if (data == nullptr) { + if (size != 0 && data == nullptr) { return Result::failure(LocalDataTrackTryPushError{ LocalDataTrackTryPushErrorCode::INTERNAL, "LocalDataTrack::tryPush: payload pointer is null"}); } @@ -58,7 +54,6 @@ Result LocalDataTrack::tryPush(const std::uint auto* msg = req.mutable_local_data_track_try_push(); msg->set_track_handle(static_cast(handle_.get())); auto* pf = msg->mutable_frame(); - // Size and data are checked above pf->set_payload(data, size); if (user_timestamp.has_value()) { pf->set_user_timestamp(user_timestamp.value()); diff --git a/src/tests/integration/test_data_track.cpp b/src/tests/integration/test_data_track.cpp index 0514cdae..abe610d7 100644 --- a/src/tests/integration/test_data_track.cpp +++ b/src/tests/integration/test_data_track.cpp @@ -975,16 +975,13 @@ TEST_F(DataTrackE2ETest, RejectsNullBorrowedPayloadWithNonZeroSize) { local_track->unpublishDataTrack(); } -TEST_F(DataTrackE2ETest, RejectsEmptyBorrowedPayload) { +TEST_F(DataTrackE2ETest, AcceptsEmptyPayload) { const auto track_name = makeTrackName("empty_borrowed_payload"); auto rooms = testRooms(1); auto local_track = requirePublishedTrack(rooms[0]->localParticipant(), track_name); - const std::uint8_t payload = 0; - const auto push_result = local_track->tryPush(&payload, 0); - ASSERT_FALSE(push_result); - EXPECT_EQ(push_result.error().code, LocalDataTrackTryPushErrorCode::INTERNAL); - EXPECT_FALSE(push_result.error().message.empty()); + EXPECT_TRUE(local_track->tryPush(nullptr, 0)); + EXPECT_TRUE(local_track->tryPush(std::vector{})); local_track->unpublishDataTrack(); } From 5d8bd3b9f31d0c683f8f08424b5f97b350f832f5 Mon Sep 17 00:00:00 2001 From: Alan George Date: Fri, 28 Aug 2026 09:21:04 -0600 Subject: [PATCH 6/6] Make empty size a no-op matching prior behavior --- include/livekit/local_data_track.h | 12 +++++++++--- src/local_data_track.cpp | 5 ++++- src/tests/integration/test_data_track.cpp | 2 ++ 3 files changed, 15 insertions(+), 4 deletions(-) diff --git a/include/livekit/local_data_track.h b/include/livekit/local_data_track.h index 8c57aeb1..9c4a2bfe 100644 --- a/include/livekit/local_data_track.h +++ b/include/livekit/local_data_track.h @@ -67,6 +67,8 @@ class LIVEKIT_API LocalDataTrack { /// Try to push a frame to all subscribers of this track. /// + /// An empty payload succeeds without sending a frame to subscribers. + /// /// @param frame The frame to push. /// @return Success on delivery acceptance, or a typed error describing why /// the frame could not be queued. @@ -74,6 +76,8 @@ class LIVEKIT_API LocalDataTrack { /// Try to push a frame to all subscribers of this track. /// + /// An empty payload succeeds without sending a frame to subscribers. + /// /// @param payload The payload to push. /// @param user_timestamp Optional application-defined timestamp. The unit is /// caller-defined. @@ -86,13 +90,15 @@ class LIVEKIT_API LocalDataTrack { /// /// Copies @p size bytes from @p data into an FFI request before returning; /// the SDK does not retain the buffer. + /// An empty payload succeeds without sending a frame to subscribers. /// /// @note This avoids an intermediate copy when the caller already owns a byte /// buffer, but it is not a zero-copy send. /// - /// @param data Pointer to @p size payload bytes. May be null when @p size is - /// zero. - /// @param size Number of bytes at @p data. May be zero. + /// @param data Pointer to @p size payload bytes. Must be non-null when + /// @p size is non-zero. + /// @param size Number of bytes at @p data. A value of zero is a successful + /// no-op. /// @param user_timestamp Optional application-defined timestamp. The unit is /// caller-defined. /// @return Success on delivery acceptance, or a typed error describing why diff --git a/src/local_data_track.cpp b/src/local_data_track.cpp index 5e619d5f..5cd0f084 100644 --- a/src/local_data_track.cpp +++ b/src/local_data_track.cpp @@ -44,7 +44,10 @@ Result LocalDataTrack::tryPush(const std::uint return Result::failure(LocalDataTrackTryPushError{ LocalDataTrackTryPushErrorCode::INVALID_HANDLE, "LocalDataTrack::tryPush: invalid FFI handle"}); } - if (size != 0 && data == nullptr) { + if (size == 0) { + return Result::success(); + } + if (data == nullptr) { return Result::failure(LocalDataTrackTryPushError{ LocalDataTrackTryPushErrorCode::INTERNAL, "LocalDataTrack::tryPush: payload pointer is null"}); } diff --git a/src/tests/integration/test_data_track.cpp b/src/tests/integration/test_data_track.cpp index 0df5c892..d584c551 100644 --- a/src/tests/integration/test_data_track.cpp +++ b/src/tests/integration/test_data_track.cpp @@ -973,6 +973,8 @@ TEST_F(DataTrackE2ETest, AcceptsEmptyPayload) { auto rooms = testRooms(1); auto local_track = requirePublishedTrack(rooms[0]->localParticipant(), track_name); + const std::uint8_t payload = 0; + EXPECT_TRUE(local_track->tryPush(&payload, 0)); EXPECT_TRUE(local_track->tryPush(nullptr, 0)); EXPECT_TRUE(local_track->tryPush(std::vector{}));