diff --git a/quiche/quic/moqt/moqt_messages.cc b/quiche/quic/moqt/moqt_messages.cc
index e395190..3e0a226 100644
--- a/quiche/quic/moqt/moqt_messages.cc
+++ b/quiche/quic/moqt/moqt_messages.cc
@@ -9,13 +9,14 @@
 #include <vector>
 
 #include "absl/algorithm/container.h"
+#include "absl/status/status.h"
 #include "absl/strings/escaping.h"
 #include "absl/strings/str_cat.h"
 #include "absl/strings/str_join.h"
 #include "absl/strings/string_view.h"
 #include "absl/types/span.h"
 #include "quiche/quic/platform/api/quic_bug_tracker.h"
-#include "quiche/common/platform/api/quiche_bug_tracker.h"
+#include "quiche/web_transport/web_transport.h"
 
 namespace moqt {
 
@@ -200,4 +201,16 @@
 FullTrackName::FullTrackName(absl::Span<const absl::string_view> elements)
     : tuple_(elements.begin(), elements.end()) {}
 
+absl::Status MoqtStreamErrorToStatus(webtransport::StreamErrorCode error_code,
+                                     absl::string_view reason_phrase) {
+  switch (error_code) {
+    case kResetCodeSubscriptionGone:
+      return absl::NotFoundError(reason_phrase);
+    case kResetCodeTimedOut:
+      return absl::DeadlineExceededError(reason_phrase);
+    default:
+      return absl::UnknownError(reason_phrase);
+  }
+}
+
 }  // namespace moqt
diff --git a/quiche/quic/moqt/moqt_messages.h b/quiche/quic/moqt/moqt_messages.h
index 4b189f9..508d103 100644
--- a/quiche/quic/moqt/moqt_messages.h
+++ b/quiche/quic/moqt/moqt_messages.h
@@ -16,6 +16,7 @@
 #include <vector>
 
 #include "absl/container/inlined_vector.h"
+#include "absl/status/status.h"
 #include "absl/strings/str_format.h"
 #include "absl/strings/string_view.h"
 #include "absl/types/span.h"
@@ -24,6 +25,7 @@
 #include "quiche/quic/core/quic_versions.h"
 #include "quiche/quic/moqt/moqt_priority.h"
 #include "quiche/common/platform/api/quiche_export.h"
+#include "quiche/web_transport/web_transport.h"
 
 namespace moqt {
 
@@ -121,9 +123,10 @@
 // Error codes used by MoQT to reset streams.
 // TODO: update with spec-defined error codes once those are available, see
 // <https://github.com/moq-wg/moq-transport/issues/481>.
-inline constexpr uint64_t kResetCodeUnknown = 0x00;
-inline constexpr uint64_t kResetCodeSubscriptionGone = 0x01;
-inline constexpr uint64_t kResetCodeTimedOut = 0x02;
+inline constexpr webtransport::StreamErrorCode kResetCodeUnknown = 0x00;
+inline constexpr webtransport::StreamErrorCode kResetCodeSubscriptionGone =
+    0x01;
+inline constexpr webtransport::StreamErrorCode kResetCodeTimedOut = 0x02;
 
 enum class QUICHE_EXPORT MoqtRole : uint64_t {
   kPublisher = 0x1,
@@ -583,6 +586,9 @@
 MoqtDataStreamType GetMessageTypeForForwardingPreference(
     MoqtForwardingPreference preference);
 
+absl::Status MoqtStreamErrorToStatus(webtransport::StreamErrorCode error_code,
+                                     absl::string_view reason_phrase);
+
 }  // namespace moqt
 
 #endif  // QUICHE_QUIC_MOQT_MOQT_MESSAGES_H_
diff --git a/quiche/quic/moqt/moqt_session.cc b/quiche/quic/moqt/moqt_session.cc
index c181578..8496188 100644
--- a/quiche/quic/moqt/moqt_session.cc
+++ b/quiche/quic/moqt/moqt_session.cc
@@ -417,7 +417,7 @@
 }
 
 void MoqtSession::Unsubscribe(const FullTrackName& name) {
-  RemoteTrack* track = RemoteTrackByName(name);
+  SubscribeRemoteTrack* track = RemoteTrackByName(name);
   if (track == nullptr) {
     return;
   }
@@ -425,10 +425,45 @@
   message.subscribe_id = track->subscribe_id();
   SendControlMessage(framer_.SerializeUnsubscribe(message));
   // Destroy state.
-  upstream_by_name_.erase(name);
+  subscribe_by_name_.erase(name);
+  subscribe_by_alias_.erase(track->track_alias());
   upstream_by_id_.erase(track->subscribe_id());
-  subscribe_by_alias_.erase(
-      static_cast<SubscribeRemoteTrack*>(track)->track_alias());
+}
+
+bool MoqtSession::Fetch(const FullTrackName& name,
+                        FetchResponseCallback callback, FullSequence start,
+                        uint64_t end_group, std::optional<uint64_t> end_object,
+                        MoqtPriority priority,
+                        std::optional<MoqtDeliveryOrder> delivery_order,
+                        MoqtSubscribeParameters parameters) {
+  if (peer_role_ == MoqtRole::kSubscriber) {
+    QUIC_DLOG(INFO) << ENDPOINT << "Tried to send FETCH to subscriber peer";
+    return false;
+  }
+  // TODO(martinduke): support authorization info
+  if (next_subscribe_id_ >= peer_max_subscribe_id_) {
+    QUIC_DLOG(INFO) << ENDPOINT << "Tried to send FETCH with ID "
+                    << next_subscribe_id_
+                    << " which is greater than the maximum ID "
+                    << peer_max_subscribe_id_;
+    return false;
+  }
+  MoqtFetch message;
+  message.full_track_name = name;
+  message.subscribe_id = next_subscribe_id_++;
+  message.start_object = start;
+  message.end_group = end_group;
+  message.end_object = end_object;
+  message.subscriber_priority = priority;
+  message.group_order = delivery_order;
+  message.parameters = parameters;
+  message.parameters.object_ack_window = std::nullopt;
+  SendControlMessage(framer_.SerializeFetch(message));
+  QUIC_DLOG(INFO) << ENDPOINT << "Sent FETCH message for "
+                  << message.full_track_name;
+  auto fetch = std::make_unique<UpstreamFetch>(message, std::move(callback));
+  upstream_by_id_.emplace(message.subscribe_id, std::move(fetch));
+  return true;
 }
 
 void MoqtSession::PublishedFetch::FetchStreamVisitor::OnCanWrite() {
@@ -517,7 +552,7 @@
                     << peer_max_subscribe_id_;
     return false;
   }
-  if (upstream_by_name_.contains(message.full_track_name)) {
+  if (subscribe_by_name_.contains(message.full_track_name)) {
     QUIC_DLOG(INFO) << ENDPOINT << "Tried to send SUBSCRIBE for track "
                     << message.full_track_name
                     << " which is already subscribed";
@@ -529,6 +564,13 @@
     return false;
   }
   message.subscribe_id = next_subscribe_id_++;
+  if (provided_track_alias.has_value()) {
+    message.track_alias = *provided_track_alias;
+    next_remote_track_alias_ =
+        std::max(next_remote_track_alias_, *provided_track_alias) + 1;
+  } else {
+    message.track_alias = next_remote_track_alias_++;
+  }
   message.track_alias =
       provided_track_alias.value_or(next_remote_track_alias_++);
   if (SupportsObjectAck() && visitor != nullptr) {
@@ -546,9 +588,9 @@
   QUIC_DLOG(INFO) << ENDPOINT << "Sent SUBSCRIBE message for "
                   << message.full_track_name;
   auto track = std::make_unique<SubscribeRemoteTrack>(message, visitor);
-  upstream_by_name_.emplace(message.full_track_name, track.get());
-  upstream_by_id_.emplace(message.subscribe_id, track.get());
-  subscribe_by_alias_.emplace(message.track_alias, std::move(track));
+  subscribe_by_name_.emplace(message.full_track_name, track.get());
+  subscribe_by_alias_.emplace(message.track_alias, track.get());
+  upstream_by_id_.emplace(message.subscribe_id, std::move(track));
   return true;
 }
 
@@ -608,7 +650,7 @@
   if (it == subscribe_by_alias_.end()) {
     return nullptr;
   }
-  return it->second.get();
+  return it->second;
 }
 
 RemoteTrack* MoqtSession::RemoteTrackById(uint64_t subscribe_id) {
@@ -616,12 +658,13 @@
   if (it == upstream_by_id_.end()) {
     return nullptr;
   }
-  return it->second;
+  return it->second.get();
 }
 
-RemoteTrack* MoqtSession::RemoteTrackByName(const FullTrackName& name) {
-  auto it = upstream_by_name_.find(name);
-  if (it == upstream_by_name_.end()) {
+SubscribeRemoteTrack* MoqtSession::RemoteTrackByName(
+    const FullTrackName& name) {
+  auto it = subscribe_by_name_.find(name);
+  if (it == subscribe_by_name_.end()) {
     return nullptr;
   }
   return it->second;
@@ -945,12 +988,10 @@
                   << ", error = " << static_cast<int>(message.error_code)
                   << " (" << message.reason_phrase << ")";
   SubscribeRemoteTrack* subscribe = static_cast<SubscribeRemoteTrack*>(track);
-  // Delete secondary references to the track. Preserve the owner
-  // (subscribe_by_alias_) to get the original subscribe, if needed. Erasing the
-  // other references now prevents an error due to a duplicate subscription in
-  // Subscribe().
-  session_->upstream_by_id_.erase(subscribe->subscribe_id());
-  session_->upstream_by_name_.erase(subscribe->full_track_name());
+  // Delete the by-name entry at this point prevents Subscribe() from throwing
+  // an error due to a duplicate track name. The other entries for this
+  // subscribe will be deleted after calling Subscribe().
+  session_->subscribe_by_name_.erase(subscribe->full_track_name());
   if (message.error_code == SubscribeErrorCode::kRetryTrackAlias) {
     // Automatically resubscribe with new alias.
     MoqtSubscribe& subscribe_message = subscribe->GetSubscribe();
@@ -961,6 +1002,7 @@
                                   message.reason_phrase);
   }
   session_->subscribe_by_alias_.erase(subscribe->track_alias());
+  session_->upstream_by_id_.erase(subscribe->subscribe_id());
 }
 
 void MoqtSession::ControlStream::OnUnsubscribeMessage(
@@ -1196,6 +1238,80 @@
   }
 }
 
+void MoqtSession::ControlStream::OnFetchOkMessage(const MoqtFetchOk& message) {
+  RemoteTrack* track = session_->RemoteTrackById(message.subscribe_id);
+  if (track == nullptr) {
+    QUIC_DLOG(INFO) << ENDPOINT << "Received the FETCH_OK for "
+                    << "subscribe_id = " << message.subscribe_id
+                    << " but no track exists";
+    // Subscription state might have been destroyed for internal reasons.
+    return;
+  }
+  if (!track->is_fetch()) {
+    session_->Error(MoqtError::kProtocolViolation,
+                    "Received FETCH_OK for a SUBSCRIBE");
+    return;
+  }
+  QUIC_DLOG(INFO) << ENDPOINT << "Received the FETCH_OK for subscribe_id = "
+                  << message.subscribe_id << " " << track->full_track_name();
+  UpstreamFetch* fetch = static_cast<UpstreamFetch*>(track);
+  fetch->OnFetchResult(message.largest_id, absl::OkStatus(),
+                       [=, session = session_]() {
+                         session->CancelFetch(message.subscribe_id);
+                       });
+}
+
+void MoqtSession::ControlStream::OnFetchErrorMessage(
+    const MoqtFetchError& message) {
+  RemoteTrack* track = session_->RemoteTrackById(message.subscribe_id);
+  if (track == nullptr) {
+    QUIC_DLOG(INFO) << ENDPOINT << "Received the FETCH_ERROR for "
+                    << "subscribe_id = " << message.subscribe_id
+                    << " but no track exists";
+    // Subscription state might have been destroyed for internal reasons.
+    return;
+  }
+  if (!track->is_fetch()) {
+    session_->Error(MoqtError::kProtocolViolation,
+                    "Received FETCH_ERROR for a SUBSCRIBE");
+    return;
+  }
+  if (!track->ErrorIsAllowed()) {
+    session_->Error(MoqtError::kProtocolViolation,
+                    "Received FETCH_ERROR after FETCH_OK or objects");
+    return;
+  }
+  QUIC_DLOG(INFO) << ENDPOINT << "Received the FETCH_ERROR for "
+                  << "subscribe_id = " << message.subscribe_id << " ("
+                  << track->full_track_name() << ")"
+                  << ", error = " << static_cast<int>(message.error_code)
+                  << " (" << message.reason_phrase << ")";
+  UpstreamFetch* fetch = static_cast<UpstreamFetch*>(track);
+  absl::Status status;
+  switch (message.error_code) {
+    case moqt::SubscribeErrorCode::kInternalError:
+      status = absl::InternalError(message.reason_phrase);
+      break;
+    case SubscribeErrorCode::kInvalidRange:
+      status = absl::OutOfRangeError(message.reason_phrase);
+      break;
+    case SubscribeErrorCode::kTrackDoesNotExist:
+      status = absl::NotFoundError(message.reason_phrase);
+      break;
+    case SubscribeErrorCode::kUnauthorized:
+      status = absl::UnauthenticatedError(message.reason_phrase);
+      break;
+    case SubscribeErrorCode::kTimeout:
+      status = absl::DeadlineExceededError(message.reason_phrase);
+      break;
+    default:
+      status = absl::UnknownError(message.reason_phrase);
+      break;
+  }
+  fetch->OnFetchResult(FullSequence(0, 0), status, nullptr);
+  session_->upstream_by_id_.erase(message.subscribe_id);
+}
+
 void MoqtSession::ControlStream::OnParsingError(MoqtError error_code,
                                                 absl::string_view reason) {
   session_->Error(error_code, absl::StrCat("Parse error: ", reason));
@@ -1688,6 +1804,22 @@
   return true;
 }
 
+void MoqtSession::CancelFetch(uint64_t subscribe_id) {
+  // This is only called from the callback where UpstreamFetchTask has been
+  // destroyed, so there is no need to notify the application.
+  upstream_by_id_.erase(subscribe_id);
+  ControlStream* stream = GetControlStream();
+  if (stream == nullptr) {
+    return;
+  }
+  MoqtFetchCancel message;
+  message.subscribe_id = subscribe_id;
+  stream->SendOrBufferMessage(framer_.SerializeFetchCancel(message));
+  // The FETCH_CANCEL will cause a RESET_STREAM to return, which would be the
+  // same as a STOP_SENDING. However, a FETCH_CANCEL works even if the stream
+  // hasn't opened yet.
+}
+
 void MoqtSession::PublishedSubscription::SendDatagram(FullSequence sequence) {
   std::optional<PublishedObject> object =
       track_publisher_->GetCachedObject(sequence);
diff --git a/quiche/quic/moqt/moqt_session.h b/quiche/quic/moqt/moqt_session.h
index ca8e80c..adf4b7e 100644
--- a/quiche/quic/moqt/moqt_session.h
+++ b/quiche/quic/moqt/moqt_session.h
@@ -182,6 +182,15 @@
   // Returns false if the subscription is not found. The session immediately
   // destroys all subscription state.
   void Unsubscribe(const FullTrackName& name);
+  // |callback| will be called when FETCH_OK or FETCH_ERROR is received, and
+  // delivers a pointer to MoqtFetchTask for application use. The callback
+  // transfers ownership of MoqtFetchTask to the application.
+  // To cancel a FETCH, simply destroy the FetchTask.
+  bool Fetch(const FullTrackName& name, FetchResponseCallback callback,
+             FullSequence start, uint64_t end_group,
+             std::optional<uint64_t> end_object, MoqtPriority priority,
+             std::optional<MoqtDeliveryOrder> delivery_order,
+             MoqtSubscribeParameters parameters = MoqtSubscribeParameters());
 
   webtransport::Session* session() { return session_; }
   MoqtSessionCallbacks& callbacks() { return callbacks_; }
@@ -264,8 +273,8 @@
     void OnMaxSubscribeIdMessage(const MoqtMaxSubscribeId& 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 {}
+    void OnFetchOkMessage(const MoqtFetchOk& message) override;
+    void OnFetchErrorMessage(const MoqtFetchError& message) override;
     void OnObjectAckMessage(const MoqtObjectAck& message) override {
       auto subscription_it =
           session_->published_subscriptions_.find(message.subscribe_id);
@@ -334,7 +343,6 @@
     webtransport::Stream* stream_;
     // Once the subscribe ID is identified, set it here.
     quiche::QuicheWeakPtr<RemoteTrack> track_;
-    // std::optional<uint64_t> subscribe_id_ = std::nullopt;
     MoqtDataParser parser_;
     std::string partial_object_;
   };
@@ -562,7 +570,7 @@
 
   SubscribeRemoteTrack* RemoteTrackByAlias(uint64_t track_alias);
   RemoteTrack* RemoteTrackById(uint64_t subscribe_id);
-  RemoteTrack* RemoteTrackByName(const FullTrackName& name);
+  SubscribeRemoteTrack* RemoteTrackByName(const FullTrackName& name);
 
   // Checks that a subscribe ID from a SUBSCRIBE or FETCH is valid, and throws
   // a session error if is not.
@@ -576,6 +584,8 @@
                            MoqtDataStreamType type, bool is_first_on_stream,
                            bool fin);
 
+  void CancelFetch(uint64_t subscribe_id);
+
   // Sends an OBJECT_ACK message for a specific subscribe ID.
   void SendObjectAck(uint64_t subscribe_id, uint64_t group_id,
                      uint64_t object_id,
@@ -606,16 +616,13 @@
   std::string error_;
 
   // Upstream SUBSCRIBE state.
-  // All the tracks the session is subscribed to, indexed by track_alias.
-  absl::flat_hash_map<uint64_t, std::unique_ptr<SubscribeRemoteTrack>>
-      subscribe_by_alias_;
-  // Upstream SUBSCRIBEs indexed by subscribe_id.
-  // TODO(martinduke): Add fetches to this.
-  absl::flat_hash_map<uint64_t, RemoteTrack*> upstream_by_id_;
-  // The application only has track names, so this allows MoqtSession to
-  // quickly find what it's looking for. Also allows a quick check for duplicate
-  // subscriptions.
-  absl::flat_hash_map<FullTrackName, RemoteTrack*> upstream_by_name_;
+  // Upstream SUBSCRIBEs and FETCHes, indexed by subscribe_id.
+  absl::flat_hash_map<uint64_t, std::unique_ptr<RemoteTrack>> upstream_by_id_;
+  // All SUBSCRIBEs, indexed by track_alias.
+  absl::flat_hash_map<uint64_t, SubscribeRemoteTrack*> subscribe_by_alias_;
+  // All SUBSCRIBEs, indexed by track name.
+  absl::flat_hash_map<FullTrackName, SubscribeRemoteTrack*> subscribe_by_name_;
+  // The next track alias to guess on a SUBSCRIBE.
   uint64_t next_remote_track_alias_ = 0;
   // The next subscribe ID that the local endpoint can send.
   uint64_t next_subscribe_id_ = 0;
diff --git a/quiche/quic/moqt/moqt_session_test.cc b/quiche/quic/moqt/moqt_session_test.cc
index 1c845b6..172a152 100644
--- a/quiche/quic/moqt/moqt_session_test.cc
+++ b/quiche/quic/moqt/moqt_session_test.cc
@@ -2202,6 +2202,60 @@
   stream_input->OnSubscribeAnnouncesMessage(announces);
 }
 
+TEST_F(MoqtSessionTest, FetchThenOkThenCancel) {
+  webtransport::test::MockStream mock_stream;
+  std::unique_ptr<MoqtControlParserVisitor> stream_input =
+      MoqtSessionPeer::CreateControlStream(&session_, &mock_stream);
+  std::unique_ptr<MoqtFetchTask> fetch_task;
+  session_.Fetch(
+      FullTrackName("foo", "bar"),
+      [&](std::unique_ptr<MoqtFetchTask> task) {
+        fetch_task = std::move(task);
+      },
+      FullSequence(0, 0), 4, std::nullopt, 128, std::nullopt,
+      MoqtSubscribeParameters());
+  MoqtFetchOk ok = {
+      /*subscribe_id=*/0,
+      /*group_order=*/MoqtDeliveryOrder::kAscending,
+      /*largest_id=*/FullSequence(3, 25),
+      MoqtSubscribeParameters(),
+  };
+  stream_input->OnFetchOkMessage(ok);
+  ASSERT_NE(fetch_task, nullptr);
+  EXPECT_EQ(fetch_task->GetLargestId(), FullSequence(3, 25));
+  EXPECT_TRUE(fetch_task->GetStatus().ok());
+  PublishedObject object;
+  EXPECT_EQ(fetch_task->GetNextObject(object),
+            MoqtFetchTask::GetNextObjectResult::kPending);
+  // Cancel the fetch.
+  EXPECT_CALL(mock_stream,
+              Writev(ControlMessageOfType(MoqtMessageType::kFetchCancel), _));
+  fetch_task.reset();
+}
+
+TEST_F(MoqtSessionTest, FetchThenError) {
+  webtransport::test::MockStream mock_stream;
+  std::unique_ptr<MoqtControlParserVisitor> stream_input =
+      MoqtSessionPeer::CreateControlStream(&session_, &mock_stream);
+  std::unique_ptr<MoqtFetchTask> fetch_task;
+  session_.Fetch(
+      FullTrackName("foo", "bar"),
+      [&](std::unique_ptr<MoqtFetchTask> task) {
+        fetch_task = std::move(task);
+      },
+      FullSequence(0, 0), 4, std::nullopt, 128, std::nullopt,
+      MoqtSubscribeParameters());
+  MoqtFetchError error = {
+      /*subscribe_id=*/0,
+      /*error_code=*/SubscribeErrorCode::kUnauthorized,
+      /*reason_phrase=*/"No username provided",
+  };
+  stream_input->OnFetchErrorMessage(error);
+  ASSERT_NE(fetch_task, nullptr);
+  EXPECT_TRUE(absl::IsUnauthenticated(fetch_task->GetStatus()));
+  EXPECT_EQ(fetch_task->GetStatus().message(), "No username provided");
+}
+
 // TODO: re-enable this test once this behavior is re-implemented.
 #if 0
 TEST_F(MoqtSessionTest, SubscribeUpdateClosesSubscription) {
diff --git a/quiche/quic/moqt/moqt_track.cc b/quiche/quic/moqt/moqt_track.cc
index 07f44d7..31980d1 100644
--- a/quiche/quic/moqt/moqt_track.cc
+++ b/quiche/quic/moqt/moqt_track.cc
@@ -4,7 +4,16 @@
 
 #include "quiche/quic/moqt/moqt_track.h"
 
+#include <memory>
+#include <optional>
+#include <utility>
+
+#include "absl/status/status.h"
+#include "absl/strings/string_view.h"
 #include "quiche/quic/moqt/moqt_messages.h"
+#include "quiche/quic/moqt/moqt_publisher.h"
+#include "quiche/common/platform/api/quiche_logging.h"
+#include "quiche/web_transport/web_transport.h"
 
 namespace moqt {
 
@@ -16,4 +25,69 @@
   return true;
 }
 
+UpstreamFetch::~UpstreamFetch() {
+  if (task_.IsValid()) {
+    // Notify the task (which the application owns) that nothing more is coming.
+    // If this has already been called, UpstreamFetchTask will ignore it.
+    task_.GetIfAvailable()->OnStreamAndFetchClosed(kResetCodeUnknown, "");
+  }
+}
+
+void UpstreamFetch::OnFetchResult(FullSequence largest_id, absl::Status status,
+                                  TaskDestroyedCallback callback) {
+  auto task = std::make_unique<UpstreamFetchTask>(largest_id, status,
+                                                  std::move(callback));
+  task_ = task->weak_ptr();
+  std::move(ok_callback_)(std::move(task));
+}
+
+UpstreamFetch::UpstreamFetchTask::~UpstreamFetchTask() {
+  if (task_destroyed_callback_) {
+    std::move(task_destroyed_callback_)();
+  }
+}
+
+MoqtFetchTask::GetNextObjectResult
+UpstreamFetch::UpstreamFetchTask::GetNextObject(PublishedObject& output) {
+  if (!next_object_.has_value()) {
+    if (!status_.ok()) {
+      return kError;
+    }
+    if (eof_) {
+      return kEof;
+    }
+    return kPending;
+  }
+  output = *std::move(next_object_);
+  next_object_.reset();
+  return kSuccess;
+}
+
+void UpstreamFetch::UpstreamFetchTask::NewObject(PublishedObject& object) {
+  QUICHE_DCHECK(!next_object_.has_value());
+  next_object_ = std::move(object);
+  if (object_available_callback_) {
+    object_available_callback_();
+  }
+}
+
+void UpstreamFetch::UpstreamFetchTask::OnStreamAndFetchClosed(
+    std::optional<webtransport::StreamErrorCode> error,
+    absl::string_view reason_phrase) {
+  if (eof_ || error.has_value()) {
+    return;
+  }
+  // Delete callbacks, because IncomingDataStream and UpstreamFetch are gone.
+  can_read_callback_ = nullptr;
+  task_destroyed_callback_ = nullptr;
+  if (!error.has_value()) {  // This was a FIN.
+    eof_ = true;
+  } else {
+    status_ = MoqtStreamErrorToStatus(*error, reason_phrase);
+  }
+  if (object_available_callback_) {
+    object_available_callback_();
+  }
+}
+
 }  // namespace moqt
diff --git a/quiche/quic/moqt/moqt_track.h b/quiche/quic/moqt/moqt_track.h
index e47464a..fe158e1 100644
--- a/quiche/quic/moqt/moqt_track.h
+++ b/quiche/quic/moqt/moqt_track.h
@@ -8,14 +8,18 @@
 #include <cstdint>
 #include <memory>
 #include <optional>
+#include <utility>
 
+#include "absl/status/status.h"
 #include "absl/strings/string_view.h"
 #include "quiche/quic/core/quic_time.h"
 #include "quiche/quic/moqt/moqt_messages.h"
 #include "quiche/quic/moqt/moqt_priority.h"
+#include "quiche/quic/moqt/moqt_publisher.h"
 #include "quiche/quic/moqt/moqt_subscribe_windows.h"
 #include "quiche/common/quiche_callbacks.h"
 #include "quiche/common/quiche_weak_ptr.h"
+#include "quiche/web_transport/web_transport.h"
 
 namespace moqt {
 
@@ -133,6 +137,106 @@
   std::unique_ptr<MoqtSubscribe> subscribe_;
 };
 
+// MoqtSession calls this when a FETCH_OK or FETCH_ERROR is received. The
+// destination of the callback owns |fetch_task| and MoqtSession will react
+// safely if the owner destroys it.
+using FetchResponseCallback =
+    quiche::SingleUseCallback<void(std::unique_ptr<MoqtFetchTask> fetch_task)>;
+
+// This is a callback to MoqtSession::IncomingDataStream. Called when the
+// FetchTask has its object cache empty, on creation, and whenever the
+// application reads it.
+using CanReadCallback = quiche::MultiUseCallback<void()>;
+
+// If the application destroys the FetchTask, this is a signal to MoqtSession to
+// cancel the FETCH and STOP_SENDING the stream.
+using TaskDestroyedCallback = quiche::SingleUseCallback<void()>;
+
+// Class for upstream FETCH. It will notify the application using |callback|
+// when a FETCH_OK or FETCH_ERROR is received.
+class UpstreamFetch : public RemoteTrack {
+ public:
+  UpstreamFetch(const MoqtFetch& fetch, FetchResponseCallback callback)
+      : RemoteTrack(fetch.full_track_name, fetch.subscribe_id,
+                    SubscribeWindow(
+                        fetch.start_object,
+                        FullSequence(fetch.end_group,
+                                     fetch.end_object.value_or(UINT64_MAX)))),
+        ok_callback_(std::move(callback)) {
+    // Immediately set the data stream type.
+    CheckDataStreamType(MoqtDataStreamType::kStreamHeaderFetch);
+  }
+  UpstreamFetch(const UpstreamFetch&) = delete;
+  ~UpstreamFetch();
+
+  class UpstreamFetchTask : public MoqtFetchTask {
+   public:
+    // If the UpstreamFetch is destroyed, it will call OnStreamAndFetchClosed
+    // which sets the TaskDestroyedCallback to nullptr. Thus, |callback| can
+    // assume that UpstreamFetch is valid.
+    UpstreamFetchTask(FullSequence largest_id, absl::Status status,
+                      TaskDestroyedCallback callback)
+        : largest_id_(largest_id),
+          status_(status),
+          task_destroyed_callback_(std::move(callback)),
+          weak_ptr_factory_(this) {}
+    ~UpstreamFetchTask() override;
+
+    // Implementation of MoqtFetchTask.
+    GetNextObjectResult GetNextObject(PublishedObject& output) override;
+    void SetObjectAvailableCallback(
+        ObjectsAvailableCallback callback) override {
+      object_available_callback_ = std::move(callback);
+    };
+    absl::Status GetStatus() override { return status_; };
+    FullSequence GetLargestId() const override { return largest_id_; }
+
+    quiche::QuicheWeakPtr<UpstreamFetchTask> weak_ptr() {
+      return weak_ptr_factory_.Create();
+    }
+
+    // Manage the relationship with the data stream.
+    void OnStreamOpened(CanReadCallback callback) {
+      can_read_callback_ = std::move(callback);
+    }
+
+    // Called when the data stream receives a new object.
+    void NewObject(PublishedObject& object);
+
+    // Deletes callbacks to session or stream, updates the status. If |error|
+    // has no value, will append an EOF to the object stream.
+    void OnStreamAndFetchClosed(
+        std::optional<webtransport::StreamErrorCode> error,
+        absl::string_view reason_phrase);
+
+   private:
+    FullSequence largest_id_;
+    absl::Status status_;
+    TaskDestroyedCallback task_destroyed_callback_;
+
+    // Object delivery state.
+    std::optional<PublishedObject> next_object_;
+    bool eof_ = false;  // The next object is EOF.
+    ObjectsAvailableCallback object_available_callback_;
+    CanReadCallback can_read_callback_;
+
+    // Must be last.
+    quiche::QuicheWeakPtrFactory<UpstreamFetchTask> weak_ptr_factory_;
+  };
+
+  // Arrival of FETCH_OK/FETCH_ERROR.
+  void OnFetchResult(FullSequence largest_id, absl::Status status,
+                     TaskDestroyedCallback callback);
+
+  UpstreamFetchTask* task() { return task_.GetIfAvailable(); }
+
+ private:
+  quiche::QuicheWeakPtr<UpstreamFetchTask> task_;
+
+  // Initial values from Fetch() call.
+  FetchResponseCallback ok_callback_;  // Will be destroyed on FETCH_OK.
+};
+
 }  // namespace moqt
 
 #endif  // QUICHE_QUIC_MOQT_MOQT_SUBSCRIPTION_H_
diff --git a/quiche/quic/moqt/moqt_track_test.cc b/quiche/quic/moqt/moqt_track_test.cc
index a5717ce..8072535 100644
--- a/quiche/quic/moqt/moqt_track_test.cc
+++ b/quiche/quic/moqt/moqt_track_test.cc
@@ -4,11 +4,16 @@
 
 #include "quiche/quic/moqt/moqt_track.h"
 
+#include <memory>
 #include <optional>
+#include <utility>
 
+#include "absl/status/status.h"
 #include "quiche/quic/moqt/moqt_messages.h"
+#include "quiche/quic/moqt/moqt_publisher.h"
 #include "quiche/quic/moqt/tools/moqt_mock_visitor.h"
 #include "quiche/quic/platform/api/quic_test.h"
+#include "quiche/common/platform/api/quiche_mem_slice.h"
 
 namespace moqt {
 
@@ -64,7 +69,96 @@
   EXPECT_FALSE(track_.InWindow(FullSequence(2, 0)));
 }
 
-// TODO: Write test for GetStreamForSequence.
+class UpstreamFetchTest : public quic::test::QuicTest {
+ protected:
+  UpstreamFetchTest()
+      : fetch_(fetch_message_, [&](std::unique_ptr<MoqtFetchTask> task) {
+          fetch_task_ = std::move(task);
+        }) {}
+
+  MoqtFetch fetch_message_ = {
+      /*fetch_id=*/1,
+      /*full_track_name=*/FullTrackName("foo", "bar"),
+      /*subscriber_priority=*/128,
+      /*group_order=*/std::nullopt,
+      /*start_object=*/FullSequence(1, 1),
+      /*end_group=*/3,
+      /*end_object=*/100,
+      /*parameters=*/MoqtSubscribeParameters(),
+  };
+  // The pointer held by the application.
+  UpstreamFetch fetch_;
+  std::unique_ptr<MoqtFetchTask> fetch_task_;
+};
+
+TEST_F(UpstreamFetchTest, Queries) {
+  EXPECT_EQ(fetch_.subscribe_id(), 1);
+  EXPECT_EQ(fetch_.full_track_name(), FullTrackName("foo", "bar"));
+  EXPECT_FALSE(
+      fetch_.CheckDataStreamType(MoqtDataStreamType::kStreamHeaderSubgroup));
+  EXPECT_TRUE(
+      fetch_.CheckDataStreamType(MoqtDataStreamType::kStreamHeaderFetch));
+  EXPECT_TRUE(fetch_.is_fetch());
+  EXPECT_FALSE(fetch_.InWindow(FullSequence{1, 0}));
+  EXPECT_TRUE(fetch_.InWindow(FullSequence{1, 1}));
+  EXPECT_TRUE(fetch_.InWindow(FullSequence{3, 100}));
+  EXPECT_FALSE(fetch_.InWindow(FullSequence{3, 101}));
+}
+
+TEST_F(UpstreamFetchTest, AllowError) {
+  EXPECT_TRUE(fetch_.ErrorIsAllowed());
+  fetch_.OnObjectOrOk();
+  EXPECT_FALSE(fetch_.ErrorIsAllowed());
+}
+
+TEST_F(UpstreamFetchTest, FetchResponse) {
+  EXPECT_EQ(fetch_task_, nullptr);
+  fetch_.OnFetchResult(FullSequence(3, 50), absl::OkStatus(), nullptr);
+  EXPECT_NE(fetch_task_, nullptr);
+  EXPECT_NE(fetch_.task(), nullptr);
+  EXPECT_TRUE(fetch_task_->GetStatus().ok());
+  EXPECT_EQ(fetch_task_->GetLargestId(), FullSequence(3, 50));
+}
+
+TEST_F(UpstreamFetchTest, FetchClosedByMoqt) {
+  bool terminated = false;
+  fetch_.OnFetchResult(FullSequence(3, 50), absl::OkStatus(),
+                       [&]() { terminated = true; });
+  bool got_eof = false;
+  fetch_task_->SetObjectAvailableCallback([&]() {
+    PublishedObject object;
+    EXPECT_EQ(fetch_task_->GetNextObject(object),
+              MoqtFetchTask::GetNextObjectResult::kEof);
+    got_eof = true;
+  });
+  fetch_.task()->OnStreamAndFetchClosed(std::nullopt, "");
+  EXPECT_FALSE(terminated);
+  EXPECT_TRUE(got_eof);
+}
+
+TEST_F(UpstreamFetchTest, FetchClosedByApplication) {
+  bool terminated = false;
+  fetch_.OnFetchResult(FullSequence(3, 50), absl::Status(),
+                       [&]() { terminated = true; });
+  fetch_task_.reset();
+  EXPECT_TRUE(terminated);
+}
+
+TEST_F(UpstreamFetchTest, ObjectRetrieval) {
+  fetch_.OnFetchResult(FullSequence(3, 50), absl::OkStatus(), nullptr);
+  PublishedObject object;
+  EXPECT_EQ(fetch_task_->GetNextObject(object),
+            MoqtFetchTask::GetNextObjectResult::kPending);
+  PublishedObject new_object(FullSequence(3, 0),
+                             MoqtObjectStatus::kGroupDoesNotExist, 128,
+                             quiche::QuicheMemSlice(), false);
+  fetch_task_->SetObjectAvailableCallback([&]() {
+    EXPECT_EQ(fetch_task_->GetNextObject(object),
+              MoqtFetchTask::GetNextObjectResult::kSuccess);
+    EXPECT_EQ(object.sequence, new_object.sequence);
+  });
+  fetch_.task()->NewObject(new_object);
+}
 
 }  // namespace test
 
diff --git a/quiche/quic/moqt/test_tools/moqt_session_peer.h b/quiche/quic/moqt/test_tools/moqt_session_peer.h
index 5426ef6..e867376 100644
--- a/quiche/quic/moqt/test_tools/moqt_session_peer.h
+++ b/quiche/quic/moqt/test_tools/moqt_session_peer.h
@@ -17,7 +17,6 @@
 #include "quiche/quic/moqt/moqt_session.h"
 #include "quiche/quic/moqt/moqt_track.h"
 #include "quiche/quic/moqt/tools/moqt_mock_visitor.h"
-#include "quiche/quic/platform/api/quic_test.h"
 #include "quiche/web_transport/test_tools/mock_web_transport.h"
 #include "quiche/web_transport/web_transport.h"
 
@@ -75,11 +74,12 @@
                                 const MoqtSubscribe& subscribe,
                                 SubscribeRemoteTrack::Visitor* visitor) {
     auto track = std::make_unique<SubscribeRemoteTrack>(subscribe, visitor);
-    session->upstream_by_id_.try_emplace(subscribe.subscribe_id, track.get());
-    session->upstream_by_name_.try_emplace(subscribe.full_track_name,
-                                           track.get());
     session->subscribe_by_alias_.try_emplace(subscribe.track_alias,
-                                             std::move(track));
+                                             track.get());
+    session->subscribe_by_name_.try_emplace(subscribe.full_track_name,
+                                            track.get());
+    session->upstream_by_id_.try_emplace(subscribe.subscribe_id,
+                                         std::move(track));
   }
 
   static MoqtObjectListener* AddSubscription(