Implement dynamic groups and new group request in relay (#112)

Summary:
Add support for DYNAMIC_GROUPS track extension and NEW_GROUP_REQUEST (NGR) parameter forwarding through the relay.

- MoQForwarder tracks outstanding NGR and clears it when upstream largest group advances; exposes tryProcessNewGroupRequest() to consolidate the extract/gate/set/fire pattern used at all call sites
- MoQRelay forwards NGR upstream via REQUEST_UPDATE when a downstream subscriber requests a new group on an existing upstream subscription; passes NGR through on new upstream subscribes without an extra update
- Track status returns local forwarder state when an active subscription exists, falling back to upstream only when no subscription is found

Co-authored by: akash-a-n

Pull Request resolved: #112

Reviewed By: sandarsh

Differential Revision: D95894248

Pulled By: afrind

fbshipit-source-id: cdda93666230a7931e41f058672e2e62bb530da7

Author: akash-a-n

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;
@@ -4458,6 +4459,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,
@@ -4544,6 +4555,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";
@@ -4879,6 +4900,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,

moxygen/MoQTypes.cpp

@@ -237,6 +237,11 @@ const folly::F14FastSet<FrameType> kAllowedFramesForForward = {
     FrameType::PUBLISH_OK,
     FrameType::SUBSCRIBE_NAMESPACE};
 
+const folly::F14FastSet<FrameType> 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<TrackRequestParamKey, folly::F14FastSet<FrameType>>
@@ -254,6 +259,8 @@ const folly::F14FastMap<TrackRequestParamKey, folly::F14FastSet<FrameType>>
         {TrackRequestParamKey::GROUP_ORDER, kAllowedFramesForGroupOrder},
         {TrackRequestParamKey::LARGEST_OBJECT, kAllowedFramesForLargestObject},
         {TrackRequestParamKey::FORWARD, kAllowedFramesForForward},
+        {TrackRequestParamKey::NEW_GROUP_REQUEST,
+         kAllowedFramesForNewGroupRequest},
 };
 
 // Frame types that allow all parameters (no validation)

moxygen/MoQTypes.h

@@ -445,6 +445,7 @@ enum class TrackRequestParamKey : uint64_t {
   GROUP_ORDER = 0x22,
   LARGEST_OBJECT = 0x9,
   FORWARD = 0x10,
+  NEW_GROUP_REQUEST = 0x32,
 };
 
 class Parameters {

moxygen/relay/CMakeLists.txt

@@ -27,6 +27,7 @@ moxygen_add_library(moxygen_relay_moq_cache
     Folly::folly_container_f14_hash
     Folly::folly_coro_baton
     Folly::folly_coro_task
+    Folly::folly_logging_logging
 )
 
 moxygen_add_library(moxygen_relay_moq_relay
@@ -35,6 +36,7 @@ moxygen_add_library(moxygen_relay_moq_relay
   EXPORTED_DEPS
     moxygen_moq
     moxygen_moq_consumers
+    moxygen_moq_types
     moxygen_relay_moq_cache
     moxygen_relay_moq_forwarder
     Folly::folly_container_f14_hash

moxygen/relay/MoQForwarder.cpp

@@ -361,6 +361,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 +534,34 @@ void MoQForwarder::removeForwardingSubscriber() {
   }
 }
 
+void MoQForwarder::tryProcessNewGroupRequest(
+    const Parameters& params,
+    bool fire) {
+  auto value =
+      getFirstIntParam(params, TrackRequestParamKey::NEW_GROUP_REQUEST);
+  if (!value.has_value()) {
+    return;
+  }
+  // the track must support dynamic groups.
+  if (!getPublisherDynamicGroups(extensions_).value_or(false)) {
+    return;
+  }
+  // non-zero value <= LargestGroup means the new group is already
+  // available — do not send upstream.
+  if (*value != 0 && largest_.has_value() && *value <= largest_->group) {
+    return;
+  }
+  // already have an outstanding request with >= value.
+  if (outstandingNewGroupRequest_.has_value() &&
+      *outstandingNewGroupRequest_ >= *value) {
+    return;
+  }
+  outstandingNewGroupRequest_ = *value;
+  if (fire && callback_) {
+    callback_->newGroupRequested(this, *value);
+  }
+}
+
 Payload MoQForwarder::maybeClone(const Payload& payload) {
   return payload ? payload->clone() : nullptr;
 }
@@ -597,13 +630,30 @@ void MoQForwarder::Subscriber::setParam(const TrackRequestParameter& param) {
   }
 }
 
+void MoQForwarder::Subscriber::onPublishOk(const PublishOk& pubOk) {
+  // Update subscriber range from PUBLISH_OK
+  std::optional<AbsoluteLocation> 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<folly::Expected<RequestOk, RequestError>>
 MoQForwarder::Subscriber::requestUpdate(RequestUpdate requestUpdate) {
   // Validation:
   // - Start location can be updated
   // - 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 +689,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};
 }
 

moxygen/relay/MoQForwarder.h

@@ -49,6 +49,12 @@ 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<AbsoluteLocation> largest() {
@@ -60,6 +66,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> callback);
@@ -107,6 +115,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<folly::Expected<RequestOk, RequestError>> requestUpdate(
         RequestUpdate requestUpdate) override;
 
@@ -317,6 +329,9 @@ class MoQForwarder : public TrackConsumer {
   GroupOrder groupOrder_{GroupOrder::OldestFirst};
   std::optional<AbsoluteLocation> 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<uint64_t> outstandingNewGroupRequest_{};
   std::shared_ptr<Callback> callback_;
   uint64_t forwardingSubscribers_{0};
   bool draining_{false};

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<void> MoQRelay::doSubscribeUpdate(
   }
 }
 
+// Sends a REQUEST_UPDATE that carries only the NEW_GROUP_REQUEST parameter.
+folly::coro::Task<void> MoQRelay::doNewGroupRequestUpdate(
+    std::shared_ptr<Publisher::SubscriptionHandle> 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::NamespaceNode> MoQRelay::findNamespaceNode(
     const TrackNamespace& ns,
     bool createMissingNodes,
@@ -509,13 +528,10 @@ folly::coro::Task<void> MoQRelay::publishToSession(
   guard.dismiss();
   XLOG(DBG1) << "Publish OK sess=" << session.get();
   auto& pubOk = pubResult.value().value();
-  std::optional<AbsoluteLocation> 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<Publisher::SubscribeResult> 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<Publisher::SubscribeResult> MoQRelay::subscribe(
           doSubscribeUpdate(subscriptionIt->second.handle, /*forward=*/true))
           .start();
     }
+
+    forwarder->tryProcessNewGroupRequest(subReq.params);
     co_return subscriber;
   }
 }
@@ -1110,4 +1130,24 @@ 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();
+  auto handle = subscription.handle;
+  co_withExecutor(exec, doNewGroupRequestUpdate(std::move(handle), group))
+      .start();
+}
+
 } // namespace moxygen

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<void> publishNamespaceToSession(
       std::shared_ptr<MoQSession> session,
@@ -203,6 +204,10 @@ class MoQRelay : public Publisher,
       std::shared_ptr<Publisher::SubscriptionHandle> handle,
       bool forward);
 
+  folly::coro::Task<void> doNewGroupRequestUpdate(
+      std::shared_ptr<Publisher::SubscriptionHandle> handle,
+      uint64_t newGroupRequestValue);
+
   void publishNamespaceDone(
       const TrackNamespace& trackNamespace,
       NamespaceNode* node);