relay: name the initial track state a subscriber must observe

src/MoqxRelay.cpp

@@ -9,6 +9,7 @@
 #include "MoqxRelay.h"
 #include "relay/CrossExecFilter.h"
 #include "relay/CrossExecForwarderCallback.h"
+#include "relay/InitialTrackState.h"
 #include "relay/LocalForwarderCallback.h"
 #include "relay/NullConsumers.h"
 #include "relay/PublisherCrossExecFilter.h"
@@ -1135,20 +1136,18 @@ folly::coro::Task<void> MoqxRelay::addSubscriberAndPublishViaLocalForwarder(
 
   // Capture largest/extensions on publisherExec; reading them on subscriberExec would
   // race the publisher advancing largest_.
-  std::optional<AbsoluteLocation> seedLargest;
-  Extensions seedExtensions;
+  InitialTrackState initial;
   co_await folly::coro::co_withExecutor(
       folly::getKeepAliveToken(publisherExec),
       [&]() -> folly::coro::Task<void> {
-        seedLargest = publisherFwd->largest();
-        seedExtensions = publisherFwd->extensions();
+        initial = InitialTrackState::capture(*publisherFwd);
         co_return;
       }()
   );
 
   auto [localFwd, isNew, localReg] = acquireLocalForwarder(ftn, [&] {
-    auto fwd = std::make_shared<MoQForwarder>(ftn, seedLargest);
-    fwd->setExtensions(seedExtensions);
+    auto fwd = std::make_shared<MoQForwarder>(ftn);
+    initial.applyTo(*fwd);
     return fwd;
   });
 
@@ -1729,10 +1728,7 @@ MoqxRelay::subscribeUpstreamAndApplyOk(
   }
   // Apply the OK to the forwarder; the NGR rides the outgoing SUBSCRIBE (record, don't fire).
   const auto& ok = subRes.value()->subscribeOk();
-  if (ok.largest) {
-    publisherFwd->updateLargest(ok.largest->group, ok.largest->object);
-  }
-  publisherFwd->setExtensions(ok.extensions);
+  InitialTrackState{ok.largest, ok.extensions}.applyTo(*publisherFwd);
   publisherFwd->tryProcessNewGroupRequest(params, /*fire=*/false);
   // Moving the handle shared_ptr keeps the pointee (and `ok`) alive, so reading ok.*
   // in the same initializer is well-defined.
@@ -2043,15 +2039,10 @@ folly::coro::Task<Publisher::SubscribeResult> MoqxRelay::subscribeFromSubscriber
   }
 
   if (attach.upstreamOk) {
-    localFwd->setExtensions(attach.upstreamOk->extensions);
-    if (attach.upstreamOk->largest) {
-      localFwd->updateLargest(
-          attach.upstreamOk->largest->group,
-          attach.upstreamOk->largest->object
-      );
-      // Seed the subscriber snapshot so a post-SUBSCRIBE_OK joining fetch resolves.
-      sub->updateLargest(*attach.upstreamOk->largest);
-    }
+    InitialTrackState initial{attach.upstreamOk->largest, attach.upstreamOk->extensions};
+    initial.applyTo(*localFwd);
+    // Also on the subscriber, so a post-SUBSCRIBE_OK joining fetch resolves.
+    initial.applyTo(*sub);
   }
   replayPendingFowarderEvents(localFwd.get(), attach.finalCallback, *pendingCb, forward);
   localFwd->tryProcessNewGroupRequest(subReq.params);
@@ -2132,9 +2123,7 @@ MoqxRelay::subscribeImpl(SubscribeRequest subReq, std::shared_ptr<TrackConsumer>
       co_return folly::makeUnexpected(std::move(okOrErr.error()));
     }
     auto& ok = okOrErr.value();
-    if (ok.largest) {
-      subscriber->updateLargest(*ok.largest);
-    }
+    InitialTrackState{ok.largest, ok.extensions}.applyTo(*subscriber);
     if (auto err = completeUpstreamSubscription(
             ftn,
             ok,

src/relay/InitialTrackState.h

@@ -0,0 +1,43 @@
+/*
+ * Copyright (c) OpenMOQ contributors.
+ * This source code is licensed under the Apache 2.0 license found in the
+ * LICENSE file in the root directory of this source tree.
+ */
+
+#pragma once
+
+#include <moxygen/relay/MoQForwarder.h>
+
+#include <optional>
+
+namespace openmoq::moqx {
+
+// The position and properties a subscriber must observe before it can be told about a
+// track — what SUBSCRIBE_OK and PUBLISH carry. A forwarder that has not had one applied
+// is not attachable: a subscriber reading its largest first gets a value the client
+// interprets as a track restart.
+struct InitialTrackState {
+  std::optional<moxygen::AbsoluteLocation> largest;
+  moxygen::Extensions extensions;
+
+  // A point-in-time copy; the publisher may advance past it before it is applied.
+  static InitialTrackState capture(moxygen::MoQForwarder& forwarder) {
+    return {forwarder.largest(), forwarder.extensions()};
+  }
+
+  void applyTo(moxygen::MoQForwarder& forwarder) const {
+    forwarder.setExtensions(extensions);
+    if (largest) {
+      forwarder.updateLargest(largest->group, largest->object);
+    }
+  }
+
+  // Extensions reach a subscriber through its forwarder; only the position is per-subscriber.
+  void applyTo(moxygen::MoQForwarder::Subscriber& subscriber) const {
+    if (largest) {
+      subscriber.updateLargest(*largest);
+    }
+  }
+};
+
+} // namespace openmoq::moqx

test/CMakeLists.txt

@@ -382,6 +382,17 @@ target_link_libraries(moqx_subscription_registry_test PRIVATE
 )
 gtest_discover_tests(moqx_subscription_registry_test)
 
+# InitialTrackState unit tests (header-only value type)
+add_executable(moqx_initial_track_state_test
+  InitialTrackStateTest.cpp
+)
+target_link_libraries(moqx_initial_track_state_test PRIVATE
+  moqx_core
+  moqx_test_main
+  GTest::gmock
+)
+gtest_discover_tests(moqx_initial_track_state_test)
+
 # TopNFilter unit tests
 add_executable(moqx_topn_filter_test
   TopNFilterTest.cpp

test/InitialTrackStateTest.cpp

@@ -0,0 +1,111 @@
+/*
+ * Copyright (c) OpenMOQ contributors.
+ */
+
+#include "relay/InitialTrackState.h"
+#include "relay/NullConsumers.h"
+
+#include <folly/portability/GMock.h>
+#include <folly/portability/GTest.h>
+
+using namespace testing;
+using namespace moxygen;
+using openmoq::moqx::InitialTrackState;
+
+namespace {
+
+const TrackNamespace kTestNs{{"test", "namespace"}};
+const FullTrackName kFtn{kTestNs, "track1"};
+
+constexpr uint64_t kExtType = 0xBEEF'0000;
+
+Extensions extensionsWith(uint64_t value) {
+  Extensions e;
+  e.insertMutableExtension(Extension{kExtType, value});
+  return e;
+}
+
+// addSubscriber only uses the session as a map key, so a null one is enough here.
+std::shared_ptr<MoQForwarder::Subscriber> addSubscriber(MoQForwarder& forwarder) {
+  SubscribeRequest req;
+  req.fullTrackName = kFtn;
+  req.requestID = RequestID(1);
+  req.forward = true;
+  return forwarder.addSubscriber(nullptr, req, std::make_shared<NullTrackConsumer>());
+}
+
+TEST(InitialTrackStateTest, CaptureRoundTripsPositionAndProperties) {
+  MoQForwarder source(kFtn, AbsoluteLocation{7, 3});
+  source.setExtensions(extensionsWith(42));
+
+  auto state = InitialTrackState::capture(source);
+
+  EXPECT_EQ(state.largest, (AbsoluteLocation{7, 3}));
+  EXPECT_EQ(state.extensions.getIntExtension(kExtType), 42);
+}
+
+TEST(InitialTrackStateTest, CaptureOfFreshForwarderHasNoPosition) {
+  MoQForwarder source(kFtn);
+  auto state = InitialTrackState::capture(source);
+  EXPECT_FALSE(state.largest.has_value());
+}
+
+TEST(InitialTrackStateTest, ApplyToForwarderSetsBothHalves) {
+  MoQForwarder target(kFtn);
+  InitialTrackState{AbsoluteLocation{4, 9}, extensionsWith(7)}.applyTo(target);
+
+  EXPECT_EQ(target.largest(), (AbsoluteLocation{4, 9}));
+  EXPECT_EQ(target.extensions().getIntExtension(kExtType), 7);
+}
+
+TEST(InitialTrackStateTest, ApplyWithoutPositionStillSetsProperties) {
+  MoQForwarder target(kFtn);
+  InitialTrackState{std::nullopt, extensionsWith(5)}.applyTo(target);
+
+  EXPECT_FALSE(target.largest().has_value());
+  EXPECT_EQ(target.extensions().getIntExtension(kExtType), 5);
+}
+
+// updateLargest is monotonic, so a stale capture cannot rewind a forwarder that has
+// already advanced past it.
+TEST(InitialTrackStateTest, ApplyDoesNotRegressAnAdvancedForwarder) {
+  MoQForwarder target(kFtn, AbsoluteLocation{10, 0});
+  InitialTrackState{AbsoluteLocation{2, 0}, Extensions{}}.applyTo(target);
+  EXPECT_EQ(target.largest(), (AbsoluteLocation{10, 0}));
+}
+
+TEST(InitialTrackStateTest, CaptureThenApplyIsIdentity) {
+  MoQForwarder source(kFtn, AbsoluteLocation{3, 1});
+  source.setExtensions(extensionsWith(99));
+
+  MoQForwarder target(kFtn);
+  InitialTrackState::capture(source).applyTo(target);
+
+  EXPECT_EQ(target.largest(), source.largest());
+  EXPECT_EQ(target.extensions().getIntExtension(kExtType), 99);
+}
+
+TEST(InitialTrackStateTest, ApplyToSubscriberSetsPositionOnly) {
+  MoQForwarder forwarder(kFtn);
+  auto sub = addSubscriber(forwarder);
+  ASSERT_NE(sub, nullptr);
+
+  InitialTrackState{AbsoluteLocation{6, 2}, extensionsWith(11)}.applyTo(*sub);
+
+  EXPECT_EQ(sub->subscribeOk().largest, (AbsoluteLocation{6, 2}));
+  // The forwarder is the source of a subscriber's extensions; applyTo must not touch them.
+  EXPECT_FALSE(sub->subscribeOk().extensions.getIntExtension(kExtType).has_value());
+}
+
+TEST(InitialTrackStateTest, ApplyToSubscriberWithoutPositionIsANoOp) {
+  MoQForwarder forwarder(kFtn, AbsoluteLocation{1, 1});
+  auto sub = addSubscriber(forwarder);
+  ASSERT_NE(sub, nullptr);
+  auto before = sub->subscribeOk().largest;
+
+  InitialTrackState{std::nullopt, Extensions{}}.applyTo(*sub);
+
+  EXPECT_EQ(sub->subscribeOk().largest, before);
+}
+
+} // namespace