diff --git a/quiche/quic/moqt/moqt_failed_fetch.h b/quiche/quic/moqt/moqt_failed_fetch.h
index a1e93bf..40e8d67 100644
--- a/quiche/quic/moqt/moqt_failed_fetch.h
+++ b/quiche/quic/moqt/moqt_failed_fetch.h
@@ -8,6 +8,7 @@
 #include <utility>
 
 #include "absl/status/status.h"
+#include "quiche/quic/moqt/moqt_messages.h"
 #include "quiche/quic/moqt/moqt_publisher.h"
 
 namespace moqt {
@@ -23,6 +24,7 @@
   absl::Status GetStatus() override { return status_; }
   void SetObjectAvailableCallback(
       ObjectsAvailableCallback /*callback*/) override {}
+  FullSequence GetLargestId() const override { return FullSequence(); }
 
  private:
   absl::Status status_;
diff --git a/quiche/quic/moqt/moqt_outgoing_queue.h b/quiche/quic/moqt/moqt_outgoing_queue.h
index 2bc46ee..83d7773 100644
--- a/quiche/quic/moqt/moqt_outgoing_queue.h
+++ b/quiche/quic/moqt/moqt_outgoing_queue.h
@@ -102,6 +102,7 @@
 
     GetNextObjectResult GetNextObject(PublishedObject&) override;
     absl::Status GetStatus() override { return status_; }
+    FullSequence GetLargestId() const override { return objects_.back(); }
 
     void SetObjectAvailableCallback(
         ObjectsAvailableCallback /*callback*/) override {
diff --git a/quiche/quic/moqt/moqt_publisher.h b/quiche/quic/moqt/moqt_publisher.h
index a0897f4..010db7f 100644
--- a/quiche/quic/moqt/moqt_publisher.h
+++ b/quiche/quic/moqt/moqt_publisher.h
@@ -84,8 +84,9 @@
   // Returns the error if fetch has completely failed, and OK otherwise.
   virtual absl::Status GetStatus() = 0;
 
-  // TODO: expose the largest sequence and the end of track bit returned in
-  // the FETCH_OK.
+  // Returns the highest sequence number that will be delivered by the fetch.
+  // It is the minimum of the end of the fetch range and the live edge.
+  virtual FullSequence GetLargestId() const = 0;
 };
 
 // MoqtTrackPublisher is an application-side API for an MoQT publisher
diff --git a/quiche/quic/moqt/moqt_session.cc b/quiche/quic/moqt/moqt_session.cc
index 0f8fad7..ec7a867 100644
--- a/quiche/quic/moqt/moqt_session.cc
+++ b/quiche/quic/moqt/moqt_session.cc
@@ -339,6 +339,46 @@
   return Subscribe(message, visitor);
 }
 
+void MoqtSession::PublishedFetch::FetchStreamVisitor::OnCanWrite() {
+  std::shared_ptr<PublishedFetch> fetch = fetch_.lock();
+  if (fetch == nullptr) {
+    return;
+  }
+  PublishedObject object;
+  while (stream_->CanWrite()) {
+    MoqtFetchTask::GetNextObjectResult result =
+        fetch->fetch_task()->GetNextObject(object);
+    switch (result) {
+      case MoqtFetchTask::GetNextObjectResult::kSuccess:
+        // Skip ObjectDoesNotExist in FETCH.
+        if (object.status == MoqtObjectStatus::kObjectDoesNotExist) {
+          continue;
+        }
+        if (fetch->session_->WriteObjectToStream(
+                stream_, fetch->fetch_id_, object,
+                MoqtDataStreamType::kStreamHeaderFetch, !stream_header_written_,
+                /*fin=*/false)) {
+          stream_header_written_ = true;
+        }
+        break;
+      case MoqtFetchTask::GetNextObjectResult::kPending:
+        return;
+      case MoqtFetchTask::GetNextObjectResult::kEof:
+        // TODO(martinduke): Either prefetch the next object, or alter the API
+        // so that we're not sending FIN in a separate frame.
+        if (!quiche::SendFinOnStream(*stream_).ok()) {
+          QUIC_DVLOG(1) << "Sending FIN onStream " << stream_->GetStreamId()
+                        << " failed";
+        }
+        return;
+      case MoqtFetchTask::GetNextObjectResult::kError:
+        stream_->ResetWithUserCode(static_cast<webtransport::StreamErrorCode>(
+            fetch->fetch_task()->GetStatus().code()));
+        return;
+    }
+  }
+}
+
 bool MoqtSession::SubscribeIsDone(uint64_t subscribe_id, SubscribeDoneCode code,
                                   absl::string_view reason_phrase) {
   auto it = published_subscriptions_.find(subscribe_id);
@@ -446,13 +486,36 @@
   return new_stream;
 }
 
+bool MoqtSession::OpenDataStream(std::shared_ptr<PublishedFetch> fetch) {
+  webtransport::Stream* new_stream =
+      session_->OpenOutgoingUnidirectionalStream();
+  if (new_stream == nullptr) {
+    QUICHE_BUG(MoqtSession_OpenDataStream_blocked)
+        << "OpenDataStream called when creation of new streams is blocked.";
+    return false;
+  }
+  fetch->SetStreamId(new_stream->GetStreamId());
+  new_stream->SetVisitor(
+      std::make_unique<PublishedFetch::FetchStreamVisitor>(fetch, new_stream));
+  if (new_stream->CanWrite()) {
+    new_stream->visitor()->OnCanWrite();
+  }
+  return true;
+}
+
 void MoqtSession::OnCanCreateNewOutgoingUnidirectionalStream() {
   while (!subscribes_with_queued_outgoing_data_streams_.empty() &&
          session_->CanOpenNextOutgoingUnidirectionalStream()) {
     auto next = subscribes_with_queued_outgoing_data_streams_.rbegin();
     auto subscription = published_subscriptions_.find(next->subscription_id);
     if (subscription == published_subscriptions_.end()) {
-      // Subscription no longer exists; delete the entry.
+      auto fetch = incoming_fetches_.find(next->subscription_id);
+      // Create the stream if the fetch still exists.
+      if (fetch != incoming_fetches_.end() && !OpenDataStream(fetch->second)) {
+        return;  // A QUIC_BUG has fired because this shouldn't happen.
+      }
+      // FETCH needs only one stream, and can be deleted from the queue. Or,
+      // there is no subscribe and no fetch; the entry in the queue is invalid.
       subscribes_with_queued_outgoing_data_streams_.erase((++next).base());
       continue;
     }
@@ -535,6 +598,28 @@
   return std::make_pair(track.full_track_name(), track.visitor());
 }
 
+bool MoqtSession::ValidateSubscribeId(uint64_t subscribe_id) {
+  if (peer_role_ == MoqtRole::kPublisher) {
+    QUIC_DLOG(INFO) << ENDPOINT << "Publisher peer sent SUBSCRIBE";
+    Error(MoqtError::kProtocolViolation, "Received SUBSCRIBE from publisher");
+    return false;
+  }
+  if (subscribe_id > local_max_subscribe_id_) {
+    QUIC_DLOG(INFO) << ENDPOINT << "Received SUBSCRIBE with too large ID";
+    Error(MoqtError::kTooManySubscribes,
+          "Received SUBSCRIBE with too large ID");
+    return false;
+  }
+  if (subscribe_id < next_incoming_subscribe_id_) {
+    QUIC_DLOG(INFO) << ENDPOINT << "Subscribe ID not monotonically increasing";
+    Error(MoqtError::kProtocolViolation,
+          "Subscribe ID not monotonically increasing");
+    return false;
+  }
+  next_incoming_subscribe_id_ = subscribe_id + 1;
+  return true;
+}
+
 template <class Parser>
 static void ForwardStreamDataToParser(webtransport::Stream& stream,
                                       Parser& parser) {
@@ -647,18 +732,19 @@
       session_->framer_.SerializeSubscribeError(subscribe_error));
 }
 
+void MoqtSession::ControlStream::SendFetchError(
+    uint64_t subscribe_id, SubscribeErrorCode error_code,
+    absl::string_view reason_phrase) {
+  MoqtFetchError fetch_error;
+  fetch_error.subscribe_id = subscribe_id;
+  fetch_error.error_code = error_code;
+  fetch_error.reason_phrase = reason_phrase;
+  SendOrBufferMessage(session_->framer_.SerializeFetchError(fetch_error));
+}
+
 void MoqtSession::ControlStream::OnSubscribeMessage(
     const MoqtSubscribe& message) {
-  if (session_->peer_role_ == MoqtRole::kPublisher) {
-    QUIC_DLOG(INFO) << ENDPOINT << "Publisher peer sent SUBSCRIBE";
-    session_->Error(MoqtError::kProtocolViolation,
-                    "Received SUBSCRIBE from publisher");
-    return;
-  }
-  if (message.subscribe_id > session_->local_max_subscribe_id_) {
-    QUIC_DLOG(INFO) << ENDPOINT << "Received SUBSCRIBE with too large ID";
-    session_->Error(MoqtError::kTooManySubscribes,
-                    "Received SUBSCRIBE with too large ID");
+  if (!session_->ValidateSubscribeId(message.subscribe_id)) {
     return;
   }
   QUIC_DLOG(INFO) << ENDPOINT << "Received a SUBSCRIBE for "
@@ -883,6 +969,66 @@
   session_->peer_max_subscribe_id_ = message.max_subscribe_id;
 }
 
+void MoqtSession::ControlStream::OnFetchMessage(const MoqtFetch& message) {
+  if (!session_->ValidateSubscribeId(message.subscribe_id)) {
+    return;
+  }
+  QUIC_DLOG(INFO) << ENDPOINT << "Received a FETCH for "
+                  << message.full_track_name;
+
+  const FullTrackName& track_name = message.full_track_name;
+  absl::StatusOr<std::shared_ptr<MoqtTrackPublisher>> track_publisher =
+      session_->publisher_->GetTrack(track_name);
+  if (!track_publisher.ok()) {
+    QUIC_DLOG(INFO) << ENDPOINT << "FETCH for " << track_name
+                    << " rejected by the application: "
+                    << track_publisher.status();
+    SendFetchError(message.subscribe_id, SubscribeErrorCode::kTrackDoesNotExist,
+                   track_publisher.status().message());
+    return;
+  }
+  std::unique_ptr<MoqtFetchTask> fetch =
+      (*track_publisher)
+          ->Fetch(message.start_object, message.end_group, message.end_object,
+                  message.group_order.value_or(
+                      (*track_publisher)->GetDeliveryOrder()));
+  if (!fetch->GetStatus().ok()) {
+    QUIC_DLOG(INFO) << ENDPOINT << "FETCH for " << track_name
+                    << " could not initialize the task";
+    SendFetchError(message.subscribe_id, SubscribeErrorCode::kInvalidRange,
+                   fetch->GetStatus().message());
+    return;
+  }
+  auto published_fetch = std::make_unique<PublishedFetch>(
+      message.subscribe_id, session_, std::move(fetch));
+  auto result = session_->incoming_fetches_.emplace(message.subscribe_id,
+                                                    std::move(published_fetch));
+  if (!result.second) {  // Emplace failed.
+    QUIC_DLOG(INFO) << ENDPOINT << "FETCH for " << track_name
+                    << " could not be added to the session";
+    SendFetchError(message.subscribe_id, SubscribeErrorCode::kInternalError,
+                   "Could not initialize FETCH state");
+    return;
+  }
+  MoqtFetchOk fetch_ok;
+  fetch_ok.subscribe_id = message.subscribe_id;
+  fetch_ok.group_order =
+      message.group_order.value_or((*track_publisher)->GetDeliveryOrder());
+  fetch_ok.largest_id = result.first->second->fetch_task()->GetLargestId();
+  SendOrBufferMessage(session_->framer_.SerializeFetchOk(fetch_ok));
+  if (!session_->session()->CanOpenNextOutgoingUnidirectionalStream() ||
+      !session_->OpenDataStream(result.first->second)) {
+    // Put the FETCH in the queue for a new stream.
+    session_->UpdateQueuedSendOrder(
+        message.subscribe_id, std::nullopt,
+        SendOrderForStream(message.subscriber_priority,
+                           (*track_publisher)->GetPublisherPriority(),
+                           /*group_id=*/0,
+                           message.group_order.value_or(
+                               (*track_publisher)->GetDeliveryOrder())));
+  }
+}
+
 void MoqtSession::ControlStream::OnParsingError(MoqtError error_code,
                                                 absl::string_view reason) {
   session_->Error(error_code, absl::StrCat("Parse error: ", reason));
@@ -1290,7 +1436,38 @@
           << "Writing FIN failed despite CanWrite() being true.";
       return;
     }
-    SendNextObject(subscription, *std::move(object));
+    QUICHE_DCHECK(next_object_ <= object->sequence);
+    MoqtTrackPublisher& publisher = subscription.publisher();
+    QUICHE_DCHECK(DoesTrackStatusImplyHavingData(*publisher.GetTrackStatus()));
+    MoqtForwardingPreference forwarding_preference =
+        publisher.GetForwardingPreference();
+    UpdateSendOrder(subscription);
+    switch (forwarding_preference) {
+      case MoqtForwardingPreference::kTrack:
+        if (object->status == MoqtObjectStatus::kEndOfGroup ||
+            object->status == MoqtObjectStatus::kGroupDoesNotExist) {
+          ++next_object_.group;
+          next_object_.object = 0;
+        } else {
+          next_object_.object = object->sequence.object + 1;
+        }
+        break;
+
+      case MoqtForwardingPreference::kSubgroup:
+        next_object_.object = object->sequence.object + 1;
+        break;
+
+      case MoqtForwardingPreference::kDatagram:
+        QUICHE_NOTREACHED();
+        break;
+    }
+    if (session_->WriteObjectToStream(
+            stream_, subscription.track_alias(), *object,
+            GetMessageTypeForForwardingPreference(forwarding_preference),
+            !stream_header_written_, object->fin_after_this)) {
+      stream_header_written_ = true;
+      subscription.OnObjectSent(object->sequence);
+    }
   }
 }
 
@@ -1305,78 +1482,41 @@
       << "Writing pure FIN failed.";
 }
 
-void MoqtSession::OutgoingDataStream::SendNextObject(
-    PublishedSubscription& subscription, PublishedObject object) {
-  QUICHE_DCHECK(next_object_ <= object.sequence);
-  QUICHE_DCHECK(stream_->CanWrite());
-
-  MoqtTrackPublisher& publisher = subscription.publisher();
-  QUICHE_DCHECK(DoesTrackStatusImplyHavingData(*publisher.GetTrackStatus()));
-  MoqtForwardingPreference forwarding_preference =
-      publisher.GetForwardingPreference();
-
-  UpdateSendOrder(subscription);
-
+bool MoqtSession::WriteObjectToStream(webtransport::Stream* stream, uint64_t id,
+                                      const PublishedObject& object,
+                                      MoqtDataStreamType type,
+                                      bool is_first_on_stream, bool fin) {
+  QUICHE_DCHECK(stream->CanWrite());
   MoqtObject header;
-  header.track_alias = subscription.track_alias();
+  header.track_alias = id;
   header.group_id = object.sequence.group;
+  header.subgroup_id = object.sequence.subgroup;
   header.object_id = object.sequence.object;
-  header.publisher_priority = publisher.GetPublisherPriority();
+  header.publisher_priority = object.publisher_priority;
   header.object_status = object.status;
-  header.forwarding_preference = forwarding_preference;
-  // TODO(martinduke): send values other than 0.
-  header.subgroup_id =
-      (forwarding_preference == MoqtForwardingPreference::kSubgroup)
-          ? 0
-          : std::optional<uint64_t>();
   header.payload_length = object.payload.length();
 
   quiche::QuicheBuffer serialized_header =
-      session_->framer_.SerializeObjectHeader(
-          header, GetMessageTypeForForwardingPreference(forwarding_preference),
-          !stream_header_written_);
-  switch (forwarding_preference) {
-    case MoqtForwardingPreference::kTrack:
-      if (object.status == MoqtObjectStatus::kEndOfGroup ||
-          object.status == MoqtObjectStatus::kGroupDoesNotExist) {
-        ++next_object_.group;
-        next_object_.object = 0;
-      } else {
-        next_object_.object = header.object_id + 1;
-      }
-      break;
-
-    case MoqtForwardingPreference::kSubgroup:
-      next_object_.object = header.object_id + 1;
-      break;
-
-    case MoqtForwardingPreference::kDatagram:
-      QUICHE_NOTREACHED();
-      break;
-  }
-
+      framer_.SerializeObjectHeader(header, type, is_first_on_stream);
   // TODO(vasilvv): add a version of WebTransport write API that accepts
   // memslices so that we can avoid a copy here.
   std::array<absl::string_view, 2> write_vector = {
       serialized_header.AsStringView(), object.payload.AsStringView()};
   quiche::StreamWriteOptions options;
-  options.set_send_fin(object.fin_after_this);
-  absl::Status write_status = stream_->Writev(write_vector, options);
+  options.set_send_fin(fin);
+  absl::Status write_status = stream->Writev(write_vector, options);
   if (!write_status.ok()) {
-    QUICHE_BUG(MoqtSession_SendNextObject_write_failed)
+    QUICHE_BUG(MoqtSession_WriteObjectToStream_write_failed)
         << "Writing into MoQT stream failed despite CanWrite() being true "
            "before; status: "
         << write_status;
-    session_->Error(MoqtError::kInternalError, "Data stream write error");
-    return;
+    Error(MoqtError::kInternalError, "Data stream write error");
+    return false;
   }
 
-  QUIC_DVLOG(1) << "Stream " << stream_->GetStreamId() << " successfully wrote "
-                << object.sequence << ", fin = " << object.fin_after_this
-                << ", next: " << next_object_;
-
-  stream_header_written_ = true;
-  subscription.OnObjectSent(object.sequence);
+  QUIC_DVLOG(1) << "Stream " << stream->GetStreamId() << " successfully wrote "
+                << object.sequence << ", fin = " << fin;
+  return true;
 }
 
 void MoqtSession::PublishedSubscription::SendDatagram(FullSequence sequence) {
diff --git a/quiche/quic/moqt/moqt_session.h b/quiche/quic/moqt/moqt_session.h
index 04972b1..ffae8d7 100644
--- a/quiche/quic/moqt/moqt_session.h
+++ b/quiche/quic/moqt/moqt_session.h
@@ -172,6 +172,8 @@
  private:
   friend class test::MoqtSessionPeer;
 
+  struct Empty {};
+
   class QUICHE_EXPORT ControlStream : public webtransport::StreamVisitor,
                                       public MoqtControlParserVisitor {
    public:
@@ -213,7 +215,7 @@
     void OnUnsubscribeAnnouncesMessage(
         const MoqtUnsubscribeAnnounces& message) override {}
     void OnMaxSubscribeIdMessage(const MoqtMaxSubscribeId& message) override;
-    void OnFetchMessage(const MoqtFetch& message) override {}
+    void OnFetchMessage(const MoqtFetch& message) override;
     void OnFetchCancelMessage(const MoqtFetchCancel& message) override {}
     void OnFetchOkMessage(const MoqtFetchOk& message) override {}
     void OnFetchErrorMessage(const MoqtFetchError& message) override {}
@@ -244,6 +246,8 @@
                             SubscribeErrorCode error_code,
                             absl::string_view reason_phrase,
                             uint64_t track_alias);
+    void SendFetchError(uint64_t subscribe_id, SubscribeErrorCode error_code,
+                        absl::string_view reason_phrase);
 
     MoqtSession* session_;
     webtransport::Stream* stream_;
@@ -418,11 +422,6 @@
     // the stream and returns nullptr.
     PublishedSubscription* GetSubscriptionIfValid();
 
-    // Actually sends an object on the stream; the object MUST have object ID
-    // of at least `next_object_`.
-    void SendNextObject(PublishedSubscription& subscription,
-                        PublishedObject object);
-
     MoqtSession* session_;
     webtransport::Stream* stream_;
     uint64_t subscription_id_;
@@ -436,6 +435,54 @@
     std::weak_ptr<void> session_liveness_;
   };
 
+  class QUICHE_EXPORT PublishedFetch {
+   public:
+    PublishedFetch(uint64_t fetch_id, MoqtSession* session,
+                   std::unique_ptr<MoqtFetchTask> fetch)
+        : session_(session), fetch_(std::move(fetch)), fetch_id_(fetch_id) {}
+
+    class FetchStreamVisitor : public webtransport::StreamVisitor {
+     public:
+      FetchStreamVisitor(std::shared_ptr<PublishedFetch> fetch,
+                         webtransport::Stream* stream)
+          : fetch_(fetch), stream_(stream) {
+        fetch->fetch_task()->SetObjectAvailableCallback(
+            [this]() { this->OnCanWrite(); });
+      }
+      ~FetchStreamVisitor() {
+        std::shared_ptr<PublishedFetch> fetch = fetch_.lock();
+        if (fetch != nullptr) {
+          fetch->session()->incoming_fetches_.erase(fetch->fetch_id_);
+        }
+      }
+      // webtransport::StreamVisitor implementation.
+      void OnCanRead() override {}  // Write-only stream.
+      void OnCanWrite() override;
+      void OnResetStreamReceived(webtransport::StreamErrorCode error) override {
+      }  // Write-only stream
+      void OnStopSendingReceived(webtransport::StreamErrorCode error) override {
+      }
+      void OnWriteSideInDataRecvdState() override {}
+
+     private:
+      std::weak_ptr<PublishedFetch> fetch_;
+      webtransport::Stream* stream_;
+      bool stream_header_written_ = false;
+    };
+
+    MoqtFetchTask* fetch_task() { return fetch_.get(); }
+    MoqtSession* session() { return session_; }
+    uint64_t fetch_id() const { return fetch_id_; }
+    void SetStreamId(webtransport::StreamId id) { stream_id_ = id; }
+
+   private:
+    MoqtSession* session_;
+    std::unique_ptr<MoqtFetchTask> fetch_;
+    uint64_t fetch_id_;
+    // Store the stream ID in case a FETCH_CANCEL requires a reset.
+    std::optional<webtransport::StreamId> stream_id_;
+  };
+
   // Private members of MoqtSession.
 
   // Returns true if SUBSCRIBE_DONE was sent.
@@ -458,12 +505,26 @@
   // blocked.
   webtransport::Stream* OpenDataStream(PublishedSubscription& subscription,
                                        FullSequence first_object);
+  // Returns false if creation failed.
+  [[nodiscard]] bool OpenDataStream(std::shared_ptr<PublishedFetch> fetch);
 
   // Get FullTrackName and visitor for a subscribe_id and track_alias. Returns
   // an empty FullTrackName tuple and nullptr if not present.
   std::pair<FullTrackName, RemoteTrack::Visitor*> TrackPropertiesFromAlias(
       const MoqtObject& message);
 
+  // Checks that a subscribe ID from a SUBSCRIBE or FETCH is valid, and throws
+  // a session error if is not.
+  bool ValidateSubscribeId(uint64_t subscribe_id);
+
+  // Actually sends an object on |stream| with track alias or fetch ID |id|
+  // and metadata in |object|. Not for use with datagrams. Returns |true| if
+  // the write was successful.
+  bool WriteObjectToStream(webtransport::Stream* stream, uint64_t id,
+                           const PublishedObject& object,
+                           MoqtDataStreamType type, bool is_first_on_stream,
+                           bool fin);
+
   // Sends an OBJECT_ACK message for a specific subscribe ID.
   void SendObjectAck(uint64_t subscribe_id, uint64_t group_id,
                      uint64_t object_id,
@@ -515,6 +576,12 @@
   absl::flat_hash_set<uint64_t> used_track_aliases_;
   uint64_t next_local_track_alias_ = 0;
 
+  // Incoming FETCHes, indexed by fetch ID. There will be other pointers to
+  // PublishedFetch, so storing a shared_ptr in the map provides pointer
+  // stability for the value.
+  absl::flat_hash_map<uint64_t, std::shared_ptr<PublishedFetch>>
+      incoming_fetches_;
+
   // Indexed by subscribe_id.
   struct ActiveSubscribe {
     MoqtSubscribe message;
@@ -547,10 +614,12 @@
   uint64_t peer_max_subscribe_id_ = 0;
   // The maximum subscribe ID sent to the peer.
   uint64_t local_max_subscribe_id_ = 0;
+  // The minimum subscribe ID the peer can use that is monotonically increasing.
+  uint64_t next_incoming_subscribe_id_ = 0;
 
   // Must be last.  Token used to make sure that the streams do not call into
   // the session when the session has already been destroyed.
-  struct Empty {};
+
   std::shared_ptr<Empty> liveness_token_;
 };
 
diff --git a/quiche/quic/moqt/moqt_session_test.cc b/quiche/quic/moqt/moqt_session_test.cc
index bc31674..5e40775 100644
--- a/quiche/quic/moqt/moqt_session_test.cc
+++ b/quiche/quic/moqt/moqt_session_test.cc
@@ -42,6 +42,7 @@
 using ::quic::test::MemSliceFromString;
 using ::testing::_;
 using ::testing::AnyNumber;
+using ::testing::Invoke;
 using ::testing::Return;
 using ::testing::StrictMock;
 
@@ -75,6 +76,20 @@
 
 }  // namespace
 
+class MockFetchTask : public MoqtFetchTask {
+ public:
+  MOCK_METHOD(MoqtFetchTask::GetNextObjectResult, GetNextObject,
+              (PublishedObject & output), (override));
+  MOCK_METHOD(absl::Status, GetStatus, (), (override));
+  MOCK_METHOD(FullSequence, GetLargestId, (), (const, override));
+
+  void SetObjectAvailableCallback(ObjectsAvailableCallback callback) override {
+    callback_ = std::move(callback);
+  }
+
+  ObjectsAvailableCallback callback_;
+};
+
 class MoqtSessionPeer {
  public:
   static std::unique_ptr<MoqtControlParserVisitor> CreateControlStream(
@@ -169,6 +184,35 @@
   static void set_peer_max_subscribe_id(MoqtSession* session, uint64_t id) {
     session->peer_max_subscribe_id_ = id;
   }
+
+  static MockFetchTask* AddFetch(MoqtSession* session, uint64_t fetch_id) {
+    auto fetch_task = std::make_unique<MockFetchTask>();
+    MockFetchTask* return_ptr = fetch_task.get();
+    auto published_fetch = std::make_unique<MoqtSession::PublishedFetch>(
+        fetch_id, session, std::move(fetch_task));
+    session->incoming_fetches_.emplace(fetch_id, std::move(published_fetch));
+    // Add the fetch to the pending stream queue.
+    session->UpdateQueuedSendOrder(fetch_id, std::nullopt, 0);
+    return return_ptr;
+  }
+
+  static MoqtSession::PublishedFetch* GetFetch(MoqtSession* session,
+                                               uint64_t fetch_id) {
+    auto it = session->incoming_fetches_.find(fetch_id);
+    if (it == session->incoming_fetches_.end()) {
+      return nullptr;
+    }
+    return it->second.get();
+  }
+
+  static void ValidateSubscribeId(MoqtSession* session, uint64_t id) {
+    session->ValidateSubscribeId(id);
+  }
+
+  static FullSequence LargestSentForSubscription(MoqtSession* session,
+                                                 uint64_t subscribe_id) {
+    return *session->published_subscriptions_[subscribe_id]->largest_sent();
+  }
 };
 
 class MoqtSessionTest : public quic::test::QuicTest {
@@ -356,6 +400,7 @@
         EXPECT_EQ(*ExtractMessageType(data[0]), MoqtMessageType::kSubscribeOk);
         return absl::OkStatus();
       });
+  request.subscribe_id = 2;
   stream_input->OnSubscribeMessage(request);
   EXPECT_TRUE(correct_message);
 }
@@ -626,6 +671,47 @@
   stream_input->OnSubscribeMessage(request);
 }
 
+TEST_F(MoqtSessionTest, SubscribeIdNotIncreasing) {
+  // Peer subscribes to (0, 0)
+  MoqtSubscribe request = {
+      /*subscribe_id=*/1,
+      /*track_alias=*/2,
+      /*full_track_name=*/FullTrackName({"foo", "bar"}),
+      /*subscriber_priority=*/0x80,
+      /*group_order=*/std::nullopt,
+      /*start_group=*/0,
+      /*start_object=*/0,
+      /*end_group=*/std::nullopt,
+      /*end_object=*/std::nullopt,
+      /*parameters=*/MoqtSubscribeParameters(),
+  };
+  webtransport::test::MockStream mock_stream;
+  std::unique_ptr<MoqtControlParserVisitor> stream_input =
+      MoqtSessionPeer::CreateControlStream(&session_, &mock_stream);
+  // Request for track returns SUBSCRIBE_ERROR.
+  bool correct_message = false;
+  EXPECT_CALL(mock_stream, Writev(_, _))
+      .WillOnce([&](absl::Span<const absl::string_view> data,
+                    const quiche::StreamWriteOptions& options) {
+        correct_message = true;
+        EXPECT_EQ(*ExtractMessageType(data[0]),
+                  MoqtMessageType::kSubscribeError);
+        return absl::OkStatus();
+      });
+  stream_input->OnSubscribeMessage(request);
+  EXPECT_TRUE(correct_message);
+
+  // Second request is a protocol violation.
+  request.subscribe_id = 0;
+  request.track_alias = 3;
+  request.full_track_name = FullTrackName({"dead", "beef"});
+  EXPECT_CALL(mock_session_,
+              CloseSession(static_cast<uint64_t>(MoqtError::kProtocolViolation),
+                           "Subscribe ID not monotonically increasing"))
+      .Times(1);
+  stream_input->OnSubscribeMessage(request);
+}
+
 TEST_F(MoqtSessionTest, TooManySubscribes) {
   MoqtSessionPeer::set_next_subscribe_id(&session_,
                                          kDefaultInitialMaxSubscribeId);
@@ -1148,7 +1234,7 @@
   control_stream->OnSubscribeOkMessage(ok);
 }
 
-TEST_F(MoqtSessionTest, CreateIncomingDataStreamAndSend) {
+TEST_F(MoqtSessionTest, CreateOutgoingDataStreamAndSend) {
   FullTrackName ftn("foo", "bar");
   auto track = SetupPublisher(ftn, MoqtForwardingPreference::kSubgroup,
                               FullSequence(4, 2));
@@ -1177,7 +1263,7 @@
 
   // Verify first six message fields are sent correctly
   bool correct_message = false;
-  const std::string kExpectedMessage = {0x04, 0x02, 0x05, 0x00, 0x00};
+  const std::string kExpectedMessage = {0x04, 0x02, 0x05, 0x00, 0x7f};
   EXPECT_CALL(mock_stream, Writev(_, _))
       .WillOnce([&](absl::Span<const absl::string_view> data,
                     const quiche::StreamWriteOptions& options) {
@@ -1195,6 +1281,8 @@
   subscription->OnNewObjectAvailable(FullSequence(5, 0));
   EXPECT_TRUE(correct_message);
   EXPECT_FALSE(fin);
+  EXPECT_EQ(MoqtSessionPeer::LargestSentForSubscription(&session_, 0),
+            FullSequence(5, 0));
 }
 
 TEST_F(MoqtSessionTest, FinDataStreamFromCache) {
@@ -1224,9 +1312,9 @@
   EXPECT_CALL(mock_session_, GetStreamById(kOutgoingUniStreamId))
       .WillRepeatedly(Return(&mock_stream));
 
-  // Verify first six message fields are sent correctly
+  // Verify first five message fields are sent correctly
   bool correct_message = false;
-  const std::string kExpectedMessage = {0x04, 0x02, 0x05, 0x00, 0x00};
+  const std::string kExpectedMessage = {0x04, 0x02, 0x05, 0x00, 0x7f};
   EXPECT_CALL(mock_stream, Writev(_, _))
       .WillOnce([&](absl::Span<const absl::string_view> data,
                     const quiche::StreamWriteOptions& options) {
@@ -1275,7 +1363,7 @@
 
   // Verify first six message fields are sent correctly
   bool correct_message = false;
-  const std::string kExpectedMessage = {0x04, 0x02, 0x05, 0x00, 0x00};
+  const std::string kExpectedMessage = {0x04, 0x02, 0x05, 0x00, 0x7f};
   EXPECT_CALL(mock_stream, Writev(_, _))
       .WillOnce([&](absl::Span<const absl::string_view> data,
                     const quiche::StreamWriteOptions& options) {
@@ -1327,7 +1415,7 @@
 
   // Verify first six message fields are sent correctly
   bool correct_message = false;
-  const std::string kExpectedMessage = {0x04, 0x02, 0x05, 0x00, 0x00};
+  const std::string kExpectedMessage = {0x04, 0x02, 0x05, 0x00, 0x7f};
   EXPECT_CALL(mock_stream, Writev(_, _))
       .WillOnce([&](absl::Span<const absl::string_view> data,
                     const quiche::StreamWriteOptions& options) {
@@ -1994,6 +2082,316 @@
   session_.OnCanCreateNewOutgoingUnidirectionalStream();
 }
 
+TEST_F(MoqtSessionTest, FetchReturnsOk) {
+  webtransport::test::MockStream control_stream;
+  std::unique_ptr<MoqtControlParserVisitor> stream_input =
+      MoqtSessionPeer::CreateControlStream(&session_, &control_stream);
+  FullTrackName ftn("foo", "bar");
+  MoqtFetch request = {
+      /*subscribe_id=*/0,
+      /*full_track_name=*/ftn,
+      /*subscriber_priority=*/0x80,
+      /*group_order=*/std::nullopt,
+      /*start=*/FullSequence(0, 0),
+      /*end_group=*/1,
+      /*end_object=*/std::nullopt,
+      /*parameters=*/MoqtSubscribeParameters(),
+  };
+  bool correct_message = false;
+  auto track = std::make_shared<MockTrackPublisher>(ftn);
+  publisher_.Add(track);
+
+  auto fetch_task_ptr = std::make_unique<MockFetchTask>();
+  MockFetchTask* fetch_task = fetch_task_ptr.get();
+  EXPECT_CALL(*track, Fetch(_, _, _, _))
+      .WillOnce(Return(std::move(fetch_task_ptr)));
+  // Compose and send the FETCH_OK.
+  EXPECT_CALL(*track, GetDeliveryOrder())
+      .WillRepeatedly(Return(MoqtDeliveryOrder::kAscending));
+  EXPECT_CALL(*fetch_task, GetLargestId()).WillOnce(Return(FullSequence(0, 0)));
+  EXPECT_CALL(control_stream, Writev(_, _))
+      .WillOnce([&](absl::Span<const absl::string_view> data,
+                    const quiche::StreamWriteOptions& options) {
+        correct_message = true;
+        EXPECT_EQ(*ExtractMessageType(data[0]), MoqtMessageType::kFetchOk);
+        return absl::OkStatus();
+      });
+  // Stream can't open yet.
+  EXPECT_CALL(mock_session_, CanOpenNextOutgoingUnidirectionalStream)
+      .WillOnce(Return(false));
+  stream_input->OnFetchMessage(request);
+  EXPECT_TRUE(correct_message);
+}
+
+TEST_F(MoqtSessionTest, FetchReturnsOkImmediateOpen) {
+  webtransport::test::MockStream control_stream;
+  std::unique_ptr<MoqtControlParserVisitor> stream_input =
+      MoqtSessionPeer::CreateControlStream(&session_, &control_stream);
+  FullTrackName ftn("foo", "bar");
+  MoqtFetch request = {
+      /*subscribe_id=*/0,
+      /*full_track_name=*/ftn,
+      /*subscriber_priority=*/0x80,
+      /*group_order=*/std::nullopt,
+      /*start=*/FullSequence(0, 0),
+      /*end_group=*/1,
+      /*end_object=*/std::nullopt,
+      /*parameters=*/MoqtSubscribeParameters(),
+  };
+  bool correct_message = false;
+  auto track = std::make_shared<MockTrackPublisher>(ftn);
+  publisher_.Add(track);
+
+  auto fetch_task_ptr = std::make_unique<MockFetchTask>();
+  MockFetchTask* fetch_task = fetch_task_ptr.get();
+  EXPECT_CALL(*track, Fetch(_, _, _, _))
+      .WillOnce(Return(std::move(fetch_task_ptr)));
+  // Compose and send the FETCH_OK.
+  EXPECT_CALL(*track, GetDeliveryOrder())
+      .WillRepeatedly(Return(MoqtDeliveryOrder::kAscending));
+  EXPECT_CALL(*fetch_task, GetLargestId()).WillOnce(Return(FullSequence(0, 0)));
+  EXPECT_CALL(control_stream, Writev(_, _))
+      .WillOnce([&](absl::Span<const absl::string_view> data,
+                    const quiche::StreamWriteOptions& options) {
+        correct_message = true;
+        EXPECT_EQ(*ExtractMessageType(data[0]), MoqtMessageType::kFetchOk);
+        return absl::OkStatus();
+      });
+  // Open stream immediately.
+  EXPECT_CALL(mock_session_, CanOpenNextOutgoingUnidirectionalStream)
+      .WillOnce(Return(true));
+  webtransport::test::MockStream data_stream;
+  EXPECT_CALL(mock_session_, OpenOutgoingUnidirectionalStream())
+      .WillOnce(Return(&data_stream));
+  std::unique_ptr<webtransport::StreamVisitor> stream_visitor;
+  EXPECT_CALL(data_stream, SetVisitor(_))
+      .WillOnce(
+          Invoke([&](std::unique_ptr<webtransport::StreamVisitor> visitor) {
+            stream_visitor = std::move(visitor);
+          }));
+  EXPECT_CALL(data_stream, CanWrite()).WillRepeatedly(Return(true));
+  EXPECT_CALL(data_stream, visitor()).WillOnce(Invoke([&]() {
+    return stream_visitor.get();
+  }));
+  EXPECT_CALL(*fetch_task, GetNextObject(_))
+      .WillOnce(Return(MoqtFetchTask::GetNextObjectResult::kPending));
+  stream_input->OnFetchMessage(request);
+  EXPECT_TRUE(correct_message);
+
+  // Signal the stream that pending object is now available.
+  correct_message = false;
+  EXPECT_CALL(data_stream, CanWrite()).WillRepeatedly(Return(true));
+  EXPECT_CALL(*fetch_task, GetNextObject(_))
+      .WillOnce(Invoke([](PublishedObject& output) {
+        output.sequence = FullSequence(0, 0, 0);
+        output.status = MoqtObjectStatus::kNormal;
+        output.publisher_priority = 128;
+        output.payload = MemSliceFromString("foo");
+        output.fin_after_this = true;
+        return MoqtFetchTask::GetNextObjectResult::kSuccess;
+      }))
+      .WillOnce(Invoke([](PublishedObject& /*output*/) {
+        return MoqtFetchTask::GetNextObjectResult::kPending;
+      }));
+  EXPECT_CALL(data_stream, Writev(_, _))
+      .WillOnce([&](absl::Span<const absl::string_view> data,
+                    const quiche::StreamWriteOptions& options) {
+        correct_message = true;
+        quic::QuicDataReader reader(data[0]);
+        uint64_t type;
+        EXPECT_TRUE(reader.ReadVarInt62(&type));
+        EXPECT_EQ(type, static_cast<uint64_t>(
+                            MoqtDataStreamType::kStreamHeaderFetch));
+        return absl::OkStatus();
+      });
+  fetch_task->callback_();
+  EXPECT_TRUE(correct_message);
+}
+
+TEST_F(MoqtSessionTest, InvalidFetch) {
+  // Update the state so that it expects ID > 0 next time.
+  MoqtSessionPeer::ValidateSubscribeId(&session_, 0);
+  webtransport::test::MockStream control_stream;
+  std::unique_ptr<MoqtControlParserVisitor> stream_input =
+      MoqtSessionPeer::CreateControlStream(&session_, &control_stream);
+  FullTrackName ftn("foo", "bar");
+  MoqtFetch request = {
+      /*subscribe_id=*/0,  // Subscribe ID is too low.
+      /*full_track_name=*/ftn,
+      /*subscriber_priority=*/0x80,
+      /*group_order=*/std::nullopt,
+      /*start=*/FullSequence(0, 0),
+      /*end_group=*/1,
+      /*end_object=*/std::nullopt,
+      /*parameters=*/MoqtSubscribeParameters(),
+  };
+  EXPECT_CALL(mock_session_,
+              CloseSession(static_cast<uint64_t>(MoqtError::kProtocolViolation),
+                           "Subscribe ID not monotonically increasing"))
+      .Times(1);
+  stream_input->OnFetchMessage(request);
+}
+
+TEST_F(MoqtSessionTest, FetchFails) {
+  webtransport::test::MockStream control_stream;
+  std::unique_ptr<MoqtControlParserVisitor> stream_input =
+      MoqtSessionPeer::CreateControlStream(&session_, &control_stream);
+  FullTrackName ftn("foo", "bar");
+  MoqtFetch request = {
+      /*subscribe_id=*/0,
+      /*full_track_name=*/ftn,
+      /*subscriber_priority=*/0x80,
+      /*group_order=*/std::nullopt,
+      /*start=*/FullSequence(0, 0),
+      /*end_group=*/1,
+      /*end_object=*/std::nullopt,
+      /*parameters=*/MoqtSubscribeParameters(),
+  };
+  bool correct_message = false;
+  auto track = std::make_shared<MockTrackPublisher>(ftn);
+  publisher_.Add(track);
+
+  auto fetch_task_ptr = std::make_unique<MockFetchTask>();
+  MockFetchTask* fetch_task = fetch_task_ptr.get();
+  EXPECT_CALL(*track, Fetch(_, _, _, _))
+      .WillOnce(Return(std::move(fetch_task_ptr)));
+  EXPECT_CALL(*fetch_task, GetStatus())
+      .WillRepeatedly(Return(absl::Status(absl::StatusCode::kInternal, "foo")));
+  EXPECT_CALL(control_stream, Writev(_, _))
+      .WillOnce([&](absl::Span<const absl::string_view> data,
+                    const quiche::StreamWriteOptions& options) {
+        correct_message = true;
+        EXPECT_EQ(*ExtractMessageType(data[0]), MoqtMessageType::kFetchError);
+        return absl::OkStatus();
+      });
+  stream_input->OnFetchMessage(request);
+  EXPECT_TRUE(correct_message);
+}
+
+TEST_F(MoqtSessionTest, FetchDelivery) {
+  constexpr uint64_t kFetchId = 0;
+  MockFetchTask* fetch = MoqtSessionPeer::AddFetch(&session_, kFetchId);
+  // Stream creation started out blocked. Allow its creation, but data is
+  // blocked.
+  webtransport::test::MockStream data_stream;
+  EXPECT_CALL(mock_session_, CanOpenNextOutgoingUnidirectionalStream())
+      .WillRepeatedly(Return(true));
+  EXPECT_CALL(mock_session_, OpenOutgoingUnidirectionalStream())
+      .WillOnce(Return(&data_stream));
+  std::unique_ptr<webtransport::StreamVisitor> stream_visitor;
+  EXPECT_CALL(data_stream, GetStreamId()).WillOnce(Return(4));
+  EXPECT_CALL(data_stream, SetVisitor(_))
+      .WillOnce(
+          Invoke([&](std::unique_ptr<webtransport::StreamVisitor> visitor) {
+            stream_visitor = std::move(visitor);
+          }));
+  EXPECT_CALL(data_stream, CanWrite()).WillOnce(Return(false));
+  session_.OnCanCreateNewOutgoingUnidirectionalStream();
+  // Unblock the stream. Provide one object and an EOF.
+  EXPECT_CALL(data_stream, CanWrite()).WillRepeatedly(Return(true));
+  EXPECT_CALL(*fetch, GetNextObject(_))
+      .WillOnce(Invoke([](PublishedObject& output) {
+        output.sequence = FullSequence(0, 0, 0);
+        output.status = MoqtObjectStatus::kNormal;
+        output.publisher_priority = 128;
+        output.payload = MemSliceFromString("foo");
+        output.fin_after_this = true;
+        return MoqtFetchTask::GetNextObjectResult::kSuccess;
+      }))
+      .WillOnce(Invoke([](PublishedObject& /*output*/) {
+        return MoqtFetchTask::GetNextObjectResult::kEof;
+      }));
+
+  int objects_received = 0;
+  EXPECT_CALL(data_stream, Writev(_, _))
+      .WillOnce(Invoke([&](absl::Span<const absl::string_view> data,
+                           const quiche::StreamWriteOptions& options) {
+        ++objects_received;
+        quic::QuicDataReader reader(data[0]);
+        uint64_t type;
+        EXPECT_TRUE(reader.ReadVarInt62(&type));
+        EXPECT_EQ(type, static_cast<uint64_t>(
+                            MoqtDataStreamType::kStreamHeaderFetch));
+        EXPECT_FALSE(options.send_fin());  // fin_after_this is ignored.
+        return absl::OkStatus();
+      }))
+      .WillOnce(Invoke([&](absl::Span<const absl::string_view> data,
+                           const quiche::StreamWriteOptions& options) {
+        ++objects_received;
+        EXPECT_TRUE(data.empty());
+        EXPECT_TRUE(options.send_fin());
+        return absl::OkStatus();
+      }));
+  stream_visitor->OnCanWrite();
+  EXPECT_EQ(objects_received, 2);
+}
+
+TEST_F(MoqtSessionTest, FetchNonNormalObjects) {
+  constexpr uint64_t kFetchId = 0;
+  MockFetchTask* fetch = MoqtSessionPeer::AddFetch(&session_, kFetchId);
+  // Stream creation started out blocked. Allow its creation, but data is
+  // blocked.
+  webtransport::test::MockStream data_stream;
+  EXPECT_CALL(mock_session_, CanOpenNextOutgoingUnidirectionalStream())
+      .WillRepeatedly(Return(true));
+  EXPECT_CALL(mock_session_, OpenOutgoingUnidirectionalStream())
+      .WillOnce(Return(&data_stream));
+  std::unique_ptr<webtransport::StreamVisitor> stream_visitor;
+  EXPECT_CALL(data_stream, SetVisitor(_))
+      .WillOnce(
+          Invoke([&](std::unique_ptr<webtransport::StreamVisitor> visitor) {
+            stream_visitor = std::move(visitor);
+          }));
+  EXPECT_CALL(data_stream, CanWrite()).WillOnce(Return(false));
+  session_.OnCanCreateNewOutgoingUnidirectionalStream();
+  // Unblock the stream. Provide one object and an EOF.
+  EXPECT_CALL(data_stream, CanWrite()).WillRepeatedly(Return(true));
+  EXPECT_CALL(*fetch, GetNextObject(_))
+      .WillOnce(Invoke([](PublishedObject& output) {
+        // DoesNotExist will be skipped.
+        output.sequence = FullSequence(0, 0, 0);
+        output.status = MoqtObjectStatus::kObjectDoesNotExist;
+        output.publisher_priority = 128;
+        output.payload = MemSliceFromString("");
+        output.fin_after_this = true;
+        return MoqtFetchTask::GetNextObjectResult::kSuccess;
+      }))
+      .WillOnce(Invoke([](PublishedObject& output) {
+        output.sequence = FullSequence(0, 0, 1);
+        output.status = MoqtObjectStatus::kEndOfGroup;
+        output.publisher_priority = 128;
+        output.payload = MemSliceFromString("");
+        output.fin_after_this = true;
+        return MoqtFetchTask::GetNextObjectResult::kSuccess;
+      }))
+      .WillOnce(Invoke([](PublishedObject& /*output*/) {
+        return MoqtFetchTask::GetNextObjectResult::kEof;
+      }));
+
+  int objects_received = 0;
+  EXPECT_CALL(data_stream, Writev(_, _))
+      .WillOnce(Invoke([&](absl::Span<const absl::string_view> data,
+                           const quiche::StreamWriteOptions& options) {
+        ++objects_received;
+        quic::QuicDataReader reader(data[0]);
+        uint64_t type;
+        EXPECT_TRUE(reader.ReadVarInt62(&type));
+        EXPECT_EQ(type, static_cast<uint64_t>(
+                            MoqtDataStreamType::kStreamHeaderFetch));
+        EXPECT_FALSE(options.send_fin());
+        return absl::OkStatus();
+      }))
+      .WillOnce(Invoke([&](absl::Span<const absl::string_view> data,
+                           const quiche::StreamWriteOptions& options) {
+        ++objects_received;
+        EXPECT_TRUE(data.empty());
+        EXPECT_TRUE(options.send_fin());
+        return absl::OkStatus();
+      }));
+  stream_visitor->OnCanWrite();
+  EXPECT_EQ(objects_received, 2);
+}
+
 // TODO: re-enable this test once this behavior is re-implemented.
 #if 0
 TEST_F(MoqtSessionTest, SubscribeUpdateClosesSubscription) {