Google Git Google Git
Sign in
quiche/quiche/04e702203755f10261fa91d35da6c4e986a4b2ba^!/.
commit
04e702203755f10261fa91d35da6c4e986a4b2ba
[log]
author
martinduke <martinduke@google.com>
Mon Oct 05 17:20:34 2026 -0700
committer
Copybara-Service <copybara-worker@google.com>
Mon Oct 05 17:21:24 2026 -0700
tree
d5885fa74d3ab75fc81b265cba027be334be9343
parent
541aa2a7889b7ea7651e41afc58211f2a256850a [diff] 
Send LARGEST_OBJECT when REQUEST_UPDATE sets Forward=1 or moves the filter end back.

If forward is set to 1, re-initialize the subscription filter with the new LARGEST_OBJECT.

Also, fixes a problem where the publisher can respond to REQUEST_UPDATE before the initial SUBSCRIBE.

PiperOrigin-RevId: 994003251

diff --git a/quiche/quic/moqt/moqt_bidi_stream.cc b/quiche/quic/moqt/moqt_bidi_stream.cc
index 15167c8..461b636 100644
--- a/quiche/quic/moqt/moqt_bidi_stream.cc
+++ b/quiche/quic/moqt/moqt_bidi_stream.cc

@@ -86,7 +86,7 @@
   request_update.request_id = request_id;
   request_update.existing_request_id = existing_request_id;
   request_update.parameters = parameters;
-  request_update_queue_.Enqueue(parameters, std::move(callback));
+  outgoing_update_queue_.Enqueue(parameters, std::move(callback));
   return SendOrBufferMessage(framer_->SerializeRequestUpdate(request_update),
                              /*fin=*/false);
 }

diff --git a/quiche/quic/moqt/moqt_bidi_stream.h b/quiche/quic/moqt/moqt_bidi_stream.h
index 0e59ab9..4188d2a 100644
--- a/quiche/quic/moqt/moqt_bidi_stream.h
+++ b/quiche/quic/moqt/moqt_bidi_stream.h

@@ -26,6 +26,7 @@
 #include "quiche/common/platform/api/quiche_logging.h"
 #include "quiche/common/quiche_buffer_allocator.h"
 #include "quiche/common/quiche_callbacks.h"
+#include "quiche/common/quiche_circular_deque.h"
 #include "quiche/web_transport/web_transport.h"
 
 namespace moqt {
@@ -149,6 +150,17 @@
   webtransport::StreamId stream_id() const {
     return stream() != nullptr ? stream()->GetStreamId() : 0;
   }
+  // If there are incoming REQUEST_UPDATEs buffered for processing, returns the
+  // parameters saved for the oldest one. Also pops it from the queue.
+  // Otherwise, returns nullopt.
+  std::optional<MessageParameters> NextIncomingUpdate() {
+    if (incoming_update_queue_.empty()) {
+      return std::nullopt;
+    }
+    MessageParameters parameters = incoming_update_queue_.front();
+    incoming_update_queue_.pop_front();
+    return parameters;
+  }
 
   absl::Status OnControlMessage(const MoqtGoAway& message);
 
@@ -163,8 +175,12 @@
       const MoqtRawControlMessage& message) = 0;
 
   MoqtRequestUpdateQueue& request_update_queue() {
-    return request_update_queue_;
+    return outgoing_update_queue_;
   }
+  void QueueIncomingUpdate(const MessageParameters& parameters) {
+    incoming_update_queue_.push_back(parameters);
+  }
+  bool IncomingUpdatesQueued() const { return !incoming_update_queue_.empty(); }
 
   // Terminates the MoQT session due to a fatal error encountered.
   void OnFatalError(absl::Status status);
@@ -187,7 +203,10 @@
   std::optional<MoqtControlStreamParser> stream_parser_;
   MoqtControlMessageParser message_parser_;
   MoqtControlMessageQueue outgoing_message_queue_;
-  MoqtRequestUpdateQueue request_update_queue_;
+  MoqtRequestUpdateQueue outgoing_update_queue_;
+  // Incoming REQUEST_UPDATEs that cannot be handled synchronously (possibly
+  // because the OK for the original request has not been sent)
+  quiche::QuicheCircularDeque<MessageParameters> incoming_update_queue_;
   SessionErrorCallback session_error_callback_;
   absl::Status stream_status_ = absl::OkStatus();
   bool received_goaway_ = false;

diff --git a/quiche/quic/moqt/moqt_bidi_stream_test.cc b/quiche/quic/moqt/moqt_bidi_stream_test.cc
index e4b3973..e56ea76 100644
--- a/quiche/quic/moqt/moqt_bidi_stream_test.cc
+++ b/quiche/quic/moqt/moqt_bidi_stream_test.cc

@@ -58,6 +58,9 @@
     }
     return status;
   }
+
+  using MoqtBidiStreamBase::IncomingUpdatesQueued;
+  using MoqtBidiStreamBase::QueueIncomingUpdate;
 };
 
 class MoqtBidiStreamTest : public quiche::test::QuicheTest {
@@ -228,4 +231,25 @@
   EXPECT_FALSE(stream_->QueueIsFull());
 }
 
+TEST_F(MoqtBidiStreamTest, IncomingUpdateQueue) {
+  EXPECT_FALSE(stream_->IncomingUpdatesQueued());
+  EXPECT_EQ(stream_->NextIncomingUpdate(), std::nullopt);
+
+  MessageParameters update1;
+  update1.subscriber_priority = 10;
+  MessageParameters update2;
+  update2.subscriber_priority = 20;
+
+  stream_->QueueIncomingUpdate(update1);
+  EXPECT_TRUE(stream_->IncomingUpdatesQueued());
+  stream_->QueueIncomingUpdate(update2);
+  EXPECT_TRUE(stream_->IncomingUpdatesQueued());
+
+  EXPECT_EQ(stream_->NextIncomingUpdate(), update1);
+  EXPECT_TRUE(stream_->IncomingUpdatesQueued());
+  EXPECT_EQ(stream_->NextIncomingUpdate(), update2);
+  EXPECT_FALSE(stream_->IncomingUpdatesQueued());
+  EXPECT_EQ(stream_->NextIncomingUpdate(), std::nullopt);
+}
+
 }  // namespace moqt::test

diff --git a/quiche/quic/moqt/moqt_fetch_stream.cc b/quiche/quic/moqt/moqt_fetch_stream.cc
index 78b97d4..1bac6fc 100644
--- a/quiche/quic/moqt/moqt_fetch_stream.cc
+++ b/quiche/quic/moqt/moqt_fetch_stream.cc

@@ -297,6 +297,25 @@
         delivery_order, std::move(response_callback));
   } else {
     // Joining Fetch.
+    // TODO(martinduke): There are some gaps in this implementation that are
+    // relevant to draft-18 Joining FETCH but not later drafts, where it occurs
+    // on the SUBSCRIBE stream and therefore does not need a reference request
+    // ID:
+    // 1. draft-18 is ambiguous about whether to use the original request ID or
+    // the request ID of a REQUEST_UPDATE. Our current request ID tracking data
+    // structures and callbacks are not well-equipped to track multiple request
+    // IDs per subscription. There are good reasons to use the REQUEST_UPDATE
+    // ID: it can prevent a FETCH from failing because forward is set to zero.
+    // However, our draft-18 implementation will fail any joining FETCH that
+    // beats such a REQUEST_UPDATE.
+    // 2. It's hard to distinguish between a request ID that used to exist and
+    // one that has yet to arrive. It would be best to use
+    // MoqtSession::next_incoming_request_id to guess if the request is new or
+    // old, and buffer the stream if it's new.
+    //
+    // draft-18 is a candidate for long term support, but it may not be worth it
+    // to fix problems that don't exist in future drafts if joining FETCH is
+    // not important to early-deployed use cases.
     uint64_t joining_request_id =
         std::holds_alternative<JoiningFetchRelative>(message.fetch)
             ? std::get<JoiningFetchRelative>(message.fetch).joining_request_id

diff --git a/quiche/quic/moqt/moqt_key_value_pair.cc b/quiche/quic/moqt/moqt_key_value_pair.cc
index ce63ffe..1ff4b4a 100644
--- a/quiche/quic/moqt/moqt_key_value_pair.cc
+++ b/quiche/quic/moqt/moqt_key_value_pair.cc

@@ -13,6 +13,7 @@
 #include "absl/strings/string_view.h"
 #include "quiche/quic/core/quic_time.h"
 #include "quiche/quic/moqt/moqt_priority.h"
+#include "quiche/quic/moqt/moqt_types.h"
 #include "quiche/common/platform/api/quiche_bug_tracker.h"
 #include "quiche/common/platform/api/quiche_logging.h"
 
@@ -54,19 +55,22 @@
   switch (type_) {
     case MoqtFilterType::kAbsoluteStart:
     case MoqtFilterType::kAbsoluteRange:
-      return;
+      break;
     case MoqtFilterType::kNextGroupStart:
       if (largest_object.has_value()) {
         start_ = Location(largest_object->group + 1, 0);
+      } else {
+        start_ = Location(0, 0);
       }
       break;
     case MoqtFilterType::kLargestObject:
       if (largest_object.has_value()) {
         start_ = largest_object->Next();
+      } else {
+        start_ = Location(0, 0);
       }
       break;
   }
-  type_ = MoqtFilterType::kAbsoluteStart;
 }
 
 void MessageParameters::Update(const MessageParameters& other) {

diff --git a/quiche/quic/moqt/moqt_key_value_pair.h b/quiche/quic/moqt/moqt_key_value_pair.h
index 5eef4d5..5f158a3 100644
--- a/quiche/quic/moqt/moqt_key_value_pair.h
+++ b/quiche/quic/moqt/moqt_key_value_pair.h

@@ -124,23 +124,20 @@
   }
 
   MoqtFilterType type() const { return type_; }
-  Location start() const { return start_; }
+  Location start() const { return start_.value_or(Location(0, 0)); }
   uint64_t end_group() const { return end_group_; }
   // If true, the filter does not depend on knowing LargestObject, or we
   // already know LargestObject. A Subscriber cannot have an unknown window,
   // because it can't process an object without getting the track alias from
   // the SubscribeOk.
-  bool WindowKnown() const {
-    return type_ == MoqtFilterType::kAbsoluteStart ||
-           type_ == MoqtFilterType::kAbsoluteRange;
-  }
+  bool WindowKnown() const { return start_.has_value(); }
   // Publishers should not call InWindow() if !WindowKnown(), as they should
   // not forward objects without knowing the window.
   bool InWindow(Location location) const {
-    return location >= start_ && location.group <= end_group_;
+    return location >= start() && location.group <= end_group_;
   }
   bool InWindow(uint64_t group) const {
-    return group >= start_.group && group <= end_group_;
+    return (group >= start().group) && group <= end_group_;
   }
   // Update the filter when LargestObject is learned. If the type is
   // LargestObject or NextGroupStart, changes the type to AbsoluteStart.
@@ -150,8 +147,7 @@
 
  private:
   MoqtFilterType type_;
-  // These could be std::optional, but it would just add to the class size.
-  Location start_ = Location(0, 0);
+  std::optional<Location> start_;
   uint64_t end_group_ = kMaxGroupId;
 };
 

diff --git a/quiche/quic/moqt/moqt_key_value_pair_test.cc b/quiche/quic/moqt/moqt_key_value_pair_test.cc
index 9e21fd8..6400ef8 100644
--- a/quiche/quic/moqt/moqt_key_value_pair_test.cc
+++ b/quiche/quic/moqt/moqt_key_value_pair_test.cc

@@ -10,8 +10,8 @@
 #include "absl/status/status.h"
 #include "absl/strings/string_view.h"
 #include "quiche/quic/core/quic_time.h"
-#include "quiche/quic/moqt/moqt_error.h"
 #include "quiche/quic/moqt/moqt_priority.h"
+#include "quiche/quic/moqt/moqt_types.h"
 #include "quiche/quic/platform/api/quic_test.h"
 #include "quiche/common/test_tools/quiche_test_utils.h"
 
@@ -70,7 +70,7 @@
   EXPECT_FALSE(filter.WindowKnown());
   filter.OnLargestObject(Location(3, 6));
   EXPECT_TRUE(filter.WindowKnown());
-  EXPECT_EQ(filter.type(), (MoqtFilterType::kAbsoluteStart));
+  EXPECT_EQ(filter.type(), (MoqtFilterType::kNextGroupStart));
   EXPECT_TRUE(filter.InWindow(Location(4, 0)));
   EXPECT_FALSE(filter.InWindow(Location(3, 7)));
   EXPECT_TRUE(filter.InWindow(4));
@@ -82,7 +82,7 @@
   EXPECT_EQ(filter.type(), (MoqtFilterType::kLargestObject));
   EXPECT_FALSE(filter.WindowKnown());
   filter.OnLargestObject(Location(3, 6));
-  EXPECT_EQ(filter.type(), (MoqtFilterType::kAbsoluteStart));
+  EXPECT_EQ(filter.type(), (MoqtFilterType::kLargestObject));
   EXPECT_TRUE(filter.WindowKnown());
   EXPECT_TRUE(filter.InWindow(Location(4, 0)));
   EXPECT_TRUE(filter.InWindow(Location(3, 7)));
@@ -97,7 +97,7 @@
   EXPECT_EQ(filter.type(), (MoqtFilterType::kLargestObject));
   EXPECT_FALSE(filter.WindowKnown());
   filter.OnLargestObject(std::nullopt);
-  EXPECT_EQ(filter.type(), (MoqtFilterType::kAbsoluteStart));
+  EXPECT_EQ(filter.type(), (MoqtFilterType::kLargestObject));
   EXPECT_TRUE(filter.WindowKnown());
   EXPECT_TRUE(filter.InWindow(Location(0, 0)));
   EXPECT_TRUE(filter.InWindow(0));

diff --git a/quiche/quic/moqt/moqt_live_publisher.cc b/quiche/quic/moqt/moqt_live_publisher.cc
index a67ddad..ee32d49 100644
--- a/quiche/quic/moqt/moqt_live_publisher.cc
+++ b/quiche/quic/moqt/moqt_live_publisher.cc

@@ -15,6 +15,7 @@
 #include "absl/base/nullability.h"
 #include "absl/container/btree_map.h"
 #include "absl/container/flat_hash_set.h"
+#include "absl/status/status.h"
 #include "absl/strings/string_view.h"
 #include "quiche/quic/core/quic_clock.h"
 #include "quiche/quic/moqt/moqt_bidi_stream.h"
@@ -72,15 +73,39 @@
   }
 }
 
-void LivePublisher::Update(const MessageParameters& parameters) {
-  if (!track_publisher_->UpdateObjectListener(this, parameters).ok()) {
+void LivePublisher::Update(const MessageParameters& parameters,
+                           bool from_request_ok) {
+  absl::Status pub_status =
+      track_publisher_->UpdateObjectListener(this, parameters);
+  if (!pub_status.ok()) {
+    if (!from_request_ok) {
+      bidi_stream_->CheckStatus(
+          bidi_stream_->SendRequestError(StatusToMoqtRequestError(pub_status)));
+    } else {
+      bidi_stream_->Reset(StatusToMoqtStreamError(pub_status));
+    }
     return;
   }
+  MessageParameters response;
   // TODO(martinduke): If the subscribe window has shrunk, close any streams
   // that are now outside the window. Also send PUBLISH_DONE if now done.
   MoqtPriority old_priority =
       parameters_.subscriber_priority.value_or(kDefaultSubscriberPriority);
+  // Does this update set forward to true?
+  bool forward_to_1 = !parameters_.forward() &&
+                      parameters.forward_has_value() && parameters.forward();
+  response.largest_object = track_publisher_->largest_location();
+  parameters_.largest_object = response.largest_object;
   parameters_.Update(parameters);
+  if (parameters_.subscription_filter.has_value() &&
+      parameters_.largest_object.has_value() &&
+      (!parameters_.subscription_filter->WindowKnown() || forward_to_1)) {
+    // Initialize the filter if:
+    // - Forward is going 0->1 (this could be a re-initialization)
+    // - This the first LARGEST_OBJECT sent.
+    // - There's a new filter that has to be initialized from LARGEST_OBJECT.
+    parameters_.subscription_filter->OnLargestObject(response.largest_object);
+  }
   if (parameters.subscriber_priority.has_value()) {  // priority changed.
     MoqtPriority new_priority = *parameters.subscriber_priority;
     // Reprioritize all active streams.
@@ -93,21 +118,19 @@
           absl::down_cast<OutgoingSubgroupStream*>(stream->visitor());
       outgoing_stream->UpdatePriority(new_priority);
     }
-    if (pending_streams_.empty()) {
-      return;
+    if (!pending_streams_.empty() && visitor() != nullptr) {
+      // Tell the session that pending stream priority has changed.
+      MoqtPriority publisher_priority =
+          pending_streams_.rbegin()->second.publisher_priority.value_or(
+              track_publisher_->properties().default_publisher_priority());
+      MoqtTrackPriority old_track_priority = {old_priority, publisher_priority};
+      visitor()->UpdateTrackPriority(
+          track_publisher_->GetTrackName(), old_track_priority,
+          MoqtTrackPriority{new_priority, publisher_priority});
     }
-    // Tell the session that pending stream priority has changed.
-    MoqtPriority publisher_priority =
-        pending_streams_.rbegin()->second.publisher_priority.value_or(
-            track_publisher_->properties().default_publisher_priority());
-    MoqtTrackPriority old_track_priority = {old_priority, publisher_priority};
-    if (visitor() == nullptr) {
-      return;
-    }
-    visitor()->UpdateTrackPriority(
-        track_publisher_->GetTrackName(), old_track_priority,
-        MoqtTrackPriority{new_priority, publisher_priority});
-    // Don't bother to update all the pending stream send orders.
+  }
+  if (!from_request_ok) {
+    bidi_stream_->CheckStatus(bidi_stream_->SendRequestOk(response));
   }
 }
 
@@ -148,6 +171,19 @@
       subscribe_ok.properties.default_publisher_priority();
   bidi_stream_->SendOrBufferMessageOrFatal(
       framer_.SerializeSubscribeOk(subscribe_ok));
+  // It is now safe to process buffered REQUEST_UPDATEs.
+  std::optional<MessageParameters> next_update =
+      bidi_stream_->NextIncomingUpdate();
+  // Get a weak_pointer to LivePublisher in case it is destroyed by Update().
+  quiche::QuicheWeakPtr<LivePublisherInterface> self =
+      weak_ptr_factory_.Create();
+  while (self.IsValid() && next_update.has_value()) {
+    Update(*next_update, /*from_request_ok=*/false);
+    if (!self.IsValid()) {
+      return;
+    }
+    next_update = bidi_stream_->NextIncomingUpdate();
+  }
   // TODO(martinduke): If we buffer objects that arrived previously, the arrival
   // of the track alias disambiguates what subscription they belong to. Send
   // them.

diff --git a/quiche/quic/moqt/moqt_live_publisher.h b/quiche/quic/moqt/moqt_live_publisher.h
index 6bbdb00..c2a5c95 100644
--- a/quiche/quic/moqt/moqt_live_publisher.h
+++ b/quiche/quic/moqt/moqt_live_publisher.h

@@ -178,8 +178,9 @@
   // MoqtObjectPublisher implementation.
   void OnCanCreateNewUniStream();
 
-  // Called when the parameters_ needs an update.
-  void Update(const MessageParameters& parameters);
+  // Called when the parameters_ need an update and will generate REQUEST_OK or
+  // REQUEST_ERROR, unless from_request_ok is true (which requires no response).
+  void Update(const MessageParameters& parameters, bool from_request_ok);
 
   bool can_have_joining_fetch() const { return parameters_.forward(); }
 

diff --git a/quiche/quic/moqt/moqt_live_publisher_test.cc b/quiche/quic/moqt/moqt_live_publisher_test.cc
index ed67d2a..299a191 100644
--- a/quiche/quic/moqt/moqt_live_publisher_test.cc
+++ b/quiche/quic/moqt/moqt_live_publisher_test.cc

@@ -34,6 +34,7 @@
 #include "quiche/quic/moqt/moqt_trace_recorder.h"
 #include "quiche/quic/moqt/moqt_types.h"
 #include "quiche/quic/moqt/test_tools/mock_moqt_session.h"
+#include "quiche/quic/moqt/test_tools/moqt_framer_utils.h"
 #include "quiche/quic/moqt/test_tools/moqt_mock_visitor.h"
 #include "quiche/quic/moqt/test_tools/moqt_session_peer.h"
 #include "quiche/quic/platform/api/quic_test.h"
@@ -83,6 +84,7 @@
     return absl::OkStatus();
   }
   void Detach() override { detached_ = true; }
+  using MoqtBidiStreamBase::QueueIncomingUpdate;
   bool detached_ = false;
 };
 
@@ -291,6 +293,50 @@
   publisher_->OnNewObjectAvailable(Location(1, 3), std::nullopt, 64);
 }
 
+TEST_F(LivePublisherTest, OnSubscribeAcceptedWithQueuedUpdates) {
+  MessageParameters update1;
+  update1.delivery_timeout = quic::QuicTimeDelta::FromSeconds(5);
+  update1.set_forward(false);
+  bidi_stream_.QueueIncomingUpdate(update1);
+
+  MessageParameters update2;
+  update2.set_forward(true);
+  update2.subscription_filter.emplace(MoqtFilterType::kLargestObject);
+  bidi_stream_.QueueIncomingUpdate(update2);
+
+  EXPECT_CALL(mock_bidi_stream_, CanWrite()).WillRepeatedly(Return(true));
+  EXPECT_CALL(*track_publisher_, largest_location())
+      .WillOnce(Return(std::nullopt))
+      .WillOnce(Return(Location(1, 2)))
+      .WillOnce(Return(Location(2, 5)));
+  EXPECT_CALL(*track_publisher_, expiration)
+      .WillOnce(Return(quic::QuicTimeDelta::FromSeconds(10)));
+  EXPECT_CALL(*track_publisher_, properties)
+      .WillRepeatedly(ReturnRef(properties_));
+  EXPECT_CALL(mock_bidi_stream_,
+              Writev(ControlMessageOfType(MoqtMessageType::kSubscribeOk), _))
+      .WillOnce(Return(absl::OkStatus()));
+  MessageParameters expected_ok;
+  expected_ok.largest_object = Location(1, 2);
+  EXPECT_CALL(mock_bidi_stream_,
+              Writev(SerializedControlMessage(MoqtRequestOk(expected_ok)), _))
+      .WillOnce(Return(absl::OkStatus()));
+  expected_ok.largest_object = Location(2, 5);
+  EXPECT_CALL(mock_bidi_stream_,
+              Writev(SerializedControlMessage(MoqtRequestOk(expected_ok)), _))
+      .WillOnce(Return(absl::OkStatus()));
+
+  publisher_->OnSubscribeAccepted();
+  EXPECT_TRUE(publisher_->established());
+  EXPECT_EQ(publisher_->parameters().delivery_timeout,
+            quic::QuicTimeDelta::FromSeconds(5));
+  EXPECT_TRUE(publisher_->parameters().forward());
+  EXPECT_EQ(publisher_->parameters().largest_object, Location(2, 5));
+  ASSERT_TRUE(publisher_->parameters().subscription_filter.has_value());
+  EXPECT_EQ(publisher_->parameters().subscription_filter->start(),
+            Location(2, 6));
+}
+
 TEST_F(LivePublisherTest, OnSubscribeRejected) {
   EXPECT_CALL(mock_bidi_stream_, CanWrite()).WillRepeatedly(Return(true));
   EXPECT_CALL(mock_bidi_stream_,
@@ -303,19 +349,108 @@
 TEST_F(LivePublisherTest, Update) {
   MessageParameters new_params;
   new_params.delivery_timeout = quic::QuicTimeDelta::FromSeconds(5);
-  publisher_->Update(new_params);
+  EXPECT_CALL(mock_bidi_stream_, CanWrite()).WillRepeatedly(Return(true));
+  EXPECT_CALL(mock_bidi_stream_,
+              Writev(SerializedControlMessage(MoqtRequestOk()), _))
+      .WillOnce(Return(absl::OkStatus()));
+  publisher_->Update(new_params, false);
+  EXPECT_EQ(publisher_->parameters().delivery_timeout,
+            quic::QuicTimeDelta::FromSeconds(5));
 
   // Changing forward preference updates can_have_joining_fetch_
   new_params.set_forward(false);
-  publisher_->Update(new_params);
+  EXPECT_CALL(mock_bidi_stream_,
+              Writev(SerializedControlMessage(MoqtRequestOk()), _))
+      .WillOnce(Return(absl::OkStatus()));
+  publisher_->Update(new_params, false);
   EXPECT_FALSE(publisher_->parameters().forward());
   EXPECT_FALSE(publisher_->can_have_joining_fetch());
+
+  // Setting a LargestObject filter while forward=false, then turning forward=
+  // true initializes the filter using the latest largest_location.
+  publisher_->parameters().subscription_filter =
+      SubscriptionFilter(MoqtFilterType::kLargestObject);
+  MessageParameters forward_on_params;
+  forward_on_params.set_forward(true);
+  EXPECT_CALL(*track_publisher_, largest_location())
+      .WillOnce(Return(Location(2, 4)));
+  MessageParameters expected_ok;
+  expected_ok.largest_object = Location(2, 4);
+  EXPECT_CALL(mock_bidi_stream_,
+              Writev(SerializedControlMessage(MoqtRequestOk{expected_ok}), _))
+      .WillOnce(Return(absl::OkStatus()));
+  publisher_->Update(forward_on_params, false);
+  EXPECT_TRUE(publisher_->parameters().forward());
+  EXPECT_TRUE(publisher_->can_have_joining_fetch());
+  EXPECT_EQ(publisher_->parameters().largest_object, Location(2, 4));
+  ASSERT_TRUE(publisher_->parameters().subscription_filter.has_value());
+  EXPECT_EQ(publisher_->parameters().subscription_filter->start(),
+            Location(2, 5));
+}
+
+TEST_F(LivePublisherTest, UpdateSubscriptionFilterAndEndGroup) {
+  EXPECT_CALL(mock_bidi_stream_, CanWrite()).WillRepeatedly(Return(true));
+
+  // Adding a new kLargestObject filter when forward is already true initializes
+  // the filter and sends LARGEST_OBJECT in REQUEST_OK.
+  MessageParameters filter_params;
+  filter_params.subscription_filter.emplace(MoqtFilterType::kLargestObject);
+  EXPECT_CALL(*track_publisher_, largest_location())
+      .WillOnce(Return(Location(3, 1)));
+  MessageParameters expected_ok1;
+  expected_ok1.largest_object = Location(3, 1);
+  EXPECT_CALL(mock_bidi_stream_,
+              Writev(SerializedControlMessage(MoqtRequestOk{expected_ok1}), _))
+      .WillOnce(Return(absl::OkStatus()));
+  publisher_->Update(filter_params, false);
+  ASSERT_TRUE(publisher_->parameters().subscription_filter.has_value());
+  EXPECT_EQ(publisher_->parameters().subscription_filter->start(),
+            Location(3, 2));
+
+  // Set an AbsoluteRange filter and then extend end_group.
+  publisher_->parameters().subscription_filter =
+      SubscriptionFilter(Location(1, 0), 5);
+  MessageParameters extend_params;
+  extend_params.subscription_filter = SubscriptionFilter(Location(1, 0), 8);
+  EXPECT_CALL(*track_publisher_, largest_location())
+      .WillOnce(Return(Location(4, 2)));
+  MessageParameters expected_ok2;
+  expected_ok2.largest_object = Location(4, 2);
+  EXPECT_CALL(mock_bidi_stream_,
+              Writev(SerializedControlMessage(MoqtRequestOk{expected_ok2}), _))
+      .WillOnce(Return(absl::OkStatus()));
+  publisher_->Update(extend_params, false);
+  EXPECT_EQ(publisher_->parameters().subscription_filter->end_group(), 8);
+  EXPECT_EQ(publisher_->parameters().largest_object, Location(4, 2));
+}
+
+TEST_F(LivePublisherTest, UpdateRejectedByPublisher) {
+  MessageParameters new_params;
+  new_params.delivery_timeout = quic::QuicTimeDelta::FromSeconds(5);
+  EXPECT_CALL(mock_bidi_stream_, CanWrite()).WillRepeatedly(Return(true));
+
+  // When from_request_ok is false, sends REQUEST_ERROR.
+  EXPECT_CALL(*track_publisher_, UpdateObjectListener(publisher_.get(), _))
+      .WillOnce(Return(absl::InvalidArgumentError("invalid update")));
+  EXPECT_CALL(mock_bidi_stream_,
+              Writev(ControlMessageOfType(MoqtMessageType::kRequestError), _))
+      .WillOnce(Return(absl::OkStatus()));
+  publisher_->Update(new_params, /*from_request_ok=*/false);
+  EXPECT_EQ(publisher_->parameters().delivery_timeout,
+            quic::QuicTimeDelta::FromSeconds(1));
+
+  // When from_request_ok is true, resets the bidi stream.
+  EXPECT_CALL(*track_publisher_, UpdateObjectListener(publisher_.get(), _))
+      .WillOnce(Return(absl::InvalidArgumentError("invalid update")));
+  EXPECT_CALL(mock_bidi_stream_, ResetWithUserCode);
+  EXPECT_CALL(mock_bidi_stream_, SendStopSending);
+  publisher_->Update(new_params, /*from_request_ok=*/true);
 }
 
 TEST_F(LivePublisherTest, UpdatePriorityNoStreams) {
   MessageParameters new_params;
   new_params.subscriber_priority = 20;
-  publisher_->Update(new_params);
+  publisher_->Update(new_params, true);
   EXPECT_EQ(publisher_->parameters().subscriber_priority, 20);
 }
 
@@ -329,7 +464,11 @@
                                             std::optional<MoqtTrackPriority>(
                                                 {subscriber_priority(), 64}),
                                             MoqtTrackPriority{20, 64}));
-  publisher_->Update(new_params);
+  EXPECT_CALL(mock_bidi_stream_, CanWrite()).WillRepeatedly(Return(true));
+  EXPECT_CALL(mock_bidi_stream_,
+              Writev(SerializedControlMessage(MoqtRequestOk()), _))
+      .WillOnce(Return(absl::OkStatus()));
+  publisher_->Update(new_params, false);
 }
 
 TEST_F(LivePublisherTest, UpdatePriorityWithActiveStreams) {
@@ -339,13 +478,17 @@
   MessageParameters new_params;
   new_params.subscriber_priority = 20;
   EXPECT_CALL(mock_uni_stream_, SetPriority);
-  publisher_->Update(new_params);
+  EXPECT_CALL(mock_bidi_stream_, CanWrite()).WillRepeatedly(Return(true));
+  EXPECT_CALL(mock_bidi_stream_,
+              Writev(SerializedControlMessage(MoqtRequestOk()), _))
+      .WillOnce(Return(absl::OkStatus()));
+  publisher_->Update(new_params, false);
 }
 
 TEST_F(LivePublisherTest, OnNewObjectAvailableNotInWindow) {
   MessageParameters params;
   params.subscription_filter = SubscriptionFilter(Location(10, 0), 10);
-  publisher_->Update(params);
+  publisher_->Update(params, true);
   EXPECT_CALL(*track_publisher_, GetCachedObject).Times(0);
   publisher_->OnNewObjectAvailable(Location(5, 0), 0, 128);
 }
@@ -373,13 +516,17 @@
   // Not in window
   MessageParameters params;
   params.subscription_filter = SubscriptionFilter(Location(10, 0), 10);
-  publisher_->Update(params);
+  publisher_->Update(params, true);
   EXPECT_CALL(webtrans_, GetStreamById).Times(0);
   EXPECT_CALL(mock_uni_stream_, Writev).Times(0);
   publisher_->OnNewFinAvailable(Location(0, 0), 0);
 
   // In window but no stream
-  publisher_->Update(parameters_);
+  EXPECT_CALL(mock_bidi_stream_, CanWrite()).WillRepeatedly(Return(true));
+  EXPECT_CALL(mock_bidi_stream_,
+              Writev(SerializedControlMessage(MoqtRequestOk()), _))
+      .WillOnce(Return(absl::OkStatus()));
+  publisher_->Update(parameters_, false);
   EXPECT_CALL(webtrans_, GetStreamById).Times(0);
   EXPECT_CALL(mock_uni_stream_, Writev).Times(0);
   publisher_->OnNewFinAvailable(Location(10, 10), 0);
@@ -410,12 +557,16 @@
   // Not in window
   MessageParameters params;
   params.subscription_filter = SubscriptionFilter(Location(10, 0), 10);
-  publisher_->Update(params);
+  publisher_->Update(params, true);
   EXPECT_CALL(webtrans_, GetStreamById).Times(0);
   publisher_->OnSubgroupAbandoned(1, 0, 17);
 
   // In window but no stream
-  publisher_->Update(parameters_);
+  EXPECT_CALL(mock_bidi_stream_, CanWrite()).WillRepeatedly(Return(true));
+  EXPECT_CALL(mock_bidi_stream_,
+              Writev(SerializedControlMessage(MoqtRequestOk()), _))
+      .WillOnce(Return(absl::OkStatus()));
+  publisher_->Update(parameters_, false);
   EXPECT_CALL(webtrans_, GetStreamById).Times(0);
   publisher_->OnSubgroupAbandoned(1, 0, 17);
 }
@@ -424,12 +575,16 @@
   // Not in window
   MessageParameters params;
   params.subscription_filter = SubscriptionFilter(Location(10, 0), 10);
-  publisher_->Update(params);
+  publisher_->Update(params, true);
   EXPECT_CALL(webtrans_, GetStreamById).Times(0);
   publisher_->OnGroupAbandoned(1);
 
   // In window
-  publisher_->Update(parameters_);
+  EXPECT_CALL(mock_bidi_stream_, CanWrite()).WillRepeatedly(Return(true));
+  EXPECT_CALL(mock_bidi_stream_,
+              Writev(SerializedControlMessage(MoqtRequestOk()), _))
+      .WillOnce(Return(absl::OkStatus()));
+  publisher_->Update(parameters_, false);
   EXPECT_CALL(webtrans_, GetStreamById).Times(0);
   publisher_->OnGroupAbandoned(1);
   EXPECT_CALL(*track_publisher_, GetCachedObject).Times(0);
@@ -448,7 +603,7 @@
 TEST_F(LivePublisherTest, OnGroupAbandonedTooFarBehind) {
   // Set the delivery timeout to infinite so that TooFarBehind is possible.
   parameters_.delivery_timeout = quic::QuicTimeDelta::Infinite();
-  publisher_->Update(parameters_);
+  publisher_->Update(parameters_, true);
   CreateStream(Location(5, 0), 0, 128);
   struct MoqtPublishDone expected_publish_done = {
       /*request_id=*/kRequestId,
@@ -639,7 +794,11 @@
   // No entries in the stream map.
   EXPECT_CALL(webtrans_, GetStreamById).Times(0);
   parameters_.subscriber_priority = 20;
-  publisher_->Update(parameters_);
+  EXPECT_CALL(mock_bidi_stream_, CanWrite()).WillRepeatedly(Return(true));
+  EXPECT_CALL(mock_bidi_stream_,
+              Writev(SerializedControlMessage(MoqtRequestOk()), _))
+      .WillOnce(Return(absl::OkStatus()));
+  publisher_->Update(parameters_, false);
 }
 
 TEST_F(LivePublisherTest, OnObjectSentTwice) {
@@ -689,7 +848,11 @@
   // Track gets to Group 5.
   CreateStream(Location(5, 0), 0, 128);
   parameters_.subscription_filter = SubscriptionFilter(Location(0, 0), 4);
-  publisher_->Update(parameters_);
+  EXPECT_CALL(mock_bidi_stream_, CanWrite()).WillRepeatedly(Return(true));
+  EXPECT_CALL(mock_bidi_stream_,
+              Writev(SerializedControlMessage(MoqtRequestOk()), _))
+      .WillOnce(Return(absl::OkStatus()));
+  publisher_->Update(parameters_, false);
   EXPECT_CALL(*track_publisher_, GetCachedObject).Times(0);
   publisher_->OnNewObjectAvailable(Location(5, 1), 0, 128);
 }
@@ -717,7 +880,7 @@
 
 TEST_F(LivePublisherTest, OnSubgroupAbandonedOutsideWindow) {
   parameters_.subscription_filter = SubscriptionFilter(Location(20, 0));
-  publisher_->Update(parameters_);
+  publisher_->Update(parameters_, true);
   EXPECT_CALL(mock_uni_stream_, ResetWithUserCode).Times(0);
   publisher_->OnSubgroupAbandoned(1, 0, 1234);
 }

diff --git a/quiche/quic/moqt/moqt_publish_stream.cc b/quiche/quic/moqt/moqt_publish_stream.cc
index d85dea5..261d8f0 100644
--- a/quiche/quic/moqt/moqt_publish_stream.cc
+++ b/quiche/quic/moqt/moqt_publish_stream.cc

@@ -75,28 +75,39 @@
         absl::InvalidArgumentError("REQUEST_OK received with properties"));
     return absl::OkStatus();
   }
-  std::move(response_callback_)(message.parameters);
-  publisher_->Update(message.parameters);
-  return absl::OkStatus();
+  if (response_callback_ != nullptr) {
+    // PUBLISH_OK
+    publisher_->Update(message.parameters, /*from_request_ok=*/true);
+    MoqtResponseCallback callback = std::move(response_callback_);
+    response_callback_ = nullptr;
+    std::move(callback)(message.parameters);
+    return absl::OkStatus();
+  }
+  // REQUEST_UPDATE_OK
+  QUICHE_ASSIGN_OR_RETURN(MessageParameters parameters,
+                          request_update_queue().NextParameters());
+  // Apply the pending parameters to the subscription.
+  publisher_->Update(parameters, /*from_request_ok=*/true);
+  if (publisher_ != nullptr) {
+    // Apply any parameters from the REQUEST_OK.
+    publisher_->Update(message.parameters, /*from_request_ok=*/true);
+  }
+  return request_update_queue().OnControlMessage(message);
 }
 
 absl::Status MoqtPublishRequestStream::OnControlMessage(
     const MoqtRequestError& message) {
-  std::move(response_callback_)(message);
-  return absl::OkStatus();
+  if (response_callback_ != nullptr) {
+    std::move(response_callback_)(message);
+    return absl::OkStatus();
+  }
+  return request_update_queue().OnControlMessage(message);
 }
 
 absl::Status MoqtPublishRequestStream::OnControlMessage(
     const MoqtRequestUpdate& message) {
   QUICHE_RETURN_IF_ERROR(validate_request_id_(message.request_id));
-  MessageParameters in_parameters = message.parameters, out_parameters;
-  out_parameters.largest_object = publisher_->publisher().largest_location();
-  if (in_parameters.subscription_filter.has_value()) {
-    in_parameters.subscription_filter->OnLargestObject(
-        out_parameters.largest_object);
-  }
-  publisher_->Update(in_parameters);
-  CheckStatus(SendRequestOk(MessageParameters()));
+  publisher_->Update(message.parameters, /*from_request_ok=*/false);
   return absl::OkStatus();
 }
 

diff --git a/quiche/quic/moqt/moqt_publish_stream_test.cc b/quiche/quic/moqt/moqt_publish_stream_test.cc
index d339823..a9552c7 100644
--- a/quiche/quic/moqt/moqt_publish_stream_test.cc
+++ b/quiche/quic/moqt/moqt_publish_stream_test.cc

@@ -202,10 +202,9 @@
             request_update.parameters.subscriber_priority);
 
   // Verify filter was updated based on largest location (1, 2) -> (1, 3)
-  // AbsoluteStart
   ASSERT_TRUE(pub_params.subscription_filter.has_value());
   EXPECT_EQ(pub_params.subscription_filter->type(),
-            MoqtFilterType::kAbsoluteStart);
+            MoqtFilterType::kLargestObject);
   EXPECT_EQ(pub_params.subscription_filter->start(), Location(1, 3));
 }
 

diff --git a/quiche/quic/moqt/moqt_session.h b/quiche/quic/moqt/moqt_session.h
index 330b6f7..f823cbc 100644
--- a/quiche/quic/moqt/moqt_session.h
+++ b/quiche/quic/moqt/moqt_session.h

@@ -96,6 +96,10 @@
   bool Subscribe(const FullTrackName& name,
                  SubscribeVisitor* absl_nonnull visitor,
                  const MessageParameters& parameters) override;
+  // TODO(martinduke): There is an edge case where the update moves the filter
+  // window to the right, and the new LARGEST_OBJECT indicates that there's a
+  // gap of objects that has be fetched. This requires a standalone FETCH that
+  // probably should have its objects reported via the SubscribeVisitor.
   bool SubscribeUpdate(const FullTrackName& name,
                        const MessageParameters& parameters,
                        MoqtResponseCallback response_callback) override;

diff --git a/quiche/quic/moqt/moqt_session_test.cc b/quiche/quic/moqt/moqt_session_test.cc
index 24da4bf..c981d33 100644
--- a/quiche/quic/moqt/moqt_session_test.cc
+++ b/quiche/quic/moqt/moqt_session_test.cc

@@ -29,6 +29,7 @@
 #include "quiche/quic/moqt/moqt_framer.h"
 #include "quiche/quic/moqt/moqt_key_value_pair.h"
 #include "quiche/quic/moqt/moqt_known_track_publisher.h"
+#include "quiche/quic/moqt/moqt_live_publisher.h"
 #include "quiche/quic/moqt/moqt_messages.h"
 #include "quiche/quic/moqt/moqt_names.h"
 #include "quiche/quic/moqt/moqt_object.h"
@@ -1903,6 +1904,46 @@
   fetch_wrapper->ReceiveMessage(fetch);
 }
 
+TEST_F(MoqtSessionTest, IncomingJoiningFetchAfterRequestUpdate) {
+  MoqtSubscribe subscribe = DefaultSubscribe();
+  subscribe.parameters.set_forward(false);
+  bidi_wrapper_ = std::make_unique<MoqtBidiStreamTestWrapper>(
+      ResponseStream(kSubscribeByte));
+  MockTrackPublisher* track = CreateTrackPublisher();
+  SetLargestId(track, Location(2, 10));
+  MoqtObjectListener* listener =
+      ReceiveSubscribeSynchronousOk(track, subscribe, bidi_wrapper_.get());
+  LivePublisher* subscription = absl::down_cast<LivePublisher*>(listener);
+  ASSERT_NE(subscription, nullptr);
+
+  SetLargestId(track, Location(4, 10));
+  MessageParameters update_parameters;
+  update_parameters.set_forward(true);
+  update_parameters.subscription_filter.emplace(MoqtFilterType::kLargestObject);
+  MessageParameters expected_ok_parameters;
+  expected_ok_parameters.largest_object = Location(4, 10);
+  EXPECT_CALL(
+      mock_bidi_stream_,
+      Writev(SerializedControlMessage(MoqtRequestOk{expected_ok_parameters}),
+             _));
+  bidi_wrapper_->ReceiveMessage(
+      MoqtRequestUpdate{3, subscribe.request_id, update_parameters});
+  ASSERT_TRUE(subscription->parameters().subscription_filter.has_value());
+  EXPECT_EQ(subscription->parameters().subscription_filter->start(),
+            Location(4, 11));
+
+  webtransport::test::MockStream fetch_stream;
+  std::unique_ptr<MoqtBidiStreamTestWrapper> fetch_wrapper =
+      std::make_unique<MoqtBidiStreamTestWrapper>(
+          ResponseStream(kFetchByte, &fetch_stream));
+  MoqtFetch fetch = DefaultFetch();
+  fetch.request_id = 5;
+  fetch.fetch = JoiningFetchRelative(subscribe.request_id, 2);
+  EXPECT_CALL(*track, StandaloneFetch(Location(2, 0), Location(4, 10), _, _))
+      .WillOnce(Return(std::make_unique<MockFetchTask>()));
+  fetch_wrapper->ReceiveMessage(fetch);
+}
+
 TEST_F(MoqtSessionTest, SendJoiningFetch) {
   webtransport::test::MockStream subscribe_stream;
   std::unique_ptr<MoqtBidiStreamTestWrapper> subscribe_wrapper;

diff --git a/quiche/quic/moqt/moqt_subscribe_stream.cc b/quiche/quic/moqt/moqt_subscribe_stream.cc
index cddee8c..ad69cd5 100644
--- a/quiche/quic/moqt/moqt_subscribe_stream.cc
+++ b/quiche/quic/moqt/moqt_subscribe_stream.cc

@@ -245,8 +245,13 @@
     return SendRequestError(RequestErrorCode::kInternalError, std::nullopt,
                             "no subscription");
   }
-  subscription_->Update(message.parameters);
-  return SendRequestOk(MessageParameters());
+  if (!subscription_->established() || IncomingUpdatesQueued()) {
+    // There are requests waiting in the queue. Buffer this one.
+    QueueIncomingUpdate(message.parameters);
+    return absl::OkStatus();
+  }
+  subscription_->Update(message.parameters, /*from_request_ok=*/false);
+  return absl::OkStatus();
 }
 
 void MoqtSubscribeResponseStream::Detach() {

diff --git a/quiche/quic/moqt/moqt_subscribe_stream_test.cc b/quiche/quic/moqt/moqt_subscribe_stream_test.cc
index ceb6200..499bc47 100644
--- a/quiche/quic/moqt/moqt_subscribe_stream_test.cc
+++ b/quiche/quic/moqt/moqt_subscribe_stream_test.cc

@@ -23,6 +23,7 @@
 #include "quiche/quic/moqt/moqt_names.h"
 #include "quiche/quic/moqt/moqt_object_subscriber.h"
 #include "quiche/quic/moqt/moqt_parser.h"
+#include "quiche/quic/moqt/moqt_publisher.h"
 #include "quiche/quic/moqt/moqt_session_callbacks.h"
 #include "quiche/quic/moqt/moqt_session_interface.h"
 #include "quiche/quic/moqt/moqt_trace_recorder.h"
@@ -351,6 +352,52 @@
   EXPECT_CALL(mock_remove_callback_, Call(_));
 }
 
+TEST_F(MoqtSubscribeResponseStreamTest,
+       ReceiveRequestUpdateBeforeSubscribeAccepted) {
+  auto mock_track_publisher = std::make_shared<MockTrackPublisher>(kTrackName);
+  EXPECT_CALL(visitor_, session).WillRepeatedly(Return(&webtrans_));
+  EXPECT_CALL(visitor_, GetTrackPublisher(kTrackName))
+      .WillOnce(Return(mock_track_publisher));
+  EXPECT_CALL(mock_add_callback_, Call(testing::NotNull()))
+      .WillOnce(Return(true));
+  MoqtObjectListener* captured_listener = nullptr;
+  EXPECT_CALL(*mock_track_publisher, AddObjectListener)
+      .WillOnce([&](MoqtObjectListener* listener, const MessageParameters&) {
+        captured_listener = listener;
+      });
+
+  MoqtSubscribe subscribe;
+  subscribe.request_id = kRequestId;
+  subscribe.full_track_name = kTrackName;
+  QUICHE_EXPECT_OK(stream_->OnControlMessage(subscribe));
+  ASSERT_NE(captured_listener, nullptr);
+
+  // REQUEST_UPDATE arrives before OnSubscribeAccepted(); it is queued and
+  // does not write REQUEST_OK yet.
+  EXPECT_CALL(mock_stream_, Writev).Times(0);
+  MoqtRequestUpdate update;
+  update.request_id = kRequestId + 2;
+  update.existing_request_id = kRequestId;
+  update.parameters.subscriber_priority = 10;
+  QUICHE_EXPECT_OK(stream_->OnControlMessage(update));
+
+  // Accepting the subscription sends SUBSCRIBE_OK followed by REQUEST_OK.
+  {
+    testing::InSequence seq;
+    EXPECT_CALL(mock_stream_,
+                Writev(ControlMessageOfType(MoqtMessageType::kSubscribeOk), _))
+        .WillOnce(Return(absl::OkStatus()));
+    EXPECT_CALL(mock_stream_,
+                Writev(ControlMessageOfType(MoqtMessageType::kRequestOk), _))
+        .WillOnce(Return(absl::OkStatus()));
+  }
+  captured_listener->OnSubscribeAccepted();
+
+  // Test cleanup.
+  EXPECT_CALL(*mock_track_publisher, RemoveObjectListener(captured_listener));
+  EXPECT_CALL(mock_remove_callback_, Call);
+}
+
 TEST_F(MoqtSubscribeResponseStreamTest, ReceiveObjectAck) {
   EXPECT_CALL(visitor_, session).WillRepeatedly(Return(&webtrans_));
   EXPECT_CALL(visitor_, GetTrackPublisher(kTrackName))