From c95e54c21a57ce24daddaf48f513faece6f96c78 Mon Sep 17 00:00:00 2001 From: afrind <8259689+afrind@users.noreply.github.com> Date: Tue, 11 Aug 2026 11:04:30 -0400 Subject: [PATCH] relay: name the initial track state a subscriber must observe A track's starting largest location and extensions were passed around as two loose values and applied by hand at five sites, in two different orders. Group them as InitialTrackState. capture() reads the pair off a live forwarder; applyTo() writes it to a forwarder or to a subscriber, whichever the caller has. The two fields are independent, so the order applyTo picks is arbitrary. No behavior change. Co-Authored-By: Claude Opus 5 (1M context) --- src/MoqxRelay.cpp | 33 ++++------ src/relay/InitialTrackState.h | 43 +++++++++++++ test/CMakeLists.txt | 11 ++++ test/InitialTrackStateTest.cpp | 111 +++++++++++++++++++++++++++++++++ 4 files changed, 176 insertions(+), 22 deletions(-) create mode 100644 src/relay/InitialTrackState.h create mode 100644 test/InitialTrackStateTest.cpp diff --git a/src/MoqxRelay.cpp b/src/MoqxRelay.cpp index 4cb1e770..5d9a9e35 100644 --- a/src/MoqxRelay.cpp +++ b/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 MoqxRelay::addSubscriberAndPublishViaLocalForwarder( // Capture largest/extensions on publisherExec; reading them on subscriberExec would // race the publisher advancing largest_. - std::optional seedLargest; - Extensions seedExtensions; + InitialTrackState initial; co_await folly::coro::co_withExecutor( folly::getKeepAliveToken(publisherExec), [&]() -> folly::coro::Task { - seedLargest = publisherFwd->largest(); - seedExtensions = publisherFwd->extensions(); + initial = InitialTrackState::capture(*publisherFwd); co_return; }() ); auto [localFwd, isNew, localReg] = acquireLocalForwarder(ftn, [&] { - auto fwd = std::make_shared(ftn, seedLargest); - fwd->setExtensions(seedExtensions); + auto fwd = std::make_shared(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 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 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, diff --git a/src/relay/InitialTrackState.h b/src/relay/InitialTrackState.h new file mode 100644 index 00000000..653806f6 --- /dev/null +++ b/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 + +#include + +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 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 diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt index da411c99..9eea3c54 100644 --- a/test/CMakeLists.txt +++ b/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 diff --git a/test/InitialTrackStateTest.cpp b/test/InitialTrackStateTest.cpp new file mode 100644 index 00000000..41f82c00 --- /dev/null +++ b/test/InitialTrackStateTest.cpp @@ -0,0 +1,111 @@ +/* + * Copyright (c) OpenMOQ contributors. + */ + +#include "relay/InitialTrackState.h" +#include "relay/NullConsumers.h" + +#include +#include + +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 addSubscriber(MoQForwarder& forwarder) { + SubscribeRequest req; + req.fullTrackName = kFtn; + req.requestID = RequestID(1); + req.forward = true; + return forwarder.addSubscriber(nullptr, req, std::make_shared()); +} + +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