Resolve publisher priority from track property extensions

Summary: Draft-16 moved PUBLISHER_PRIORITY from a request parameter to a track property extension, but setPublisherPriorityFromParams only ever looked at params. On draft-16+ that means the publisher never learns the priority it advertised, so it never elides it from subgroup headers or datagrams, and a subscriber talking to a peer that does elide it substitutes the default 128 for the real value. Both overloads now take the message's extensions alongside its params and check the extension first. The param fallback stays for a locally built message whose application inserted the raw param instead of calling the setter.

Reviewed By: sandarsh

Differential Revision: D116502271

fbshipit-source-id: 1935b9926c85f02bf7d38bdd914939e999633f87

Author: afrind

moxygen/MoQSession.cpp

@@ -4305,7 +4305,8 @@ folly::coro::Task<void> MoQSession::handleSubscribe(
     auto subOk = subHandle->subscribeOk();
     subOk.requestID = requestID;
 
-    setPublisherPriorityFromParams(subOk.params, trackPublisher);
+    setPublisherPriorityFromParams(
+        subOk.params, subOk.extensions, trackPublisher);
     trackPublisher->subscribeOkSent(subOk);
 
     // TODO: verify TrackAlias is unique
@@ -4314,43 +4315,64 @@ folly::coro::Task<void> MoQSession::handleSubscribe(
   }
 }
 
+// Extensions are canonical; the param fallback only fires for a locally built
+// message that set the raw param.
+static std::optional<uint64_t> getPublisherPriorityProperty(
+    const Extensions& trackProperties,
+    const TrackRequestParameters& params) {
+  auto publisherPriority = getPublisherPriority(trackProperties);
+  if (publisherPriority) {
+    return *publisherPriority;
+  }
+  return getFirstIntParam(params, TrackRequestParamKey::PUBLISHER_PRIORITY);
+}
+
+// nullopt leaves the priority unset, which disables elision on the publisher.
+static std::optional<uint8_t> toPublisherPriority(uint64_t advertised) {
+  if (advertised > kMaxPriority) {
+    XLOG(WARN) << "Invalid publisher priority value: " << advertised;
+    return std::nullopt;
+  }
+  return static_cast<uint8_t>(advertised);
+}
+
 void MoQSession::setPublisherPriorityFromParams(
     const TrackRequestParameters& params,
+    const Extensions& trackProperties,
     const std::shared_ptr<TrackPublisherImpl>& trackPublisher) {
-  // Extract PUBLISHER_PRIORITY parameter if present (version 15+)
-  if (getDraftMajorVersion(*getNegotiatedVersion()) >= 15) {
-    auto publisherPriority =
-        getFirstIntParam(params, TrackRequestParamKey::PUBLISHER_PRIORITY);
-    if (publisherPriority) {
-      if (*publisherPriority <= 255) {
-        trackPublisher->setPublisherPriority(
-            static_cast<uint8_t>(*publisherPriority));
-      } else {
-        XLOG(WARN) << "Invalid priority value: "
-                   << static_cast<uint64_t>(*publisherPriority);
-      }
-    } else {
-      trackPublisher->setPublisherPriority(kDefaultPriority);
-    }
+  if (getDraftMajorVersion(*getNegotiatedVersion()) < 15) {
+    return;
+  }
+  auto advertised = getPublisherPriorityProperty(trackProperties, params);
+  if (!advertised) {
+    // Naming the default lets objects published at it elide the priority byte.
+    trackPublisher->setPublisherPriority(kDefaultPriority);
+    return;
+  }
+  auto publisherPriority = toPublisherPriority(*advertised);
+  if (publisherPriority) {
+    trackPublisher->setPublisherPriority(*publisherPriority);
   }
 }
 
 void MoQSession::setPublisherPriorityFromParams(
     const TrackRequestParameters& params,
+    const Extensions& trackProperties,
     const std::shared_ptr<SubscribeTrackReceiveState>& trackReceiveState) {
-  // Extract PUBLISHER_PRIORITY parameter if present (version 15+)
-  if (getDraftMajorVersion(*getNegotiatedVersion()) >= 15) {
-    auto publisherPriority =
-        getFirstIntParam(params, TrackRequestParamKey::PUBLISHER_PRIORITY);
-    if (publisherPriority) {
-      if (*publisherPriority <= 255) {
-        trackReceiveState->setPublisherPriority(
-            static_cast<uint8_t>(*publisherPriority));
-      } else {
-        XLOG(ERR) << "Invalid priority value: "
-                  << static_cast<uint64_t>(*publisherPriority);
-      }
-    }
+  applyResolvedPublisherPriority(
+      getPublisherPriorityProperty(trackProperties, params), trackReceiveState);
+}
+
+// No default needed here; the receive state reads unset as kDefaultPriority.
+void MoQSession::applyResolvedPublisherPriority(
+    std::optional<uint64_t> advertised,
+    const std::shared_ptr<SubscribeTrackReceiveState>& trackReceiveState) {
+  if (getDraftMajorVersion(*getNegotiatedVersion()) < 15 || !advertised) {
+    return;
+  }
+  auto publisherPriority = toPublisherPriority(*advertised);
+  if (publisherPriority) {
+    trackReceiveState->setPublisherPriority(*publisherPriority);
   }
 }
 
@@ -4686,7 +4708,8 @@ void MoQSession::onSubscribeOk(SubscribeOk subOk) {
     close(SessionCloseErrorCode::DUPLICATE_TRACK_ALIAS);
   }
   auto trackAlias = subOk.trackAlias;
-  setPublisherPriorityFromParams(subOk.params, trackReceiveState);
+  setPublisherPriorityFromParams(
+      subOk.params, subOk.extensions, trackReceiveState);
   trackReceiveState->processSubscribeOK(std::move(subOk));
   if (trackReceiveState->getSubscribeCallback()) {
     trackReceiveState->getSubscribeCallback()->setTrackAlias(trackAlias);
@@ -4884,11 +4907,12 @@ folly::coro::Task<void> MoQSession::handlePublish(
   co_await folly::coro::co_safe_point;
   folly::RequestContextScopeGuard guard;
   setRequestSession();
-  // Capture params before moving publish
+  // Read what we need before publish is moved; copying Extensions clones IOBufs
   auto requestID = publish.requestID;
   auto alias = publish.trackAlias;
   auto ftn = publish.fullTrackName;
-  auto params = publish.params;
+  auto publisherPriority =
+      getPublisherPriorityProperty(publish.extensions, publish.params);
 
   // PubError if error occurs
   auto publishErr =
@@ -4925,8 +4949,7 @@ folly::coro::Task<void> MoQSession::handlePublish(
         auto trackReceiveState = std::make_shared<SubscribeTrackReceiveState>(
             ftn, requestID, initiator.consumer, this, alias, logger_, true);
 
-        // Extract PUBLISHER_PRIORITY parameter if present (version 15+)
-        setPublisherPriorityFromParams(params, trackReceiveState);
+        applyResolvedPublisherPriority(publisherPriority, trackReceiveState);
 
         initiator.consumer->setTrackAlias(alias);
         subTracks_.emplace(alias, trackReceiveState);
@@ -5859,8 +5882,7 @@ Subscriber::PublishResult MoQSession::publish(
   // Set publishHandle in trackPublisher so it can cancel on unsubscribes
   trackPublisher->setSubscriptionHandle(std::move(handle));
 
-  // Extract PUBLISHER_PRIORITY parameter if present (version 15+)
-  setPublisherPriorityFromParams(pub.params, trackPublisher);
+  setPublisherPriorityFromParams(pub.params, pub.extensions, trackPublisher);
 
   // Set reply context so sendPublishDone can write on the correct stream
   auto& control = sendResult.value();

moxygen/MoQSession.h

@@ -826,10 +826,15 @@ class MoQSession : public Subscriber,
       const SetupParameters& params);
   void setPublisherPriorityFromParams(
       const TrackRequestParameters& params,
+      const Extensions& trackProperties,
       const std::shared_ptr<TrackPublisherImpl>& trackPublisher);
   void setPublisherPriorityFromParams(
       const TrackRequestParameters& params,
+      const Extensions& trackProperties,
       const std::shared_ptr<SubscribeTrackReceiveState>& trackPublisher);
+  void applyResolvedPublisherPriority(
+      std::optional<uint64_t> advertised,
+      const std::shared_ptr<SubscribeTrackReceiveState>& trackReceiveState);
 
  protected:
   // Protected members and methods for MoQRelaySession subclass access

moxygen/MoQTrackProperties.h

@@ -47,7 +47,8 @@ std::optional<uint64_t> getIntExtension(const Msg& msg, uint64_t type) {
   return msg.extensions.getIntExtension(type);
 }
 
-// Overload for Extensions directly
+// Lets the getters below be called with a bare Extensions, which the relay
+// does when it only has the track properties and not the message.
 inline std::optional<uint64_t> getIntExtension(
     const Extensions& extensions,
     uint64_t type) {
@@ -142,14 +143,25 @@ std::optional<std::chrono::milliseconds> getPublisherMaxCacheDuration(
   return std::nullopt;
 }
 
-// PUBLISHER_PRIORITY getter
+// PUBLISHER_PRIORITY getter.  A value too wide for the wire is logged and
+// reported as absent, so callers fall back as they would for a peer that
+// advertised nothing.
+inline std::optional<uint8_t> getPublisherPriority(
+    const Extensions& extensions) {
+  auto val = extensions.getIntExtension(kPublisherPriorityExtensionType);
+  if (!val) {
+    return std::nullopt;
+  }
+  if (*val > kMaxPriority) {
+    XLOG(WARN) << "Invalid publisher priority extension: " << *val;
+    return std::nullopt;
+  }
+  return static_cast<uint8_t>(*val);
+}
+
 template <typename Msg>
 std::optional<uint8_t> getPublisherPriority(const Msg& msg) {
-  auto val = detail::getIntExtension(msg, kPublisherPriorityExtensionType);
-  if (val && *val <= 255) {
-    return static_cast<uint8_t>(*val);
-  }
-  return std::nullopt;
+  return getPublisherPriority(msg.extensions);
 }
 
 // GROUP_ORDER getter

moxygen/MoQTypes.h

@@ -21,6 +21,7 @@ namespace moxygen {
 //////// Constants ////////
 const size_t kMaxNamespaceLength = 32;
 const uint8_t kDefaultPriority = 128;
+const uint64_t kMaxPriority = 255;
 
 // From QUIC, until MOQ supports 64 bit varint
 constexpr uint64_t kEightByteLimit = 0x3FFFFFFFFFFFFFFF;

moxygen/test/MoQSessionObjectDeliveryTests.cpp

@@ -262,6 +262,110 @@ CO_TEST_P_X(MoQSessionTest, SubgroupWireFormatHintsRoundTrip) {
   co_await publishDone_;
   serverSession_->close(SessionCloseErrorCode::NO_ERROR);
 }
+CO_TEST_P_X(MoQSessionTest, AdvertisedPublisherPriorityRoundTrips) {
+  // The publisher advertises a non-default priority on the SubscribeOk (a
+  // track property extension from draft 16, a param before that) and then
+  // publishes at exactly that priority, so draft-15+ elides it from the
+  // subgroup header.  Both ends have to resolve the advertised value: if only
+  // the sender honours it, the subscriber backfills the protocol default and
+  // the two disagree about every object's priority.
+  constexpr uint8_t kAdvertisedPriority = 100;
+  auto majorVersion = getDraftMajorVersion(GetParam().serverVersion);
+  co_await setupMoQSession();
+  expectSubscribe(
+      [this, majorVersion, priority = kAdvertisedPriority](
+          auto sub, auto pub) -> TaskSubscribeResult {
+        eventBase_.add([pub, sub, priority] {
+          auto sgp = pub->beginSubgroup(0, 0, priority).value();
+          sgp->object(0, moxygen::test::makeBuf(10), noExtensions(), true);
+          pub->publishDone(getTrackEndedPublishDone(sub.requestID));
+        });
+        co_return makeSubscribeOkResult(
+            sub, std::nullopt, priority, majorVersion);
+      },
+      MoQControlCodec::Direction::CLIENT);
+
+  auto sg = std::make_shared<testing::StrictMock<MockSubgroupConsumer>>();
+  EXPECT_CALL(*subscribeCallback_, beginSubgroup(0, 0, kAdvertisedPriority, _))
+      .WillOnce(testing::Return(sg));
+  EXPECT_CALL(*sg, object(0, _, _, true))
+      .WillOnce(testing::Return(folly::unit));
+
+  expectPublishDone(MoQControlCodec::Direction::SERVER);
+  auto res = co_await serverSession_->subscribe(
+      getSubscribe(kTestTrackName), subscribeCallback_);
+  co_await publishDone_;
+  serverSession_->close(SessionCloseErrorCode::NO_ERROR);
+}
+CO_TEST_P_X(MoQSessionTest, ElidedPriorityResolvesToAdvertisedPriority) {
+  // A conforming peer that advertises a publisher priority may omit it from
+  // every subgroup header.  Write such a header by hand -- a moxygen publisher
+  // only elides once it has resolved the advertised value, which is the very
+  // thing under test -- and check the subscriber recovers the advertised
+  // priority rather than falling back to the protocol default.
+  constexpr uint8_t kAdvertisedPriority = 100;
+  auto version = GetParam().serverVersion;
+  if (getDraftMajorVersion(version) < 16) {
+    // Before draft 16 the priority rides a param, not a track property
+    co_return;
+  }
+  co_await setupMoQSession();
+
+  expectSubscribe(
+      [version, priority = kAdvertisedPriority](
+          auto sub, auto /*pub*/) -> TaskSubscribeResult {
+        co_return makeSubscribeOkResult(
+            sub,
+            AbsoluteLocation{0, 0},
+            priority,
+            getDraftMajorVersion(version));
+      },
+      MoQControlCodec::Direction::CLIENT);
+
+  auto sg = std::make_shared<testing::StrictMock<MockSubgroupConsumer>>();
+  folly::coro::Baton objectReceived;
+  EXPECT_CALL(*subscribeCallback_, beginSubgroup(0, 0, kAdvertisedPriority, _))
+      .WillOnce(testing::Return(sg));
+  EXPECT_CALL(*sg, object(0, _, _, _))
+      .WillOnce([&](auto, auto, const auto&, auto) {
+        objectReceived.post();
+        return folly::unit;
+      });
+  EXPECT_CALL(*sg, reset(_));
+
+  auto res = co_await serverSession_->subscribe(
+      getSubscribe(kTestTrackName), subscribeCallback_);
+  EXPECT_FALSE(res.hasError());
+
+  // Publisher side writes a subgroup whose header carries no priority
+  auto wh = CHECK_NOTNULL(clientWt_->createUniStream().value_or(nullptr));
+  folly::IOBufQueue dataBuf{folly::IOBufQueue::cacheChainLength()};
+  MoQFrameWriter writer;
+  writer.initializeVersion(version);
+  ObjectHeader objHeader(0, 0, 0, std::nullopt, ObjectStatus::NORMAL);
+  objHeader.length = 5;
+  writer.writeSubgroupHeader(
+      dataBuf,
+      res.value()->subscribeOk().trackAlias,
+      objHeader,
+      SubgroupIDFormat::Present,
+      false);
+  writer.writeStreamObject(
+      dataBuf,
+      getSubgroupStreamType(
+          version,
+          SubgroupIDFormat::Present,
+          false,
+          false,
+          /*priorityPresent=*/false),
+      objHeader,
+      makeBuf(5));
+  wh->writeStreamData(dataBuf.move(), false, nullptr);
+
+  co_await objectReceived;
+  res.value()->unsubscribe();
+  serverSession_->close(SessionCloseErrorCode::NO_ERROR);
+}
 CO_TEST_P_X(MoQSessionTest, MixSubgroupsAndDatagrams) {
   // Test that subgroups and datagrams can be mixed on the same track
   co_await setupMoQSession();