Expose data track pipeline options (#186)

* Expose data track pipeline options

---------

Co-authored-by: Alan George <alan.george@livekit.io>

Author: ladvoc

include/livekit/remote_data_track.h

@@ -16,7 +16,9 @@
 
 #pragma once
 
+#include <cstdint>
 #include <memory>
+#include <optional>
 #include <string>
 
 #include "livekit/data_track_error.h"
@@ -32,6 +34,21 @@ namespace proto {
 class OwnedRemoteDataTrack;
 }
 
+/// Track-level options that configure how the incoming-frame pipeline
+/// reassembles packets for a remote data track.
+///
+/// Applied via RemoteDataTrack::setPipelineOptions().
+struct DataTrackPipelineOptions {
+  /// Maximum number of partial frames the depacketizer will track
+  /// concurrently for this track.
+  ///
+  /// Defaults to 1. Higher values give more out-of-order tolerance for
+  /// high-frequency senders at the cost of additional buffering. Zero is
+  /// not a valid value; if a value of zero is provided, it will be
+  /// clamped to one. Leave unset to keep the current value.
+  std::optional<std::uint32_t> max_partial_frames{std::nullopt};
+};
+
 /// Represents a data track published by a remote participant.
 ///
 /// Discovered via the DataTrackPublishedEvent room event. Unlike
@@ -65,6 +82,16 @@ class RemoteDataTrack {
   /// Whether the track is still published by the remote participant.
   LIVEKIT_API bool isPublished() const;
 
+  /// @brief Configures options for the pipeline handling incoming packets for
+  /// this track.
+  ///
+  /// These options apply to all current and future subscriptions of this
+  /// track, and may be set at any time. New options take effect with the
+  /// next received packet.
+  ///
+  /// @param options Pipeline options to apply to this remote data track.
+  LIVEKIT_API void setPipelineOptions(const DataTrackPipelineOptions& options);
+
 #ifdef LIVEKIT_TEST_ACCESS
   /// Test-only accessor for exercising lower-level FFI subscription paths.
   uintptr_t testFfiHandleId() const noexcept { return ffiHandleId(); }

src/remote_data_track.cpp

@@ -46,6 +46,22 @@ bool RemoteDataTrack::isPublished() const {
   return resp.remote_data_track_is_published().is_published();
 }
 
+void RemoteDataTrack::setPipelineOptions(const DataTrackPipelineOptions& options) {
+  if (!handle_.valid()) {
+    return;
+  }
+
+  proto::FfiRequest req;
+  auto* msg = req.mutable_remote_data_track_set_pipeline_options();
+  msg->set_track_handle(static_cast<uint64_t>(handle_.get()));
+  auto* opts = msg->mutable_options();
+  if (options.max_partial_frames) {
+    opts->set_max_partial_frames(*options.max_partial_frames);
+  }
+
+  FfiClient::instance().sendRequest(req);
+}
+
 Result<std::shared_ptr<DataTrackStream>, SubscribeDataTrackError> RemoteDataTrack::subscribe(
     const DataTrackStream::Options& options) {
   if (!handle_.valid()) {

src/tests/integration/test_data_track.cpp

@@ -367,6 +367,78 @@ TEST_P(DataTrackTransportTest, PublishesAndReceivesFramesEndToEnd) {
   EXPECT_TRUE(remote_track_published_after_read);
 }
 
+TEST_F(DataTrackE2ETest, ReceivesFramesWithCustomPipelineOptionsEndToEnd) {
+  constexpr std::size_t payload_len = 1024;
+  const auto track_name = makeTrackName("pipeline_options");
+
+  std::vector<TestRoomConnectionOptions> room_configs(2);
+  room_configs[0].room_options.single_peer_connection = false;
+  room_configs[1].room_options.single_peer_connection = false;
+
+  DataTrackPublishedDelegate subscriber_delegate;
+  room_configs[1].delegate = &subscriber_delegate;
+
+  auto rooms = testRooms(room_configs);
+  auto& publisher_room = rooms[0];
+  const auto publisher_identity = lockLocalParticipant(*publisher_room)->identity();
+
+  auto local_track = requirePublishedTrack(publisher_room->localParticipant(), track_name);
+  ASSERT_TRUE(local_track->isPublished());
+  EXPECT_FALSE(local_track->info().uses_e2ee);
+  EXPECT_EQ(local_track->info().name, track_name);
+
+  auto remote_track = subscriber_delegate.waitForTrack(kTrackWaitTimeout);
+  ASSERT_NE(remote_track, nullptr) << "Timed out waiting for remote data track";
+  EXPECT_TRUE(remote_track->isPublished());
+  EXPECT_FALSE(remote_track->info().uses_e2ee);
+  EXPECT_EQ(remote_track->info().name, track_name);
+  EXPECT_EQ(remote_track->publisherIdentity(), publisher_identity);
+
+  DataTrackPipelineOptions pipeline_options;
+  pipeline_options.max_partial_frames = 4;
+  remote_track->setPipelineOptions(pipeline_options);
+
+  auto subscribe_result = remote_track->subscribe();
+  if (!subscribe_result) {
+    FAIL() << describeDataTrackError(subscribe_result.error());
+  }
+  auto subscription = subscribe_result.value();
+
+  std::atomic<bool> keep_publishing{true};
+  auto publisher = std::async(std::launch::async, [&]() {
+    DataTrackFrame frame;
+    frame.payload.assign(payload_len, kTransportPayloadValue);
+    while (keep_publishing.load()) {
+      requirePushSuccess(local_track->tryPush(frame), "Failed to push data frame");
+      std::this_thread::sleep_for(50ms);
+    }
+  });
+
+  DataTrackFrame frame;
+  std::exception_ptr read_error;
+  try {
+    frame = readFrameWithTimeout(subscription, kTransportFrameTimeout);
+  } catch (...) {
+    read_error = std::current_exception();
+  }
+
+  const bool remote_track_published_after_read = remote_track->isPublished();
+  keep_publishing.store(false);
+  subscription->close();
+  local_track->unpublishDataTrack();
+
+  publisher.get();
+  if (read_error) {
+    std::rethrow_exception(read_error);
+  }
+
+  ASSERT_EQ(frame.payload.size(), payload_len);
+  EXPECT_TRUE(std::all_of(frame.payload.begin(), frame.payload.end(),
+                          [](std::uint8_t byte) { return byte == kTransportPayloadValue; }));
+  EXPECT_FALSE(frame.user_timestamp.has_value());
+  EXPECT_TRUE(remote_track_published_after_read);
+}
+
 TEST_F(DataTrackE2ETest, UnpublishUpdatesPublishedStateEndToEnd) {
   const auto track_name = makeTrackName("published_state");