diff --git a/moxygen/MoQFramer.cpp b/moxygen/MoQFramer.cpp index 53d3e2bc0..3e0e8728c 100644 --- a/moxygen/MoQFramer.cpp +++ b/moxygen/MoQFramer.cpp @@ -49,6 +49,7 @@ bool isRequestSpecificParam(moxygen::TrackRequestParamKey key) { case moxygen::TrackRequestParamKey::GROUP_ORDER: case moxygen::TrackRequestParamKey::SUBSCRIBER_PRIORITY: case moxygen::TrackRequestParamKey::FORWARD: + case moxygen::TrackRequestParamKey::NEW_GROUP_REQUEST: return true; default: return false; @@ -4452,6 +4453,16 @@ WriteResult MoQFrameWriter::writeSubscribeRequestHelper( forwardParam.asUint64 = 0; requestSpecificParams.push_back(forwardParam); } + + auto newGroupRequestValue = getFirstIntParam( + subscribeRequest.params, TrackRequestParamKey::NEW_GROUP_REQUEST); + if (newGroupRequestValue.has_value()) { + Parameter newGroupRequestParam; + newGroupRequestParam.key = + folly::to_underlying(TrackRequestParamKey::NEW_GROUP_REQUEST); + newGroupRequestParam.asUint64 = *newGroupRequestValue; + requestSpecificParams.push_back(newGroupRequestParam); + } } else { writeVarint( writeBuf, @@ -4538,6 +4549,16 @@ WriteResult MoQFrameWriter::writeRequestUpdate( forwardParam.asUint64 = *update.forward ? 1 : 0; requestSpecificParams.push_back(forwardParam); } + + auto newGroupRequestValue = getFirstIntParam( + update.params, TrackRequestParamKey::NEW_GROUP_REQUEST); + if (newGroupRequestValue.has_value()) { + Parameter newGroupRequestParam; + newGroupRequestParam.key = + folly::to_underlying(TrackRequestParamKey::NEW_GROUP_REQUEST); + newGroupRequestParam.asUint64 = *newGroupRequestValue; + requestSpecificParams.push_back(newGroupRequestParam); + } } else { // For draft < 15, start and endGroup are mandatory XCHECK(update.start.has_value()) << "start is required for draft < 15"; @@ -4873,6 +4894,16 @@ WriteResult MoQFrameWriter::writePublishOk( forwardParam.asUint64 = 0; requestSpecificParams.push_back(forwardParam); } + + auto newGroupRequestValue = getFirstIntParam( + publishOk.params, TrackRequestParamKey::NEW_GROUP_REQUEST); + if (newGroupRequestValue.has_value()) { + Parameter newGroupRequestParam; + newGroupRequestParam.key = + folly::to_underlying(TrackRequestParamKey::NEW_GROUP_REQUEST); + newGroupRequestParam.asUint64 = *newGroupRequestValue; + requestSpecificParams.push_back(newGroupRequestParam); + } } else { writeVarint( writeBuf, diff --git a/moxygen/MoQTypes.cpp b/moxygen/MoQTypes.cpp index 518ab8421..b891ae818 100644 --- a/moxygen/MoQTypes.cpp +++ b/moxygen/MoQTypes.cpp @@ -235,6 +235,11 @@ const folly::F14FastSet kAllowedFramesForForward = { FrameType::PUBLISH_OK, FrameType::SUBSCRIBE_NAMESPACE}; +const folly::F14FastSet kAllowedFramesForNewGroupRequest = { + FrameType::SUBSCRIBE, + FrameType::REQUEST_UPDATE, + FrameType::PUBLISH_OK}; + // Allowlist mapping: TrackRequestParamKey -> set of allowed FrameTypes // Empty set means allowed for all frame types const folly::F14FastMap> @@ -252,6 +257,8 @@ const folly::F14FastMap> {TrackRequestParamKey::GROUP_ORDER, kAllowedFramesForGroupOrder}, {TrackRequestParamKey::LARGEST_OBJECT, kAllowedFramesForLargestObject}, {TrackRequestParamKey::FORWARD, kAllowedFramesForForward}, + {TrackRequestParamKey::NEW_GROUP_REQUEST, + kAllowedFramesForNewGroupRequest}, }; // Frame types that allow all parameters (no validation) diff --git a/moxygen/MoQTypes.h b/moxygen/MoQTypes.h index 60bd0115b..8f1673349 100644 --- a/moxygen/MoQTypes.h +++ b/moxygen/MoQTypes.h @@ -443,6 +443,7 @@ enum class TrackRequestParamKey : uint64_t { GROUP_ORDER = 0x22, LARGEST_OBJECT = 0x9, FORWARD = 0x10, + NEW_GROUP_REQUEST = 0x32, }; class Parameters { diff --git a/moxygen/relay/MoQForwarder.cpp b/moxygen/relay/MoQForwarder.cpp index 1c04723e6..550fc6727 100644 --- a/moxygen/relay/MoQForwarder.cpp +++ b/moxygen/relay/MoQForwarder.cpp @@ -153,6 +153,10 @@ void MoQForwarder::setExtensions(Extensions extensions) { } } +void MoQForwarder::setOutstandingNewGroupRequest(uint64_t value) { + outstandingNewGroupRequest_ = value; +} + void MoQForwarder::setLargest(AbsoluteLocation largest) { largest_ = largest; } @@ -361,6 +365,11 @@ void MoQForwarder::removeSubscriberIt( void MoQForwarder::updateLargest(uint64_t group, uint64_t object) { AbsoluteLocation now{group, object}; if (!largest_ || now > *largest_) { + // Clear any outstanding NEW_GROUP_REQUEST once the upstream group advances. + if (outstandingNewGroupRequest_.has_value() && + (!largest_ || group > largest_->group)) { + outstandingNewGroupRequest_.reset(); + } largest_ = now; } } @@ -529,6 +538,44 @@ void MoQForwarder::removeForwardingSubscriber() { } } +bool MoQForwarder::shouldForwardNewGroupRequest(uint64_t requestedGroup) const { + // the track must support dynamic groups. + if (!getPublisherDynamicGroups(extensions_).value_or(false)) { + return false; + } + // non-zero value <= LargestGroup means the new group is already + // available — do not send upstream. + if (requestedGroup != 0 && largest_.has_value() && + requestedGroup <= largest_->group) { + return false; + } + // already have an outstanding request with >= value. + if (outstandingNewGroupRequest_.has_value() && + *outstandingNewGroupRequest_ >= requestedGroup) { + return false; + } + return true; +} + +void MoQForwarder::fireNewGroupRequest() { + if (callback_ && outstandingNewGroupRequest_.has_value()) { + callback_->newGroupRequested(this, *outstandingNewGroupRequest_); + } +} + +void MoQForwarder::tryProcessNewGroupRequest( + const Parameters& params, + bool fire) { + auto value = + getFirstIntParam(params, TrackRequestParamKey::NEW_GROUP_REQUEST); + if (value.has_value() && shouldForwardNewGroupRequest(*value)) { + setOutstandingNewGroupRequest(*value); + if (fire) { + fireNewGroupRequest(); + } + } +} + Payload MoQForwarder::maybeClone(const Payload& payload) { return payload ? payload->clone() : nullptr; } @@ -597,6 +644,22 @@ void MoQForwarder::Subscriber::setParam(const TrackRequestParameter& param) { } } +void MoQForwarder::Subscriber::onPublishOk(const PublishOk& pubOk) { + // Update subscriber range from PUBLISH_OK + std::optional end; + if (pubOk.endGroup) { + end = AbsoluteLocation{*pubOk.endGroup, 0}; + } + range = + toSubscribeRange(pubOk.start, end, pubOk.locType, forwarder.largest()); + + // Update forward flag + shouldForward = pubOk.forward; + + // Handle NEW_GROUP_REQUEST forwarding if present + forwarder.tryProcessNewGroupRequest(pubOk.params); +} + folly::coro::Task> MoQForwarder::Subscriber::requestUpdate(RequestUpdate requestUpdate) { // Validation: @@ -604,6 +667,7 @@ MoQForwarder::Subscriber::requestUpdate(RequestUpdate requestUpdate) { // - End location can increase or decrease // - For bounded subscriptions (endGroup > 0), end must be >= start // - Forward state is optional and only updated if explicitly provided + // - A New Group can be requested // Only update start if provided if (requestUpdate.start.has_value()) { @@ -639,6 +703,9 @@ MoQForwarder::Subscriber::requestUpdate(RequestUpdate requestUpdate) { forwarder.removeForwardingSubscriber(); } + // Only update new group request if provided + forwarder.tryProcessNewGroupRequest(requestUpdate.params); + co_return RequestOk{.requestID = requestUpdate.requestID}; } diff --git a/moxygen/relay/MoQForwarder.h b/moxygen/relay/MoQForwarder.h index 0a91369c9..36014d697 100644 --- a/moxygen/relay/MoQForwarder.h +++ b/moxygen/relay/MoQForwarder.h @@ -49,6 +49,13 @@ class MoQForwarder : public TrackConsumer { return extensions_; } + // Extract the NEW_GROUP_REQUEST param from `params`, check if it should + // be forwarded upstream, record it as outstanding, and optionally fire + // the newGroupRequested callback. Pass fire=false when the NGR already + // rides an outgoing SUBSCRIBE and no extra REQUEST_UPDATE is needed. + void tryProcessNewGroupRequest(const Parameters& params, bool fire = true); + + void setLargest(AbsoluteLocation largest); std::optional largest() { @@ -60,6 +67,8 @@ class MoQForwarder : public TrackConsumer { virtual ~Callback() = default; virtual void onEmpty(MoQForwarder*) = 0; virtual void forwardChanged(MoQForwarder*) {} + // This fires whenever an unseen NGR is received + virtual void newGroupRequested(MoQForwarder*, uint64_t group) {} }; void setCallback(std::shared_ptr callback); @@ -107,6 +116,10 @@ class MoQForwarder : public TrackConsumer { // requestID and trackAlias are placeholders (overwritten by MoQSession). PublishRequest getPublishRequest() const; + // Process PUBLISH_OK response, updating range, forward flag, and handling + // NEW_GROUP_REQUEST forwarding via callback + void onPublishOk(const PublishOk& pubOk); + folly::coro::Task> requestUpdate( RequestUpdate requestUpdate) override; @@ -306,6 +319,10 @@ class MoQForwarder : public TrackConsumer { const MoQPublishError& err, const std::string& callsite); + void setOutstandingNewGroupRequest(uint64_t value); + bool shouldForwardNewGroupRequest(uint64_t requestedGroup) const; + void fireNewGroupRequest(); + FullTrackName fullTrackName_; std::optional trackAlias_; folly::F14FastMap> subscribers_; @@ -317,6 +334,9 @@ class MoQForwarder : public TrackConsumer { GroupOrder groupOrder_{GroupOrder::OldestFirst}; std::optional largest_; Extensions extensions_; + // The NEW_GROUP_REQUEST value most recently forwarded upstream; cleared when + // the upstream Largest Group advances (indicating the request was fulfilled). + std::optional outstandingNewGroupRequest_{}; std::shared_ptr callback_; uint64_t forwardingSubscribers_{0}; bool draining_{false}; diff --git a/moxygen/relay/MoQRelay.cpp b/moxygen/relay/MoQRelay.cpp index dd8214756..960aea8c7 100644 --- a/moxygen/relay/MoQRelay.cpp +++ b/moxygen/relay/MoQRelay.cpp @@ -6,6 +6,7 @@ #include "moxygen/relay/MoQRelay.h" #include "moxygen/MoQFilters.h" +#include "moxygen/MoQTrackProperties.h" namespace { constexpr uint8_t kDefaultUpstreamPriority = 128; @@ -34,6 +35,24 @@ folly::coro::Task MoQRelay::doSubscribeUpdate( } } +// Sends a REQUEST_UPDATE that carries only the NEW_GROUP_REQUEST parameter. +folly::coro::Task MoQRelay::doNewGroupRequestUpdate( + std::shared_ptr handle, + uint64_t newGroupRequestValue) { + XLOG(DBG4) << "Sending NEW_GROUP_REQUEST update: " << newGroupRequestValue; + RequestUpdate update; + update.requestID = RequestID(0); + update.existingRequestID = handle->subscribeOk().requestID; + update.params.insertParam(Parameter( + folly::to_underlying(TrackRequestParamKey::NEW_GROUP_REQUEST), + newGroupRequestValue)); + auto updateRes = co_await handle->requestUpdate(std::move(update)); + if (updateRes.hasError()) { + XLOG(ERR) << "NEW_GROUP_REQUEST update failed: " + << updateRes.error().reasonPhrase; + } +} + std::shared_ptr MoQRelay::findNamespaceNode( const TrackNamespace& ns, bool createMissingNodes, @@ -509,13 +528,10 @@ folly::coro::Task MoQRelay::publishToSession( guard.dismiss(); XLOG(DBG1) << "Publish OK sess=" << session.get(); auto& pubOk = pubResult.value().value(); - std::optional end; - if (pubOk.endGroup) { - end = AbsoluteLocation{*pubOk.endGroup, 0}; - } - subscriber->range = - toSubscribeRange(pubOk.start, end, pubOk.locType, forwarder->largest()); - subscriber->shouldForward = pubOk.forward; + + // Process the PUBLISH_OK response - updates range, forward flag, and + // handles NEW_GROUP_REQUEST forwarding via callback + subscriber->onPublishOk(pubOk); } class MoQRelay::NamespaceSubscription @@ -871,6 +887,8 @@ folly::coro::Task MoQRelay::subscribe( auto& rsub = it->second; rsub.requestID = subRes.value()->subscribeOk().requestID; rsub.handle = std::move(subRes.value()); + // Record NGR as outstanding (no fire — it rides the outgoing SUBSCRIBE). + forwarder->tryProcessNewGroupRequest(subReq.params, /*fire=*/false); rsub.promise.setValue(folly::unit); co_return subscriber; } else { @@ -911,6 +929,8 @@ folly::coro::Task MoQRelay::subscribe( doSubscribeUpdate(subscriptionIt->second.handle, /*forward=*/true)) .start(); } + + forwarder->tryProcessNewGroupRequest(subReq.params); co_return subscriber; } } @@ -1019,6 +1039,7 @@ folly::coro::Task MoQRelay::trackStatus( } } + TrackStatusOk trackStatusOk; trackStatusOk.requestID = trackStatus.requestID; trackStatusOk.groupOrder = forwarder->groupOrder(); @@ -1110,4 +1131,23 @@ void MoQRelay::forwardChanged(MoQForwarder* forwarder) { .start(); } +void MoQRelay::newGroupRequested(MoQForwarder* forwarder, uint64_t group) { + auto subscriptionIt = subscriptions_.find(forwarder->fullTrackName()); + if (subscriptionIt == subscriptions_.end()) { + return; + } + auto& subscription = subscriptionIt->second; + // Check if handle is still valid (publisher may have terminated) + if (!subscription.handle) { + XLOG(DBG4) << "Ignoring NEW_GROUP_REQUEST for " << subscriptionIt->first + << " - publisher terminated"; + return; + } + XLOG(INFO) << "New group request detected for " << subscriptionIt->first; + + auto exec = subscription.upstream->getExecutor(); + co_withExecutor(exec, doNewGroupRequestUpdate(subscription.handle, group)) + .start(); +} + } // namespace moxygen diff --git a/moxygen/relay/MoQRelay.h b/moxygen/relay/MoQRelay.h index 3d1d15e31..1af4529a8 100644 --- a/moxygen/relay/MoQRelay.h +++ b/moxygen/relay/MoQRelay.h @@ -188,6 +188,7 @@ class MoQRelay : public Publisher, void onEmpty(MoQForwarder* forwarder) override; void forwardChanged(MoQForwarder* forwarder) override; + void newGroupRequested(MoQForwarder* forwarder, uint64_t group) override; folly::coro::Task publishNamespaceToSession( std::shared_ptr session, @@ -203,6 +204,10 @@ class MoQRelay : public Publisher, std::shared_ptr handle, bool forward); + folly::coro::Task doNewGroupRequestUpdate( + std::shared_ptr handle, + uint64_t newGroupRequestValue); + void publishNamespaceDone( const TrackNamespace& trackNamespace, NamespaceNode* node); diff --git a/moxygen/relay/test/MoQRelayTest.cpp b/moxygen/relay/test/MoQRelayTest.cpp index 9e9494edc..22d8403b8 100644 --- a/moxygen/relay/test/MoQRelayTest.cpp +++ b/moxygen/relay/test/MoQRelayTest.cpp @@ -8,6 +8,7 @@ #include #include #include +#include #include #include #include @@ -1948,6 +1949,10 @@ TEST_F(MoQRelayTest, ResetDuringDrainingMultipleSubscribersDoesNotCrash) { EXPECT_EQ(forwarder, nullptr); } +// ============================================================ +// Extensions Tests +// ============================================================ + // Test: Extensions from publish are forwarded to subscribers via // subscribeNamespace TEST_F(MoQRelayTest, PublishExtensionsForwardedToSubscribers) { @@ -2005,6 +2010,42 @@ TEST_F(MoQRelayTest, PublishExtensionsForwardedToSubscribers) { removeSession(subscriber); } +// ============================================================ +// Dynamic Groups Extension Tests +// ============================================================ + +// Test: relay PUBLISH path – dynamic groups from PublishRequest extensions +// is stored in the forwarder and forwarded to every downstream subscriber +TEST_F(MoQRelayTest, RelayPublishPropagatesDynamicGroupsToSubscribers) { + auto publisherSession = createMockSession(); + auto subscriberSession = createMockSession(); + + // Build a PublishRequest with DYNAMIC_GROUPS enabled + PublishRequest pub; + pub.fullTrackName = kTestTrackName; + setPublisherDynamicGroups(pub, true); + + withSessionContext(publisherSession, [&]() { + auto res = relay_->publish(std::move(pub), createMockSubscriptionHandle()); + ASSERT_TRUE(res.hasValue()); + getOrCreateMockState(publisherSession)->publishConsumers.push_back( + res->consumer); + }); + + auto consumer = createMockConsumer(); + auto handle = + subscribeToTrack(subscriberSession, kTestTrackName, consumer, RequestID(1)); + ASSERT_NE(handle, nullptr); + + auto dynGroups = getPublisherDynamicGroups(handle->subscribeOk()); + ASSERT_TRUE(dynGroups.has_value()); + EXPECT_TRUE(*dynGroups); + + removeSession(subscriberSession); + exec_->drive(); + removeSession(publisherSession); +} + // Test: Extensions from publish are forwarded to late-joining subscribers TEST_F(MoQRelayTest, PublishExtensionsForwardedToLateJoiners) { auto publisherSession = createMockSession(); @@ -2086,6 +2127,57 @@ TEST_F(MoQRelayTest, PublishExtensionsForwardedToLateJoiners) { removeSession(subscriber2); } +// Test: relay SUBSCRIBE path – dynamic groups from the upstream SubscribeOk is +// stored in the forwarder and forwarded to both the first and late-joining +// downstream subscribers +TEST_F(MoQRelayTest, RelaySubscribePropagatesDynamicGroupsToAllSubscribers) { + auto publisherSession = createMockSession(); + auto subscriber1 = createMockSession(); + auto subscriber2 = createMockSession(); + + doPublishNamespace(publisherSession, kTestNamespace); + + // Upstream returns a SubscribeOk with DYNAMIC_GROUPS = true + SubscribeOk upstreamOk; + upstreamOk.requestID = RequestID(1); + upstreamOk.trackAlias = TrackAlias(1); + upstreamOk.expires = std::chrono::milliseconds(0); + upstreamOk.groupOrder = GroupOrder::OldestFirst; + setPublisherDynamicGroups(upstreamOk, true); + + EXPECT_CALL(*publisherSession, subscribe(_, _)) + .WillOnce([upstreamOk](auto /*req*/, auto /*consumer*/) { + auto handle = + std::make_shared>(upstreamOk); + return folly::coro::makeTask( + folly::Expected, SubscribeError>( + handle)); + }); + + // First subscriber + auto consumer1 = createMockConsumer(); + auto handle1 = + subscribeToTrack(subscriber1, kTestTrackName, consumer1, RequestID(1)); + ASSERT_NE(handle1, nullptr); + auto dynGroups1 = getPublisherDynamicGroups(handle1->subscribeOk()); + ASSERT_TRUE(dynGroups1.has_value()); + EXPECT_TRUE(*dynGroups1); + + // Late-joining second subscriber – forwarder should propagate the stored + // dynamic groups value without another upstream roundtrip + auto consumer2 = createMockConsumer(); + auto handle2 = + subscribeToTrack(subscriber2, kTestTrackName, consumer2, RequestID(2)); + ASSERT_NE(handle2, nullptr); + auto dynGroups2 = getPublisherDynamicGroups(handle2->subscribeOk()); + ASSERT_TRUE(dynGroups2.has_value()); + EXPECT_TRUE(*dynGroups2); + + removeSession(publisherSession); + removeSession(subscriber1); + removeSession(subscriber2); +} + // Test: Extensions are included in SubscribeOk for subscribe path subscribers TEST_F(MoQRelayTest, ExtensionsIncludedInSubscribeOkForSubscribers) { auto session1 = createMockSession(); @@ -2275,4 +2367,325 @@ TEST_F(MoQRelayTest, TrackStatusViaPrefixMatching) { removeSession(requester); } +// ============================================================ +// New Group Request (NGR) Tests +// ============================================================ + +namespace { +// Simple callback that records every newGroupRequested call. +struct TestNGRCallback : public MoQForwarder::Callback { + void onEmpty(MoQForwarder*) override {} + void newGroupRequested(MoQForwarder*, uint64_t group) override { + calls.push_back(group); + } + std::vector calls; +}; + +// Build a minimal params object carrying NEW_GROUP_REQUEST=val. +auto makeNGRParams(uint64_t val) { + RequestUpdate upd; + upd.params.insertParam(Parameter( + folly::to_underlying(TrackRequestParamKey::NEW_GROUP_REQUEST), val)); + return upd.params; +} +} // namespace + +// Forwarder unit test: tryProcessNewGroupRequest gating and clearing logic. +// Verifies all three gates (dynamic groups, largest visibility, outstanding +// deduplication) and that updateLargest clears the outstanding request. +TEST_F(MoQRelayTest, ForwarderNGRGatingAndClearingLogic) { + auto cb = std::make_shared(); + auto forwarder = std::make_shared(kTestTrackName); + forwarder->setCallback(cb); + + // Gate 1: dynamic groups not enabled — never fires + forwarder->tryProcessNewGroupRequest(makeNGRParams(5)); + EXPECT_TRUE(cb->calls.empty()); + + // Enable dynamic groups + PublishRequest pub; + setPublisherDynamicGroups(pub, true); + forwarder->setExtensions(pub.extensions); + + // Baseline: no largest, no outstanding — fires for group 5 + forwarder->tryProcessNewGroupRequest(makeNGRParams(5)); + ASSERT_EQ(cb->calls.size(), 1u); + EXPECT_EQ(cb->calls[0], 5u); + cb->calls.clear(); + + // Deduplication: outstanding=5 blocks re-requesting group 5 + forwarder->tryProcessNewGroupRequest(makeNGRParams(5)); + EXPECT_TRUE(cb->calls.empty()); + + // Gate 2: setLargest(10) blocks groups <= 10; group 11 fires (outstanding=5 + // is superseded) + forwarder->setLargest({10, 0}); + forwarder->tryProcessNewGroupRequest(makeNGRParams(5)); + forwarder->tryProcessNewGroupRequest(makeNGRParams(10)); + EXPECT_TRUE(cb->calls.empty()); + forwarder->tryProcessNewGroupRequest(makeNGRParams(11)); + ASSERT_EQ(cb->calls.size(), 1u); + EXPECT_EQ(cb->calls[0], 11u); + cb->calls.clear(); + + // Gate 3: outstanding=11 blocks values <= 11; group 12 fires + forwarder->tryProcessNewGroupRequest(makeNGRParams(11)); + forwarder->tryProcessNewGroupRequest(makeNGRParams(10)); + EXPECT_TRUE(cb->calls.empty()); + forwarder->tryProcessNewGroupRequest(makeNGRParams(12)); + ASSERT_EQ(cb->calls.size(), 1u); + EXPECT_EQ(cb->calls[0], 12u); + cb->calls.clear(); + + // updateLargest past outstanding clears it — group 14 can fire again + forwarder->updateLargest(13, 0); + forwarder->tryProcessNewGroupRequest(makeNGRParams(14)); + ASSERT_EQ(cb->calls.size(), 1u); + EXPECT_EQ(cb->calls[0], 14u); +} + + +// Relay test: When a late-joining subscriber sends NEW_GROUP_REQUEST in its +// SUBSCRIBE, the relay forwards it upstream via REQUEST_UPDATE +TEST_F(MoQRelayTest, RelaySubscribeLateJoinerNGRForwardedUpstream) { + auto publisherSession = createMockSession(); + auto subscriber1 = createMockSession(); + auto subscriber2 = createMockSession(); + + doPublishNamespace(publisherSession, kTestNamespace); + + // Upstream SubscribeOk advertises DYNAMIC_GROUPS = true + SubscribeOk upstreamOk; + upstreamOk.requestID = RequestID(100); + upstreamOk.trackAlias = TrackAlias(1); + upstreamOk.expires = std::chrono::milliseconds(0); + upstreamOk.groupOrder = GroupOrder::OldestFirst; + setPublisherDynamicGroups(upstreamOk, true); + + auto upstreamHandle = + std::make_shared>(upstreamOk); + ON_CALL(*upstreamHandle, requestUpdateResult()) + .WillByDefault( + Return(folly::makeExpected(RequestOk{RequestID(0)}))); + + EXPECT_CALL(*publisherSession, subscribe(_, _)) + .WillOnce([&upstreamHandle](auto /*req*/, auto /*consumer*/) { + return folly::coro::makeTask( + folly::Expected, SubscribeError>( + upstreamHandle)); + }); + + // First subscriber establishes the upstream subscription (no NGR) + auto consumer1 = createMockConsumer(); + auto handle1 = + subscribeToTrack(subscriber1, kTestTrackName, consumer1, RequestID(1)); + ASSERT_NE(handle1, nullptr); + + // Second subscriber includes NEW_GROUP_REQUEST=8 + auto consumer2 = createMockConsumer(); + SubscribeRequest sub2; + sub2.fullTrackName = kTestTrackName; + sub2.requestID = RequestID(2); + sub2.locType = LocationType::LargestObject; + sub2.params.insertParam(Parameter( + folly::to_underlying(TrackRequestParamKey::NEW_GROUP_REQUEST), + uint64_t(8))); + + // The relay must forward NGR=8 upstream via REQUEST_UPDATE + EXPECT_CALL(*upstreamHandle, requestUpdateCalled(_)) + .WillOnce([](RequestUpdate update) { + auto ngrValue = getFirstIntParam( + update.params, TrackRequestParamKey::NEW_GROUP_REQUEST); + ASSERT_TRUE(ngrValue.has_value()); + EXPECT_EQ(*ngrValue, 8); + }); + + std::shared_ptr handle2{nullptr}; + withSessionContext(subscriber2, [&]() { + auto task = relay_->subscribe(std::move(sub2), consumer2); + auto res = folly::coro::blockingWait(std::move(task), exec_.get()); + ASSERT_TRUE(res.hasValue()); + handle2 = *res; + }); + exec_->drive(); + + // Register handle2 so removeSession(subscriber2) will unsubscribe it, + // allowing the forwarder to become empty and release the upstream handle. + getOrCreateMockState(subscriber2)->subscribeHandles.push_back(handle2); + + removeSession(publisherSession); + removeSession(subscriber1); + removeSession(subscriber2); + exec_->drive(); +} + +// Relay test: A downstream subscriber sending REQUEST_UPDATE with +// NEW_GROUP_REQUEST causes the relay to cascade the NGR upstream +TEST_F(MoQRelayTest, RelayRequestUpdateNGRCascadedUpstream) { + auto publisherSession = createMockSession(); + auto subscriberSession = createMockSession(); + + doPublishNamespace(publisherSession, kTestNamespace); + + // Upstream SubscribeOk advertises DYNAMIC_GROUPS = true + SubscribeOk upstreamOk; + upstreamOk.requestID = RequestID(100); + upstreamOk.trackAlias = TrackAlias(1); + upstreamOk.expires = std::chrono::milliseconds(0); + upstreamOk.groupOrder = GroupOrder::OldestFirst; + setPublisherDynamicGroups(upstreamOk, true); + + auto upstreamHandle = + std::make_shared>(upstreamOk); + ON_CALL(*upstreamHandle, requestUpdateResult()) + .WillByDefault( + Return(folly::makeExpected(RequestOk{RequestID(0)}))); + + EXPECT_CALL(*publisherSession, subscribe(_, _)) + .WillOnce([&upstreamHandle](auto /*req*/, auto /*consumer*/) { + return folly::coro::makeTask( + folly::Expected, SubscribeError>( + upstreamHandle)); + }); + + // Subscribe downstream session - triggers upstream subscribe + auto consumer = createMockConsumer(); + auto handle = + subscribeToTrack(subscriberSession, kTestTrackName, consumer, RequestID(1)); + ASSERT_NE(handle, nullptr); + + auto* subscriber = dynamic_cast(handle.get()); + ASSERT_NE(subscriber, nullptr); + + // The relay must cascade NGR=9 upstream via REQUEST_UPDATE + EXPECT_CALL(*upstreamHandle, requestUpdateCalled(_)) + .WillOnce([](RequestUpdate update) { + auto ngrValue = getFirstIntParam( + update.params, TrackRequestParamKey::NEW_GROUP_REQUEST); + ASSERT_TRUE(ngrValue.has_value()); + EXPECT_EQ(*ngrValue, 9); + }); + + // Downstream subscriber sends REQUEST_UPDATE carrying NEW_GROUP_REQUEST=9 + RequestUpdate update; + update.requestID = RequestID(2); + update.existingRequestID = RequestID(1); + update.params.insertParam(Parameter( + folly::to_underlying(TrackRequestParamKey::NEW_GROUP_REQUEST), + uint64_t(9))); + folly::coro::blockingWait(subscriber->requestUpdate(std::move(update))); + exec_->drive(); + + removeSession(publisherSession); + removeSession(subscriberSession); +} + +// Unit test: Subscriber::onPublishOk postprocessing +// Verifies that onPublishOk correctly updates: +// 1. Subscriber range based on PublishOk fields +// 2. shouldForward flag +// 3. NEW_GROUP_REQUEST forwarding when it passes gating checks +TEST_F(MoQRelayTest, SubscriberOnPublishOkPostprocessing) { + auto publisherSession = createMockSession(); + auto subscriberSession = createMockSession(); + + // Publish track and subscribe + auto publishConsumer = doPublish(publisherSession, kTestTrackName); + auto consumer = createMockConsumer(); + auto handle = + subscribeToTrack(subscriberSession, kTestTrackName, consumer, RequestID(1)); + ASSERT_NE(handle, nullptr); + + // Cast to access Subscriber internals + auto* subscriber = dynamic_cast(handle.get()); + ASSERT_NE(subscriber, nullptr); + + // Test 1: Range update from PublishOk + // Create a PublishOk with specific start/end locations + PublishOk pubOk1{ + RequestID(1), // requestID + true, // forward + 0, // subscriberPriority + GroupOrder::OldestFirst, // groupOrder + LocationType::AbsoluteStart, // locType + AbsoluteLocation{5, 0}, // start + uint64_t(15), // endGroup + TrackRequestParameters(FrameType::PUBLISH_OK)}; + + // Apply the postprocessing + subscriber->onPublishOk(pubOk1); + + // Verify range was updated + EXPECT_EQ(subscriber->range.start.group, 5); + + // Test 2: Forward flag update + subscriber->shouldForward = true; + PublishOk pubOk2{ + RequestID(2), + false, // forward = false + 0, + GroupOrder::OldestFirst, + LocationType::AbsoluteStart, + AbsoluteLocation{5, 0}, + uint64_t(15), + TrackRequestParameters(FrameType::PUBLISH_OK)}; + subscriber->onPublishOk(pubOk2); + EXPECT_FALSE(subscriber->shouldForward) + << "Forward flag should be updated to false"; + + subscriber->shouldForward = false; + PublishOk pubOk3{ + RequestID(3), + true, // forward = true + 0, + GroupOrder::OldestFirst, + LocationType::AbsoluteStart, + AbsoluteLocation{5, 0}, + uint64_t(15), + TrackRequestParameters(FrameType::PUBLISH_OK)}; + subscriber->onPublishOk(pubOk3); + EXPECT_TRUE(subscriber->shouldForward) + << "Forward flag should be updated to true"; + + // Test 3: NEW_GROUP_REQUEST forwarding via onPublishOk + // Enable dynamic groups and attach a callback to observe NGR fires + PublishRequest pub; + setPublisherDynamicGroups(pub, true); + subscriber->forwarder.setExtensions(pub.extensions); + + auto cb = std::make_shared(); + subscriber->forwarder.setCallback(cb); + + // Build a PublishOk carrying NEW_GROUP_REQUEST=20 + TrackRequestParameters ngrParams(FrameType::PUBLISH_OK); + ngrParams.insertParam(Parameter( + folly::to_underlying(TrackRequestParamKey::NEW_GROUP_REQUEST), + uint64_t(20))); + PublishOk pubOk4{ + RequestID(4), + true, + 0, + GroupOrder::OldestFirst, + LocationType::AbsoluteStart, + AbsoluteLocation{10, 0}, + std::nullopt, + std::move(ngrParams)}; + + // onPublishOk should fire the NGR callback for group 20 + subscriber->onPublishOk(pubOk4); + ASSERT_EQ(cb->calls.size(), 1u); + EXPECT_EQ(cb->calls[0], 20u); + cb->calls.clear(); + + // outstanding=20: re-requesting group 20 is a no-op; group 21 fires + subscriber->forwarder.tryProcessNewGroupRequest(makeNGRParams(20)); + EXPECT_TRUE(cb->calls.empty()) << "Group 20 already outstanding"; + subscriber->forwarder.tryProcessNewGroupRequest(makeNGRParams(21)); + ASSERT_EQ(cb->calls.size(), 1u); + EXPECT_EQ(cb->calls[0], 21u); + + removeSession(publisherSession); + removeSession(subscriberSession); +} + } // namespace moxygen::test