diff --git a/quiche/quic/moqt/moqt_framer.cc b/quiche/quic/moqt/moqt_framer.cc
index d2f57fc..21588c9 100644
--- a/quiche/quic/moqt/moqt_framer.cc
+++ b/quiche/quic/moqt/moqt_framer.cc
@@ -193,6 +193,8 @@
   return WireUint8(0xff);
 }
 
+WireUint8 WireBoolean(bool value) { return WireUint8(value ? 0x01 : 0x00); }
+
 uint64_t SignedVarintSerializedForm(int64_t value) {
   if (value < 0) {
     return ((-value) << 1) | 0x01;
@@ -416,11 +418,6 @@
 
 quiche::QuicheBuffer MoqtFramer::SerializeSubscribe(
     const MoqtSubscribe& message) {
-  MoqtFilterType filter_type = GetFilterType(message);
-  if (filter_type == MoqtFilterType::kNone) {
-    QUICHE_BUG(MoqtFramer_invalid_subscribe) << "Invalid object range";
-    return quiche::QuicheBuffer();
-  }
   KeyValuePairList parameters;
   VersionSpecificParametersToKeyValuePairList(message.parameters, parameters);
   if (!ValidateVersionSpecificParameters(parameters,
@@ -429,33 +426,44 @@
         << "Serializing invalid MoQT parameters";
     return quiche::QuicheBuffer();
   }
-  switch (filter_type) {
+  switch (message.filter_type) {
+    case MoqtFilterType::kNextGroupStart:
     case MoqtFilterType::kLatestObject:
       return SerializeControlMessage(
-          MoqtMessageType::kSubscribe, WireVarInt62(message.subscribe_id),
+          MoqtMessageType::kSubscribe, WireVarInt62(message.request_id),
           WireVarInt62(message.track_alias),
           WireFullTrackName(message.full_track_name, true),
           WireUint8(message.subscriber_priority),
-          WireDeliveryOrder(message.group_order), WireVarInt62(filter_type),
-          WireKeyValuePairList(parameters));
+          WireDeliveryOrder(message.group_order), WireBoolean(message.forward),
+          WireVarInt62(message.filter_type), WireKeyValuePairList(parameters));
     case MoqtFilterType::kAbsoluteStart:
+      if (!message.start.has_value()) {
+        return quiche::QuicheBuffer();
+      };
       return SerializeControlMessage(
-          MoqtMessageType::kSubscribe, WireVarInt62(message.subscribe_id),
+          MoqtMessageType::kSubscribe, WireVarInt62(message.request_id),
           WireVarInt62(message.track_alias),
           WireFullTrackName(message.full_track_name, true),
           WireUint8(message.subscriber_priority),
-          WireDeliveryOrder(message.group_order), WireVarInt62(filter_type),
-          WireVarInt62(message.start->group),
+          WireDeliveryOrder(message.group_order), WireBoolean(message.forward),
+          WireVarInt62(message.filter_type), WireVarInt62(message.start->group),
           WireVarInt62(message.start->object),
           WireKeyValuePairList(parameters));
     case MoqtFilterType::kAbsoluteRange:
+      if (!message.start.has_value() || !message.end_group.has_value()) {
+        return quiche::QuicheBuffer();
+      }
+      if (*message.end_group < message.start->group) {
+        QUICHE_BUG(MoqtFramer_invalid_end_group) << "Invalid object range";
+        return quiche::QuicheBuffer();
+      }
       return SerializeControlMessage(
-          MoqtMessageType::kSubscribe, WireVarInt62(message.subscribe_id),
+          MoqtMessageType::kSubscribe, WireVarInt62(message.request_id),
           WireVarInt62(message.track_alias),
           WireFullTrackName(message.full_track_name, true),
           WireUint8(message.subscriber_priority),
-          WireDeliveryOrder(message.group_order), WireVarInt62(filter_type),
-          WireVarInt62(message.start->group),
+          WireDeliveryOrder(message.group_order), WireBoolean(message.forward),
+          WireVarInt62(message.filter_type), WireVarInt62(message.start->group),
           WireVarInt62(message.start->object), WireVarInt62(*message.end_group),
           WireKeyValuePairList(parameters));
     default:
diff --git a/quiche/quic/moqt/moqt_framer_test.cc b/quiche/quic/moqt/moqt_framer_test.cc
index e772bc8..16939e3 100644
--- a/quiche/quic/moqt/moqt_framer_test.cc
+++ b/quiche/quic/moqt/moqt_framer_test.cc
@@ -402,35 +402,24 @@
 }
 
 TEST_F(MoqtFramerSimpleTest, AllSubscribeInputs) {
-  for (std::optional<Location> start :
-       {std::optional<Location>(),
-        std::optional<Location>(std::in_place, 4, 0)}) {
-    for (std::optional<uint64_t> end_group :
-         {std::optional<uint64_t>(), std::optional<uint64_t>(7)}) {
-      MoqtSubscribe subscribe = {
-          /*subscribe_id=*/3,
-          /*track_alias=*/4,
-          /*full_track_name=*/FullTrackName({"foo", "abcd"}),
-          /*subscriber_priority=*/0x20,
-          /*group_order=*/std::nullopt,
-          start,
-          end_group,
-          VersionSpecificParameters(AuthTokenType::kOutOfBand, "bar"),
-      };
-      quiche::QuicheBuffer buffer;
-      MoqtFilterType expected_filter_type = GetFilterType(subscribe);
-      if (expected_filter_type == MoqtFilterType::kNone) {
-        EXPECT_QUIC_BUG(buffer = framer_.SerializeSubscribe(subscribe),
-                        "Invalid object range");
-        EXPECT_EQ(buffer.size(), 0);
-        continue;
-      }
-      buffer = framer_.SerializeSubscribe(subscribe);
-      // Go to the filter type.
-      const uint8_t* read = BufferAtOffset(buffer, 17);
-      EXPECT_EQ(static_cast<MoqtFilterType>(*read), expected_filter_type);
-      EXPECT_GT(buffer.size(), 0);
-    }
+  for (auto filter :
+       {MoqtFilterType::kNextGroupStart, MoqtFilterType::kLatestObject,
+        MoqtFilterType::kAbsoluteStart, MoqtFilterType::kAbsoluteRange}) {
+    MoqtSubscribe subscribe = {
+        /*subscribe_id=*/3,
+        /*track_alias=*/4,
+        /*full_track_name=*/FullTrackName({"foo", "abcd"}),
+        /*subscriber_priority=*/0x20,
+        /*group_order=*/std::nullopt,
+        /*forward=*/true,
+        /*filter_type=*/filter,
+        /*start=*/std::make_optional<Location>(4, 1),
+        /*end_group=*/std::make_optional<uint64_t>(6ULL),
+        VersionSpecificParameters(AuthTokenType::kOutOfBand, "bar"),
+    };
+    quiche::QuicheBuffer buffer;
+    buffer = framer_.SerializeSubscribe(subscribe);
+    EXPECT_GT(buffer.size(), 0);
   }
 }
 
@@ -441,13 +430,15 @@
       /*full_track_name=*/FullTrackName({"foo", "abcd"}),
       /*subscriber_priority=*/0x20,
       /*group_order=*/std::nullopt,
-      /*start=*/Location(4, 3),
-      /*end_group=*/3,
+      /*forward=*/true,
+      /*filter_type=*/MoqtFilterType::kAbsoluteRange,
+      /*start=*/std::make_optional<Location>(4, 3),
+      /*end_group=*/std::make_optional<uint64_t>(3ULL),
       VersionSpecificParameters(AuthTokenType::kOutOfBand, "bar"),
   };
   quiche::QuicheBuffer buffer;
-  EXPECT_QUIC_BUG(buffer = framer_.SerializeSubscribe(subscribe),
-                  "Invalid object range");
+  EXPECT_QUICHE_BUG(buffer = framer_.SerializeSubscribe(subscribe),
+                    "Invalid object range");
   EXPECT_EQ(buffer.size(), 0);
 }
 
diff --git a/quiche/quic/moqt/moqt_messages.cc b/quiche/quic/moqt/moqt_messages.cc
index 904954c..7e6f480 100644
--- a/quiche/quic/moqt/moqt_messages.cc
+++ b/quiche/quic/moqt/moqt_messages.cc
@@ -111,22 +111,6 @@
   }
 }
 
-MoqtFilterType GetFilterType(const MoqtSubscribe& message) {
-  if (message.start.has_value()) {
-    if (message.end_group.has_value()) {
-      if (*message.end_group < message.start->group) {
-        return MoqtFilterType::kNone;
-      }
-      return MoqtFilterType::kAbsoluteRange;
-    }
-    return MoqtFilterType::kAbsoluteStart;
-  }
-  if (message.end_group.has_value()) {
-    return MoqtFilterType::kNone;  // End group without start is invalid.
-  }
-  return MoqtFilterType::kLatestObject;
-}
-
 MoqtError ValidateSetupParameters(const KeyValuePairList& parameters,
                                   bool webtrans,
                                   quic::Perspective perspective) {
diff --git a/quiche/quic/moqt/moqt_messages.h b/quiche/quic/moqt/moqt_messages.h
index ac22147..9ba07f8 100644
--- a/quiche/quic/moqt/moqt_messages.h
+++ b/quiche/quic/moqt/moqt_messages.h
@@ -518,33 +518,25 @@
 
 enum class QUICHE_EXPORT MoqtFilterType : uint64_t {
   kNone = 0x0,
+  kNextGroupStart = 0x1,
   kLatestObject = 0x2,
   kAbsoluteStart = 0x3,
   kAbsoluteRange = 0x4,
 };
 
 struct QUICHE_EXPORT MoqtSubscribe {
-  uint64_t subscribe_id;
+  uint64_t request_id;
   uint64_t track_alias;
   FullTrackName full_track_name;
   MoqtPriority subscriber_priority;
   std::optional<MoqtDeliveryOrder> group_order;
-
-  // The combinations of these that have values indicate the filter type.
-  // (none): KLatestObject
-  // start: kAbsoluteStart
-  // start, end_group: kAbsoluteRange (request whole last group)
-  // All other combinations are invalid.
+  bool forward;
+  MoqtFilterType filter_type;
   std::optional<Location> start;
   std::optional<uint64_t> end_group;
-  // If the mode is kNone, the these are std::nullopt.
   VersionSpecificParameters parameters;
 };
 
-// Deduce the filter type from the combination of group and object IDs. Returns
-// kNone if the state of the subscribe is invalid.
-MoqtFilterType GetFilterType(const MoqtSubscribe& message);
-
 struct QUICHE_EXPORT MoqtSubscribeOk {
   uint64_t subscribe_id;
   // The message uses ms, but expires is in us.
diff --git a/quiche/quic/moqt/moqt_parser.cc b/quiche/quic/moqt/moqt_parser.cc
index 5dd2508..e9a093b 100644
--- a/quiche/quic/moqt/moqt_parser.cc
+++ b/quiche/quic/moqt/moqt_parser.cc
@@ -402,23 +402,30 @@
 size_t MoqtControlParser::ProcessSubscribe(quic::QuicDataReader& reader) {
   MoqtSubscribe subscribe;
   uint64_t filter, group, object;
-  uint8_t group_order;
+  uint8_t group_order, forward;
   absl::string_view track_name;
-  if (!reader.ReadVarInt62(&subscribe.subscribe_id) ||
+  if (!reader.ReadVarInt62(&subscribe.request_id) ||
       !reader.ReadVarInt62(&subscribe.track_alias) ||
       !ReadTrackNamespace(reader, subscribe.full_track_name) ||
       !reader.ReadStringPieceVarInt62(&track_name) ||
       !reader.ReadUInt8(&subscribe.subscriber_priority) ||
-      !reader.ReadUInt8(&group_order) || !reader.ReadVarInt62(&filter)) {
+      !reader.ReadUInt8(&group_order) || !reader.ReadUInt8(&forward) ||
+      !reader.ReadVarInt62(&filter)) {
     return 0;
   }
   subscribe.full_track_name.AddElement(track_name);
   if (!ParseDeliveryOrder(group_order, subscribe.group_order)) {
-    ParseError("Invalid group order value in SUBSCRIBE message");
+    ParseError("Invalid group order value in SUBSCRIBE");
     return 0;
   }
-  MoqtFilterType filter_type = static_cast<MoqtFilterType>(filter);
-  switch (filter_type) {
+  if (forward > 1) {
+    ParseError("Invalid forward value in SUBSCRIBE");
+    return 0;
+  }
+  subscribe.forward = (forward == 1);
+  subscribe.filter_type = static_cast<MoqtFilterType>(filter);
+  switch (subscribe.filter_type) {
+    case MoqtFilterType::kNextGroupStart:
     case MoqtFilterType::kLatestObject:
       break;
     case MoqtFilterType::kAbsoluteStart:
@@ -427,7 +434,7 @@
         return 0;
       }
       subscribe.start = Location(group, object);
-      if (filter_type == MoqtFilterType::kAbsoluteStart) {
+      if (subscribe.filter_type == MoqtFilterType::kAbsoluteStart) {
         break;
       }
       if (!reader.ReadVarInt62(&group)) {
diff --git a/quiche/quic/moqt/moqt_parser_test.cc b/quiche/quic/moqt/moqt_parser_test.cc
index 7e4ba28..4779c89 100644
--- a/quiche/quic/moqt/moqt_parser_test.cc
+++ b/quiche/quic/moqt/moqt_parser_test.cc
@@ -648,8 +648,8 @@
       0x20, 0x00, 0x0d, 0x02, 0x01, 0x02,  // versions = 1, 2
       0x03,                                // 4 params
       0x01, 0x03, 0x66, 0x6f, 0x6f,        // path = "foo"
-      0x02, 0x32,                          // max_subscribe_id = 50
-      0x02, 0x32,                          // max_subscribe_id = 50
+      0x02, 0x32,                          // max_request_id = 50
+      0x02, 0x32,                          // max_request_id = 50
   };
   stream.Receive(absl::string_view(setup, sizeof(setup)), false);
   parser.ReadAndDispatchMessages();
@@ -663,10 +663,10 @@
   webtransport::test::InMemoryStream stream(/*stream_id=*/0);
   MoqtControlParser parser(kWebTrans, &stream, visitor_);
   char subscribe[] = {
-      0x03, 0x00, 0x1a, 0x01, 0x02,
+      0x03, 0x00, 0x1b, 0x01, 0x02,
       0x01, 0x03, 0x66, 0x6f, 0x6f,  // track_namespace = "foo"
       0x04, 0x61, 0x62, 0x63, 0x64,  // track_name = "abcd"
-      0x20, 0x02,                    // priority = 0x20 descending
+      0x20, 0x02, 0x01,              // priority, order, forward
       0x02,                          // filter_type = kLatestObject
       0x02,                          // two params
       0x1f, 0x03, 0x62, 0x61, 0x72,  // 0x1f = "bar"
@@ -681,10 +681,10 @@
   webtransport::test::InMemoryStream stream(/*stream_id=*/0);
   MoqtControlParser parser(kRawQuic, &stream, visitor_);
   char subscribe[] = {
-      0x03, 0x00, 0x16, 0x01, 0x02,
+      0x03, 0x00, 0x17, 0x01, 0x02,
       0x01, 0x03, 0x66, 0x6f, 0x6f,  // track_namespace = "foo"
       0x04, 0x61, 0x62, 0x63, 0x64,  // track_name = "abcd"
-      0x20, 0x02,                    // priority = 0x20 descending
+      0x20, 0x02, 0x01,              // priority, order, forward
       0x02,                          // filter_type = kLatestObject
       0x02,                          // two params
       0x02, 0x67, 0x10,              // delivery_timeout = 10000
@@ -701,10 +701,10 @@
   webtransport::test::InMemoryStream stream(/*stream_id=*/0);
   MoqtControlParser parser(kRawQuic, &stream, visitor_);
   char subscribe[] = {
-      0x03, 0x00, 0x16, 0x01, 0x02,
+      0x03, 0x00, 0x17, 0x01, 0x02,
       0x01, 0x03, 0x66, 0x6f, 0x6f,  // track_namespace = "foo"
       0x04, 0x61, 0x62, 0x63, 0x64,  // track_name = "abcd"
-      0x20, 0x02,                    // priority = 0x20 descending
+      0x20, 0x02, 0x01,              // priority, order, forward
       0x02,                          // filter_type = kLatestObject
       0x02,                          // two params
       0x04, 0x67, 0x10,              // max_cache_duration = 10000
@@ -721,10 +721,10 @@
   webtransport::test::InMemoryStream stream(/*stream_id=*/0);
   MoqtControlParser parser(kRawQuic, &stream, visitor_);
   char subscribe[] = {
-      0x03, 0x00, 0x14, 0x01, 0x02,
+      0x03, 0x00, 0x15, 0x01, 0x02,
       0x01, 0x03, 0x66, 0x6f, 0x6f,  // track_namespace = "foo"
       0x04, 0x61, 0x62, 0x63, 0x64,  // track_name = "abcd"
-      0x20, 0x02,                    // priority = 0x20 descending
+      0x20, 0x02, 0x01,              // priority, order, forward
       0x02,                          // filter_type = kLatestObject
       0x01,                          // one param
       0x01, 0x02, 0x00, 0x00,        // authorization_token = DELETE 0;
@@ -740,10 +740,10 @@
   webtransport::test::InMemoryStream stream(/*stream_id=*/0);
   MoqtControlParser parser(kRawQuic, &stream, visitor_);
   char subscribe[] = {
-      0x03, 0x00, 0x18, 0x01, 0x02, 0x01, 0x03, 0x66, 0x6f,
-      0x6f,                          // track_namespace = "foo"
+      0x03, 0x00, 0x19, 0x01, 0x02, 0x01, 0x03, 0x66,
+      0x6f, 0x6f,                    // track_namespace = "foo"
       0x04, 0x61, 0x62, 0x63, 0x64,  // track_name = "abcd"
-      0x20, 0x02,                    // priority = 0x20 descending
+      0x20, 0x02, 0x01,              // priority, order, forward
       0x02,                          // filter_type = kLatestObject
       0x01,                          // one param
       0x01, 0x06, 0x01, 0x10, 0x00, 0x62, 0x61, 0x72,  // REGISTER 0x01
@@ -759,10 +759,10 @@
   webtransport::test::InMemoryStream stream(/*stream_id=*/0);
   MoqtControlParser parser(kRawQuic, &stream, visitor_);
   char subscribe[] = {
-      0x03, 0x00, 0x14, 0x01, 0x02,
+      0x03, 0x00, 0x15, 0x01, 0x02,
       0x01, 0x03, 0x66, 0x6f, 0x6f,  // track_namespace = "foo"
       0x04, 0x61, 0x62, 0x63, 0x64,  // track_name = "abcd"
-      0x20, 0x02,                    // priority = 0x20 descending
+      0x20, 0x02, 0x01,              // priority, order, forward
       0x02,                          // filter_type = kLatestObject
       0x01,                          // one param
       0x01, 0x02, 0x02, 0x07,        // authorization_token = USE 7;
@@ -779,10 +779,10 @@
   webtransport::test::InMemoryStream stream(/*stream_id=*/0);
   MoqtControlParser parser(kRawQuic, &stream, visitor_);
   char subscribe[] = {
-      0x03, 0x00, 0x14, 0x01, 0x02,
+      0x03, 0x00, 0x15, 0x01, 0x02,
       0x01, 0x03, 0x66, 0x6f, 0x6f,  // track_namespace = "foo"
       0x04, 0x61, 0x62, 0x63, 0x64,  // track_name = "abcd"
-      0x20, 0x02,                    // priority = 0x20 descending
+      0x20, 0x02, 0x01,              // priority, order, forward
       0x02,                          // filter_type = kLatestObject
       0x01,                          // one param
       0x01, 0x02, 0x04, 0x07,        // authorization_token type 4
@@ -799,10 +799,10 @@
   webtransport::test::InMemoryStream stream(/*stream_id=*/0);
   MoqtControlParser parser(kRawQuic, &stream, visitor_);
   char subscribe[] = {
-      0x03, 0x00, 0x16, 0x01, 0x02, 0x01,
+      0x03, 0x00, 0x17, 0x01, 0x02, 0x01,
       0x03, 0x66, 0x6f, 0x6f,             // track_namespace = "foo"
       0x04, 0x61, 0x62, 0x63, 0x64,       // track_name = "abcd"
-      0x20, 0x02,                         // priority = 0x20 descending
+      0x20, 0x02, 0x01,                   // priority, order, forward
       0x02,                               // filter_type = kLatestObject
       0x01,                               // one param
       0x01, 0x04, 0x03, 0x01, 0x00, 0x00  // authorization_token type 1
@@ -814,6 +814,81 @@
   EXPECT_EQ(visitor_.parsing_error_code_, MoqtError::kKeyValueFormattingError);
 }
 
+TEST_F(MoqtMessageSpecificTest, SubscribeInvalidGroupOrder) {
+  webtransport::test::InMemoryStream stream(/*stream_id=*/0);
+  MoqtControlParser parser(kRawQuic, &stream, visitor_);
+  char subscribe[] = {
+      0x03, 0x00, 0x1d, 0x01, 0x02,  // id and alias
+      0x01, 0x03, 0x66, 0x6f, 0x6f,  // track_namespace = "foo"
+      0x04, 0x61, 0x62, 0x63, 0x64,  // track_name = "abcd"
+      0x20,                          // subscriber priority = 0x20
+      0x03,                          // group order = invalid
+      0x01,                          // forward = true
+      0x03,                          // Filter type: Absolute Start
+      0x04,                          // start_group = 4 (relative previous)
+      0x01,                          // start_object = 1 (absolute)
+      // No EndGroup or EndObject
+      0x02,                                      // 2 parameters
+      0x02, 0x67, 0x10,                          // delivery_timeout = 10000 ms
+      0x01, 0x05, 0x03, 0x00, 0x62, 0x61, 0x72,  // authorization_tag = "bar"
+  };
+  stream.Receive(absl::string_view(subscribe, sizeof(subscribe)), false);
+  parser.ReadAndDispatchMessages();
+  EXPECT_EQ(visitor_.messages_received_, 0);
+  EXPECT_EQ(visitor_.parsing_error_, "Invalid group order value in SUBSCRIBE");
+  EXPECT_EQ(visitor_.parsing_error_code_, MoqtError::kProtocolViolation);
+}
+
+TEST_F(MoqtMessageSpecificTest, SubscribeInvalidForward) {
+  webtransport::test::InMemoryStream stream(/*stream_id=*/0);
+  MoqtControlParser parser(kRawQuic, &stream, visitor_);
+  char subscribe[] = {
+      0x03, 0x00, 0x1d, 0x01, 0x02,  // id and alias
+      0x01, 0x03, 0x66, 0x6f, 0x6f,  // track_namespace = "foo"
+      0x04, 0x61, 0x62, 0x63, 0x64,  // track_name = "abcd"
+      0x20,                          // subscriber priority = 0x20
+      0x02,                          // group order = descending
+      0x02,                          // forward = invalid
+      0x03,                          // Filter type: Absolute Start
+      0x04,                          // start_group = 4 (relative previous)
+      0x01,                          // start_object = 1 (absolute)
+      // No EndGroup or EndObject
+      0x02,                                      // 2 parameters
+      0x02, 0x67, 0x10,                          // delivery_timeout = 10000 ms
+      0x01, 0x05, 0x03, 0x00, 0x62, 0x61, 0x72,  // authorization_tag = "bar"
+  };
+  stream.Receive(absl::string_view(subscribe, sizeof(subscribe)), false);
+  parser.ReadAndDispatchMessages();
+  EXPECT_EQ(visitor_.messages_received_, 0);
+  EXPECT_EQ(visitor_.parsing_error_, "Invalid forward value in SUBSCRIBE");
+  EXPECT_EQ(visitor_.parsing_error_code_, MoqtError::kProtocolViolation);
+}
+
+TEST_F(MoqtMessageSpecificTest, SubscribeInvalidFilter) {
+  webtransport::test::InMemoryStream stream(/*stream_id=*/0);
+  MoqtControlParser parser(kRawQuic, &stream, visitor_);
+  char subscribe[] = {
+      0x03, 0x00, 0x1d, 0x01, 0x02,  // id and alias
+      0x01, 0x03, 0x66, 0x6f, 0x6f,  // track_namespace = "foo"
+      0x04, 0x61, 0x62, 0x63, 0x64,  // track_name = "abcd"
+      0x20,                          // subscriber priority = 0x20
+      0x02,                          // group order = descending
+      0x01,                          // forward = true
+      0x05,                          // Filter type: Absolute Start
+      0x04,                          // start_group = 4 (relative previous)
+      0x01,                          // start_object = 1 (absolute)
+      // No EndGroup or EndObject
+      0x02,                                      // 2 parameters
+      0x02, 0x67, 0x10,                          // delivery_timeout = 10000 ms
+      0x01, 0x05, 0x03, 0x00, 0x62, 0x61, 0x72,  // authorization_tag = "bar"
+  };
+  stream.Receive(absl::string_view(subscribe, sizeof(subscribe)), false);
+  parser.ReadAndDispatchMessages();
+  EXPECT_EQ(visitor_.messages_received_, 0);
+  EXPECT_EQ(visitor_.parsing_error_, "Invalid filter type");
+  EXPECT_EQ(visitor_.parsing_error_code_, MoqtError::kProtocolViolation);
+}
+
 TEST_F(MoqtMessageSpecificTest, SubscribeOkHasAuthorizationToken) {
   webtransport::test::InMemoryStream stream(/*stream_id=*/0);
   MoqtControlParser parser(kWebTrans, &stream, visitor_);
@@ -978,13 +1053,13 @@
   webtransport::test::InMemoryStream stream(/*stream_id=*/0);
   MoqtControlParser parser(kRawQuic, &stream, visitor_);
   char subscribe[] = {
-      0x03, 0x00, 0x15, 0x01, 0x02,  // id and alias
+      0x03, 0x00, 0x18, 0x01, 0x02,  // id and alias
       0x01, 0x03, 0x66, 0x6f, 0x6f,  // track_namespace = "foo"
       0x04, 0x61, 0x62, 0x63, 0x64,  // track_name = "abcd"
-      0x20, 0x02,                    // priority = 0x20, group order descending
+      0x20, 0x02, 0x01,              // priority = 0x20, group order, forward
       0x02,                          // filter_type = kLatestObject
       0x01,                          // 1 parameter
-      0x03, 0x03, 0x62, 0x61, 0x72,  // authorization_info = "bar"
+      0x01, 0x05, 0x03, 0x00, 0x62, 0x61, 0x72,  // authorization_tag = "bar"
   };
   stream.Receive(absl::string_view(subscribe, sizeof(subscribe)), false);
   parser.ReadAndDispatchMessages();
@@ -1000,28 +1075,28 @@
   webtransport::test::InMemoryStream stream(/*stream_id=*/0);
   MoqtControlParser parser(kRawQuic, &stream, visitor_);
   char subscribe[] = {
-      0x03, 0x00, 0x15, 0x01, 0x02,  // id and alias
+      0x03, 0x00, 0x18, 0x01, 0x02,  // id and alias
       0x01, 0x03, 0x66, 0x6f, 0x6f,  // track_namespace = "foo"
       0x04, 0x61, 0x62, 0x63, 0x64,  // track_name = "abcd"
-      0x20, 0x08,                    // priority = 0x20 ???
-      0x01,                          // filter_type = kLatestGroup
+      0x20, 0x08, 0x01,              // priority, invalid order, forward
+      0x01,                          // filter_type = kNextGroupStart
       0x01,                          // 1 parameter
-      0x03, 0x03, 0x62, 0x61, 0x72,  // authorization_info = "bar"
+      0x01, 0x05, 0x03, 0x00, 0x62, 0x61, 0x72,  // authorization_tag = "bar"
   };
   stream.Receive(absl::string_view(subscribe, sizeof(subscribe)), false);
   parser.ReadAndDispatchMessages();
   EXPECT_EQ(visitor_.messages_received_, 0);
-  EXPECT_THAT(visitor_.parsing_error_, Optional(HasSubstr("group order")));
+  EXPECT_EQ(visitor_.parsing_error_, "Invalid group order value in SUBSCRIBE");
 }
 
 TEST_F(MoqtMessageSpecificTest, AbsoluteStart) {
   webtransport::test::InMemoryStream stream(/*stream_id=*/0);
   MoqtControlParser parser(kRawQuic, &stream, visitor_);
   char subscribe[] = {
-      0x03, 0x00, 0x17, 0x01, 0x02,  // id and alias
+      0x03, 0x00, 0x18, 0x01, 0x02,  // id and alias
       0x01, 0x03, 0x66, 0x6f, 0x6f,  // track_namespace = "foo"
       0x04, 0x61, 0x62, 0x63, 0x64,  // track_name = "abcd"
-      0x20, 0x02,                    // priority = 0x20 descending
+      0x20, 0x02, 0x01,              // priority, order, forward
       0x03,                          // filter_type = kAbsoluteStart
       0x04,                          // start_group = 4
       0x01,                          // start_object = 1
@@ -1043,10 +1118,10 @@
   webtransport::test::InMemoryStream stream(/*stream_id=*/0);
   MoqtControlParser parser(kRawQuic, &stream, visitor_);
   char subscribe[] = {
-      0x03, 0x00, 0x18, 0x01, 0x02,  // id and alias
+      0x03, 0x00, 0x19, 0x01, 0x02,  // id and alias
       0x01, 0x03, 0x66, 0x6f, 0x6f,  // track_namespace = "foo"
       0x04, 0x61, 0x62, 0x63, 0x64,  // track_name = "abcd"
-      0x20, 0x02,                    // priority = 0x20 descending
+      0x20, 0x02, 0x01,              // priority, order, forward
       0x04,                          // filter_type = kAbsoluteRange
       0x04,                          // start_group = 4
       0x01,                          // start_object = 1
@@ -1069,10 +1144,10 @@
   webtransport::test::InMemoryStream stream(/*stream_id=*/0);
   MoqtControlParser parser(kRawQuic, &stream, visitor_);
   char subscribe[] = {
-      0x03, 0x00, 0x18, 0x01, 0x02,  // id and alias
+      0x03, 0x00, 0x19, 0x01, 0x02,  // id and alias
       0x01, 0x03, 0x66, 0x6f, 0x6f,  // track_namespace = "foo"
       0x04, 0x61, 0x62, 0x63, 0x64,  // track_name = "abcd"
-      0x20, 0x02,                    // priority = 0x20 descending
+      0x20, 0x02, 0x01,              // priority, order, forward
       0x04,                          // filter_type = kAbsoluteRange
       0x04,                          // start_group = 4
       0x01,                          // start_object = 1
@@ -1090,10 +1165,10 @@
   webtransport::test::InMemoryStream stream(/*stream_id=*/0);
   MoqtControlParser parser(kRawQuic, &stream, visitor_);
   char subscribe[] = {
-      0x03, 0x00, 0x13, 0x01, 0x02,  // id and alias
+      0x03, 0x00, 0x14, 0x01, 0x02,  // id and alias
       0x01, 0x03, 0x66, 0x6f, 0x6f,  // track_namespace = "foo"
       0x04, 0x61, 0x62, 0x63, 0x64,  // track_name = "abcd"
-      0x20, 0x02,                    // priority = 0x20 descending
+      0x20, 0x02, 0x01,              // priority, order, forward
       0x04,                          // filter_type = kAbsoluteRange
       0x04,                          // start_group = 4
       0x01,                          // start_object = 1
diff --git a/quiche/quic/moqt/moqt_session.cc b/quiche/quic/moqt/moqt_session.cc
index 2d7ab24..5645885 100644
--- a/quiche/quic/moqt/moqt_session.cc
+++ b/quiche/quic/moqt/moqt_session.cc
@@ -71,9 +71,17 @@
   return status.ok() && DoesTrackStatusImplyHavingData(*status);
 }
 
-SubscribeWindow SubscribeMessageToWindow(const MoqtSubscribe& subscribe) {
-  return SubscribeWindow(subscribe.start.value_or(Location(0, 0)),
-                         subscribe.end_group);
+std::optional<SubscribeWindow> SubscribeMessageToWindow(
+    const MoqtSubscribe& subscribe) {
+  if (!subscribe.forward ||
+      subscribe.filter_type == MoqtFilterType::kLatestObject ||
+      subscribe.filter_type == MoqtFilterType::kNextGroupStart) {
+    return std::nullopt;
+  }
+  if (!subscribe.start.has_value()) {
+    return std::nullopt;
+  }
+  return SubscribeWindow(*subscribe.start, subscribe.end_group);
 }
 
 class DefaultPublisher : public MoqtPublisher {
@@ -111,6 +119,11 @@
       }
     });
   }
+  if (parameters_.perspective == Perspective::IS_SERVER) {
+    next_request_id_ = 1;
+  } else {
+    next_incoming_request_id_ = 1;
+  }
 }
 
 MoqtSession::ControlStream* MoqtSession::GetControlStream() {
@@ -327,6 +340,8 @@
   message.full_track_name = name;
   message.subscriber_priority = kDefaultSubscriberPriority;
   message.group_order = std::nullopt;
+  message.forward = true;
+  message.filter_type = MoqtFilterType::kAbsoluteStart;
   message.start = Location(start_group, start_object);
   message.end_group = std::nullopt;
   message.parameters = parameters;
@@ -346,6 +361,8 @@
   message.full_track_name = name;
   message.subscriber_priority = kDefaultSubscriberPriority;
   message.group_order = std::nullopt;
+  message.forward = true;
+  message.filter_type = MoqtFilterType::kAbsoluteRange;
   message.start = Location(start_group, start_object);
   message.end_group = end_group;
   message.parameters = parameters;
@@ -359,6 +376,8 @@
   message.full_track_name = name;
   message.subscriber_priority = kDefaultSubscriberPriority;
   message.group_order = std::nullopt;
+  message.forward = true;
+  message.filter_type = MoqtFilterType::kLatestObject;
   message.start = std::nullopt;
   message.end_group = std::nullopt;
   message.parameters = parameters;
@@ -396,7 +415,8 @@
   }
   MoqtFetch message;
   message.full_track_name = name;
-  message.fetch_id = next_request_id_++;
+  message.fetch_id = next_request_id_;
+  next_request_id_ += 2;
   message.start_object = start;
   message.end_group = end_group;
   message.end_object = end_object;
@@ -439,9 +459,9 @@
                                MoqtPriority priority,
                                std::optional<MoqtDeliveryOrder> delivery_order,
                                VersionSpecificParameters parameters) {
-  if ((next_request_id_ + 1) >= peer_max_request_id_) {
+  if ((next_request_id_ + 2) >= peer_max_request_id_) {
     QUIC_DLOG(INFO) << ENDPOINT << "Tried to send JOINING_FETCH with ID "
-                    << (next_request_id_ + 1)
+                    << (next_request_id_ + 2)
                     << " which is greater than the maximum ID "
                     << peer_max_request_id_;
     return false;
@@ -450,7 +470,8 @@
   subscribe.full_track_name = name;
   subscribe.subscriber_priority = priority;
   subscribe.group_order = delivery_order;
-  // Must be "Current Object" filter.
+  subscribe.forward = true;
+  subscribe.filter_type = MoqtFilterType::kLatestObject;
   subscribe.start = std::nullopt;
   subscribe.end_group = std::nullopt;
   subscribe.parameters = parameters;
@@ -458,10 +479,11 @@
     return false;
   }
   MoqtFetch fetch;
-  fetch.fetch_id = next_request_id_++;
+  fetch.fetch_id = next_request_id_;
+  next_request_id_ += 2;
   fetch.subscriber_priority = priority;
   fetch.group_order = delivery_order;
-  fetch.joining_fetch = {subscribe.subscribe_id, num_previous_groups};
+  fetch.joining_fetch = {subscribe.request_id, num_previous_groups};
   fetch.parameters = parameters;
   SendControlMessage(framer_.SerializeFetch(fetch));
   QUIC_DLOG(INFO) << ENDPOINT << "Sent Joining FETCH message for " << name;
@@ -614,7 +636,8 @@
     QUIC_DLOG(INFO) << ENDPOINT << "Tried to send SUBSCRIBE after GOAWAY";
     return false;
   }
-  message.subscribe_id = next_request_id_++;
+  message.request_id = next_request_id_;
+  next_request_id_ += 2;
   if (provided_track_alias.has_value()) {
     message.track_alias = *provided_track_alias;
     next_remote_track_alias_ =
@@ -626,7 +649,7 @@
     // Since we do not expose subscribe IDs directly in the API, instead wrap
     // the session and subscribe ID in a callback.
     visitor->OnCanAckObjects(absl::bind_front(&MoqtSession::SendObjectAck, this,
-                                              message.subscribe_id));
+                                              message.request_id));
   } else {
     QUICHE_DLOG_IF(WARNING, message.parameters.oack_window_size.has_value())
         << "Attempting to set object_ack_window on a connection that does not "
@@ -639,7 +662,7 @@
   auto track = std::make_unique<SubscribeRemoteTrack>(message, visitor);
   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));
+  upstream_by_id_.emplace(message.request_id, std::move(track));
   return true;
 }
 
@@ -776,7 +799,7 @@
 }
 
 void MoqtSession::GrantMoreRequests(uint64_t num_requests) {
-  local_max_request_id_ += num_requests;
+  local_max_request_id_ += (num_requests * 2);
   MoqtMaxRequestId message;
   message.max_request_id = local_max_request_id_;
   SendControlMessage(framer_.SerializeMaxRequestId(message));
@@ -788,13 +811,13 @@
     Error(MoqtError::kTooManyRequests, "Received request with too large ID");
     return false;
   }
-  if (request_id < next_incoming_request_id_) {
+  if (request_id != next_incoming_request_id_) {
     QUIC_DLOG(INFO) << ENDPOINT << "Request ID not monotonically increasing";
     Error(MoqtError::kInvalidRequestId,
           "Request ID not monotonically increasing");
     return false;
   }
-  next_incoming_request_id_ = request_id + 1;
+  next_incoming_request_id_ = request_id + 2;
   return true;
 }
 
@@ -902,14 +925,14 @@
 
 void MoqtSession::ControlStream::OnSubscribeMessage(
     const MoqtSubscribe& message) {
-  if (!session_->ValidateRequestId(message.subscribe_id)) {
+  if (!session_->ValidateRequestId(message.request_id)) {
     return;
   }
   QUIC_DLOG(INFO) << ENDPOINT << "Received a SUBSCRIBE for "
                   << message.full_track_name;
   if (session_->sent_goaway_) {
     QUIC_DLOG(INFO) << ENDPOINT << "Received a SUBSCRIBE after GOAWAY";
-    SendSubscribeError(message.subscribe_id, SubscribeErrorCode::kUnauthorized,
+    SendSubscribeError(message.request_id, SubscribeErrorCode::kUnauthorized,
                        "SUBSCRIBE after GOAWAY", message.track_alias);
     return;
   }
@@ -925,7 +948,7 @@
     QUIC_DLOG(INFO) << ENDPOINT << "SUBSCRIBE for " << track_name
                     << " rejected by the application: "
                     << track_publisher.status();
-    SendSubscribeError(message.subscribe_id, SubscribeErrorCode::kDoesNotExist,
+    SendSubscribeError(message.request_id, SubscribeErrorCode::kDoesNotExist,
                        track_publisher.status().message(), message.track_alias);
     return;
   }
@@ -945,7 +968,7 @@
   subscription->set_delivery_timeout(message.parameters.delivery_timeout);
   MoqtSession::PublishedSubscription* subscription_ptr = subscription.get();
   auto [it, success] = session_->published_subscriptions_.emplace(
-      message.subscribe_id, std::move(subscription));
+      message.request_id, std::move(subscription));
   if (!success) {
     QUICHE_NOTREACHED();  // ValidateRequestId() should have caught this.
   }
@@ -1692,11 +1715,12 @@
     MoqtSession* session, std::shared_ptr<MoqtTrackPublisher> track_publisher,
     const MoqtSubscribe& subscribe,
     MoqtPublishingMonitorInterface* monitoring_interface)
-    : filter_type_(GetFilterType(subscribe)),
-      subscription_id_(subscribe.subscribe_id),
-      session_(session),
+    : session_(session),
       track_publisher_(track_publisher),
+      request_id_(subscribe.request_id),
       track_alias_(subscribe.track_alias),
+      filter_type_(subscribe.filter_type),
+      forward_(subscribe.forward),
       window_(SubscribeMessageToWindow(subscribe)),
       subscriber_priority_(subscribe.subscriber_priority),
       subscriber_delivery_order_(subscribe.group_order),
@@ -1731,12 +1755,18 @@
 void MoqtSession::PublishedSubscription::Update(
     Location start, std::optional<uint64_t> end_group,
     MoqtPriority subscriber_priority) {
-  window_.TruncateStart(start);
-  if (end_group.has_value()) {
-    window_.TruncateEnd(*end_group);
-  }
   subscriber_priority_ = subscriber_priority;
+  if (!window_.has_value()) {
+    window_ = SubscribeWindow(start, end_group);
+    return;
+  }
+  window_->TruncateStart(start);
+  if (end_group.has_value()) {
+    window_->TruncateEnd(*end_group);
+  }
   // TODO: update priority of all data streams that are currently open.
+  // TODO: update delivery timeout.
+  // TODO: update forward and subscribe filter.
 
   // TODO: reset streams that are no longer in-window.
   // TODO: send SUBSCRIBE_DONE if required.
@@ -1755,7 +1785,7 @@
   webtransport::SendOrder old_send_order =
       FinalizeSendOrder(queued_outgoing_data_streams_.rbegin()->first);
   subscriber_priority_ = priority;
-  session_->UpdateQueuedSendOrder(subscription_id_, old_send_order,
+  session_->UpdateQueuedSendOrder(request_id_, old_send_order,
                                   FinalizeSendOrder(old_send_order));
 };
 
@@ -1764,21 +1794,26 @@
   ControlStream* stream = session_->GetControlStream();
   if (PublisherHasData(*track_publisher_)) {
     largest_id = track_publisher_->GetLargestSequence();
-    if (window_.end() < *largest_id) {
-      stream->SendSubscribeError(subscription_id_,
-                                 SubscribeErrorCode::kInvalidRange,
-                                 "SUBSCRIBE ends in past group", track_alias_);
-      session_->published_subscriptions_.erase(subscription_id_);
-      // No class access below this line!
-      return;
+    QUICHE_CHECK(largest_id.has_value());
+    if (forward_) {
+      switch (filter_type_) {
+        case MoqtFilterType::kLatestObject:
+          window_ = SubscribeWindow(largest_id->next());
+          break;
+        case MoqtFilterType::kNextGroupStart:
+          window_ = SubscribeWindow(Location(largest_id->group + 1, 0));
+          break;
+        default:
+          break;
+      }
     }
-    if (!window_.TruncateStart(largest_id->next())) {
-      QUICHE_NOTREACHED();
-    };
+  } else if (filter_type_ == MoqtFilterType::kLatestObject ||
+             filter_type_ == MoqtFilterType::kNextGroupStart) {
+    // No data yet. All objects will be in-window.
+    window_ = SubscribeWindow(Location(0, 0));
   }
-
   MoqtSubscribeOk subscribe_ok;
-  subscribe_ok.subscribe_id = subscription_id_;
+  subscribe_ok.subscribe_id = request_id_;
   subscribe_ok.group_order = track_publisher_->GetDeliveryOrder();
   subscribe_ok.largest_id = largest_id;
   // TODO(martinduke): Support sending DELIVERY_TIMEOUT parameter as the
@@ -1790,15 +1825,15 @@
 void MoqtSession::PublishedSubscription::OnSubscribeRejected(
     MoqtSubscribeErrorReason reason, std::optional<uint64_t> track_alias) {
   session_->GetControlStream()->SendSubscribeError(
-      subscription_id_, reason.error_code, reason.reason_phrase,
+      request_id_, reason.error_code, reason.reason_phrase,
       track_alias.value_or(track_alias_));
-  session_->published_subscriptions_.erase(subscription_id_);
+  session_->published_subscriptions_.erase(request_id_);
   // No class access below this line!
 }
 
 void MoqtSession::PublishedSubscription::OnNewObjectAvailable(
     Location sequence) {
-  if (!window_.InWindow(sequence)) {
+  if (!InWindow(sequence)) {
     return;
   }
   if (reset_subgroups_.contains(
@@ -1841,7 +1876,7 @@
   if (stream_id.has_value()) {
     raw_stream = session_->session_->GetStreamById(*stream_id);
   } else {
-    raw_stream = session_->OpenOrQueueDataStream(subscription_id_, sequence);
+    raw_stream = session_->OpenOrQueueDataStream(request_id_, sequence);
   }
   if (raw_stream == nullptr) {
     return;
@@ -1853,12 +1888,12 @@
 }
 
 void MoqtSession::PublishedSubscription::OnTrackPublisherGone() {
-  session_->SubscribeIsDone(subscription_id_, SubscribeDoneCode::kGoingAway,
+  session_->SubscribeIsDone(request_id_, SubscribeDoneCode::kGoingAway,
                             "Publisher is gone");
 }
 
 void MoqtSession::PublishedSubscription::OnNewFinAvailable(Location sequence) {
-  if (!window_.InWindow(sequence)) {
+  if (!InWindow(sequence)) {
     return;
   }
   if (reset_subgroups_.contains(
@@ -1884,7 +1919,7 @@
 
 void MoqtSession::PublishedSubscription::OnSubgroupAbandoned(
     Location sequence, webtransport::StreamErrorCode error_code) {
-  if (!window_.InWindow(sequence)) {
+  if (!InWindow(sequence)) {
     return;
   }
   if (reset_subgroups_.contains(
@@ -1907,6 +1942,11 @@
 }
 
 void MoqtSession::PublishedSubscription::OnGroupAbandoned(uint64_t group_id) {
+  if (!window_.has_value() || window_->end().group < group_id ||
+      window_->start().group > group_id) {
+    // The group is not in the window, ignore.
+    return;
+  }
   std::vector<webtransport::StreamId> streams =
       stream_map().GetStreamsForGroup(group_id);
   for (webtransport::StreamId stream_id : streams) {
@@ -1961,10 +2001,10 @@
   queued_outgoing_data_streams_.emplace(
       UpdateSendOrderForSubscriberPriority(send_order, 0), first_object);
   if (!start_send_order.has_value()) {
-    session_->UpdateQueuedSendOrder(subscription_id_, std::nullopt, send_order);
+    session_->UpdateQueuedSendOrder(request_id_, std::nullopt, send_order);
   } else if (*start_send_order < send_order) {
     session_->UpdateQueuedSendOrder(
-        subscription_id_, FinalizeSendOrder(*start_send_order), send_order);
+        request_id_, FinalizeSendOrder(*start_send_order), send_order);
   }
 }
 
@@ -1980,13 +2020,12 @@
   // then taking base().
   queued_outgoing_data_streams_.erase((++it).base());
   if (queued_outgoing_data_streams_.empty()) {
-    session_->UpdateQueuedSendOrder(subscription_id_, old_send_order,
-                                    std::nullopt);
+    session_->UpdateQueuedSendOrder(request_id_, old_send_order, std::nullopt);
   } else {
     webtransport::SendOrder new_send_order =
         FinalizeSendOrder(queued_outgoing_data_streams_.rbegin()->first);
     if (old_send_order != new_send_order) {
-      session_->UpdateQueuedSendOrder(subscription_id_, old_send_order,
+      session_->UpdateQueuedSendOrder(request_id_, old_send_order,
                                       new_send_order);
     }
   }
@@ -2017,7 +2056,7 @@
     PublishedSubscription& subscription, Location first_object)
     : session_(session),
       stream_(stream),
-      subscription_id_(subscription.subscription_id()),
+      subscription_id_(subscription.request_id()),
       next_object_(first_object),
       session_liveness_(session->liveness_token_) {
   UpdateSendOrder(subscription);
diff --git a/quiche/quic/moqt/moqt_session.h b/quiche/quic/moqt/moqt_session.h
index eceaa98..bc4e80d 100644
--- a/quiche/quic/moqt/moqt_session.h
+++ b/quiche/quic/moqt/moqt_session.h
@@ -113,6 +113,7 @@
   // the track, the message will still be sent. However, the visitor will be
   // ignored.
   // Subscribe from (start_group, start_object) to the end of the track.
+  // TODO(martinduke): Allow setting forward = false.
   bool SubscribeAbsolute(const FullTrackName& name, uint64_t start_group,
                          uint64_t start_object,
                          SubscribeRemoteTrack::Visitor* visitor,
@@ -125,6 +126,7 @@
   bool SubscribeCurrentObject(const FullTrackName& name,
                               SubscribeRemoteTrack::Visitor* visitor,
                               VersionSpecificParameters parameters) override;
+  // TODO(martinduke): SubscribeNextGroup
   // Returns false if the subscription is not found. The session immediately
   // destroys all subscription state.
   void Unsubscribe(const FullTrackName& name);
@@ -337,7 +339,7 @@
     PublishedSubscription& operator=(const PublishedSubscription&) = delete;
     PublishedSubscription& operator=(PublishedSubscription&&) = delete;
 
-    uint64_t subscription_id() const { return subscription_id_; }
+    uint64_t request_id() const { return request_id_; }
     MoqtTrackPublisher& publisher() { return *track_publisher_; }
     uint64_t track_alias() const { return track_alias_; }
     std::optional<Location> largest_sent() const { return largest_sent_; }
@@ -372,8 +374,13 @@
                 MoqtPriority subscriber_priority);
     // Checks if the specified sequence is within the window of this
     // subscription.
-    bool InWindow(Location sequence) { return window_.InWindow(sequence); }
-    Location GetWindowStart() const { return window_.start(); }
+    bool InWindow(Location sequence) {
+      return forward_ && window_.has_value() && window_->InWindow(sequence);
+    }
+    Location GetWindowStart() const {
+      QUICHE_CHECK(window_.has_value());
+      return window_->start();
+    }
     MoqtFilterType filter_type() const { return filter_type_; };
 
     void OnDataStreamCreated(webtransport::StreamId id,
@@ -427,12 +434,16 @@
                                                   subscriber_priority_);
     }
 
-    MoqtFilterType filter_type_;
-    uint64_t subscription_id_;
     MoqtSession* session_;
     std::shared_ptr<MoqtTrackPublisher> track_publisher_;
+    uint64_t request_id_;
     uint64_t track_alias_;
-    SubscribeWindow window_;
+    MoqtFilterType filter_type_;
+    bool forward_;
+    // If window_ is nullopt, any arriving objects are ignored. This could be
+    // because forward=0, or because the subscription is waiting for a
+    // SUBSCRIBE_OK and doesn't know what the window should be yet.
+    std::optional<SubscribeWindow> window_;
     MoqtPriority subscriber_priority_;
     uint64_t streams_opened_ = 0;
 
diff --git a/quiche/quic/moqt/moqt_session_test.cc b/quiche/quic/moqt/moqt_session_test.cc
index 2c8797c..1487e25 100644
--- a/quiche/quic/moqt/moqt_session_test.cc
+++ b/quiche/quic/moqt/moqt_session_test.cc
@@ -60,11 +60,13 @@
 
 MoqtSubscribe DefaultSubscribe() {
   MoqtSubscribe subscribe = {
-      /*subscribe_id=*/1,
+      /*request_id=*/1,
       /*track_alias=*/2,
       kDefaultTrackName(),
       /*subscriber_priority=*/0x80,
       /*group_order=*/std::nullopt,
+      /*forward=*/true,
+      /*filter_type=*/MoqtFilterType::kAbsoluteStart,
       /*start=*/Location(0, 0),
       /*end_group=*/std::nullopt,
       /*parameters=*/VersionSpecificParameters(),
@@ -74,7 +76,7 @@
 
 MoqtFetch DefaultFetch() {
   MoqtFetch fetch = {
-      /*fetch_id=*/2,
+      /*fetch_id=*/1,
       /*subscriber_priority=*/0x80,
       /*group_order=*/std::nullopt,
       /*joining_fetch=*/std::nullopt,
@@ -155,7 +157,7 @@
       return nullptr;
     }
     MoqtSubscribeOk expected_ok = {
-        /*subscribe_id=*/subscribe.subscribe_id,
+        /*request_id=*/subscribe.request_id,
         /*expires=*/quic::QuicTimeDelta::FromMilliseconds(0),
         /*group_order=*/MoqtDeliveryOrder::kAscending,
         (*track_status == MoqtTrackStatusCode::kInProgress)
@@ -346,7 +348,7 @@
   // Add the track. Now Subscribe should succeed.
   MockTrackPublisher* track = CreateTrackPublisher();
   std::make_shared<MockTrackPublisher>(request.full_track_name);
-  ++request.subscribe_id;
+  request.request_id += 2;
   ReceiveSubscribeSynchronousOk(track, request, stream_input.get());
 }
 
@@ -504,28 +506,56 @@
   ReceiveSubscribeSynchronousOk(track, request, stream_input.get());
 }
 
-TEST_F(MoqtSessionTest, SubscribeEntirelyInPast) {
+TEST_F(MoqtSessionTest, SubscribeDoNotForward) {
   std::unique_ptr<MoqtControlParserVisitor> stream_input =
       MoqtSessionPeer::CreateControlStream(&session_, &mock_stream_);
   MockTrackPublisher* track = CreateTrackPublisher();
-  SetLargestId(track, Location(10, 20));
-
   MoqtSubscribe request = DefaultSubscribe();
-  request.end_group = 9;
-  EXPECT_CALL(*track, AddObjectListener)
-      .WillOnce([&](MoqtObjectListener* listener) {
-        listener->OnSubscribeAccepted();
-      });
-  MoqtSubscribeError expected_error = {
-      /*subscribe_id=*/request.subscribe_id,
-      /*error_code=*/SubscribeErrorCode::kInvalidRange,
-      /*reason_phrase=*/"SUBSCRIBE ends in past group",
-      /*track_alias=*/request.track_alias,
-  };
-  EXPECT_CALL(mock_stream_,
-              Writev(SerializedControlMessage(expected_error), _));
-  stream_input->OnSubscribeMessage(request);
-  EXPECT_EQ(MoqtSessionPeer::GetSubscription(&session_, 1), nullptr);
+  request.forward = false;
+  request.filter_type = MoqtFilterType::kLatestObject;
+  EXPECT_CALL(*track, GetTrackStatus)
+      .WillRepeatedly(Return(MoqtTrackStatusCode::kNotYetBegun));
+  MoqtObjectListener* listener =
+      ReceiveSubscribeSynchronousOk(track, request, stream_input.get());
+  // forward=false, so incoming objects are ignored.
+  EXPECT_CALL(mock_session_, CanOpenNextOutgoingUnidirectionalStream())
+      .Times(0);
+  listener->OnNewObjectAvailable(Location(0, 0));
+}
+
+TEST_F(MoqtSessionTest, SubscribeAbsoluteStartNoDataYet) {
+  std::unique_ptr<MoqtControlParserVisitor> stream_input =
+      MoqtSessionPeer::CreateControlStream(&session_, &mock_stream_);
+  MockTrackPublisher* track = CreateTrackPublisher();
+  MoqtSubscribe request = DefaultSubscribe();
+  request.start = Location(1, 0);
+  EXPECT_CALL(*track, GetTrackStatus)
+      .WillRepeatedly(Return(MoqtTrackStatusCode::kNotYetBegun));
+  MoqtObjectListener* listener =
+      ReceiveSubscribeSynchronousOk(track, request, stream_input.get());
+  // Window was not set to (0, 0) by SUBSCRIBE acceptance.
+  EXPECT_CALL(mock_session_, CanOpenNextOutgoingUnidirectionalStream())
+      .Times(0);
+  listener->OnNewObjectAvailable(Location(0, 0));
+}
+
+TEST_F(MoqtSessionTest, SubscribeNextGroup) {
+  std::unique_ptr<MoqtControlParserVisitor> stream_input =
+      MoqtSessionPeer::CreateControlStream(&session_, &mock_stream_);
+  MockTrackPublisher* track = CreateTrackPublisher();
+  MoqtSubscribe request = DefaultSubscribe();
+  request.filter_type = MoqtFilterType::kNextGroupStart;
+  SetLargestId(track, Location(10, 20));
+  MoqtObjectListener* listener =
+      ReceiveSubscribeSynchronousOk(track, request, stream_input.get());
+  // Later objects in group 10 ignored.
+  EXPECT_CALL(mock_session_, CanOpenNextOutgoingUnidirectionalStream())
+      .Times(0);
+  listener->OnNewObjectAvailable(Location(10, 21));
+  // Group 11 is sent.
+  EXPECT_CALL(mock_session_, CanOpenNextOutgoingUnidirectionalStream())
+      .WillOnce(Return(false));
+  listener->OnNewObjectAvailable(Location(11, 0));
 }
 
 TEST_F(MoqtSessionTest, TwoSubscribesForTrack) {
@@ -535,7 +565,7 @@
   MoqtSubscribe request = DefaultSubscribe();
   ReceiveSubscribeSynchronousOk(track, request, stream_input.get());
 
-  request.subscribe_id = 2;
+  request.request_id = 3;
   request.start = Location(12, 0);
   EXPECT_CALL(mock_session_,
               CloseSession(static_cast<uint64_t>(MoqtError::kProtocolViolation),
@@ -553,13 +583,13 @@
 
   // Peer unsubscribes.
   MoqtUnsubscribe unsubscribe = {
-      /*subscribe_id=*/1,
+      /*request_id=*/1,
   };
   stream_input->OnUnsubscribeMessage(unsubscribe);
   EXPECT_EQ(MoqtSessionPeer::GetSubscription(&session_, 1), nullptr);
 
   // Subscribe again, succeeds.
-  request.subscribe_id = 2;
+  request.request_id = 3;
   request.start = Location(12, 0);
   ReceiveSubscribeSynchronousOk(track, request, stream_input.get());
 }
@@ -567,7 +597,7 @@
 TEST_F(MoqtSessionTest, RequestIdTooHigh) {
   // Peer subscribes to (0, 0)
   MoqtSubscribe request = DefaultSubscribe();
-  request.subscribe_id = kDefaultInitialMaxRequestId + 1;
+  request.request_id = kDefaultInitialMaxRequestId + 1;
 
   std::unique_ptr<MoqtControlParserVisitor> stream_input =
       MoqtSessionPeer::CreateControlStream(&session_, &mock_stream_);
@@ -577,6 +607,10 @@
   stream_input->OnSubscribeMessage(request);
 }
 
+TEST_F(MoqtSessionTest, RequestIdWrongLsb) {
+  // TODO(martinduke): Implement this test.
+}
+
 TEST_F(MoqtSessionTest, SubscribeIdNotIncreasing) {
   MoqtSubscribe request = DefaultSubscribe();
   std::unique_ptr<MoqtControlParserVisitor> stream_input =
@@ -650,7 +684,7 @@
                                   VersionSpecificParameters());
 
   MoqtSubscribeOk ok = {
-      /*subscribe_id=*/0,
+      /*request_id=*/0,
       /*expires=*/quic::QuicTimeDelta::FromMilliseconds(0),
   };
   EXPECT_CALL(remote_track_visitor, OnReply(_, _, _))
@@ -709,7 +743,9 @@
   session_.GrantMoreRequests(1);
   // Peer subscribes to (0, 0)
   MoqtSubscribe request = DefaultSubscribe();
-  request.subscribe_id = kDefaultInitialMaxRequestId;
+  MoqtSessionPeer::set_next_incoming_request_id(
+      &session_, kDefaultInitialMaxRequestId + 1);
+  request.request_id = kDefaultInitialMaxRequestId + 1;
   MockTrackPublisher* track = CreateTrackPublisher();
   ReceiveSubscribeSynchronousOk(track, request, stream_input.get());
 }
@@ -726,7 +762,7 @@
                                   VersionSpecificParameters());
 
   MoqtSubscribeError error = {
-      /*subscribe_id=*/0,
+      /*request_id=*/0,
       /*error_code=*/SubscribeErrorCode::kInvalidRange,
       /*reason_phrase=*/"deadbeef",
       /*track_alias=*/2,
@@ -1011,7 +1047,7 @@
 
   // SUBSCRIBE_OK arrives
   MoqtSubscribeOk ok = {
-      /*subscribe_id=*/1,
+      /*request_id=*/1,
       /*expires=*/quic::QuicTimeDelta::FromMilliseconds(0),
       /*group_order=*/MoqtDeliveryOrder::kAscending,
       /*largest_id=*/std::nullopt,
@@ -1056,7 +1092,7 @@
 
   // SUBSCRIBE_ERROR arrives
   MoqtSubscribeError subscribe_error = {
-      /*subscribe_id=*/1,
+      /*request_id=*/1,
       /*error_code=*/SubscribeErrorCode::kRetryTrackAlias,
       /*reason_phrase=*/"foo",
       /*track_alias =*/3,
@@ -1078,7 +1114,7 @@
 
   // SUBSCRIBE_ERROR arrives
   MoqtSubscribeError subscribe_error = {
-      /*subscribe_id=*/1,
+      /*request_id=*/1,
       /*error_code=*/SubscribeErrorCode::kRetryTrackAlias,
       /*reason_phrase=*/"foo",
       /*track_alias =*/3,
@@ -1098,7 +1134,7 @@
 
   // SUBSCRIBE_ERROR arrives
   MoqtSubscribeError subscribe_error = {
-      /*subscribe_id=*/1,
+      /*request_id=*/1,
       /*error_code=*/SubscribeErrorCode::kRetryTrackAlias,
       /*reason_phrase=*/"foo",
       /*track_alias =*/2,
@@ -1680,7 +1716,7 @@
   std::unique_ptr<MoqtControlParserVisitor> stream_input =
       MoqtSessionPeer::CreateControlStream(&session_, &mock_stream_);
   MoqtUnsubscribe unsubscribe = {
-      /*subscribe_id=*/0,
+      /*request_id=*/0,
   };
   stream_input->OnUnsubscribeMessage(unsubscribe);
   EXPECT_EQ(MoqtSessionPeer::GetSubscription(&session_, 0), nullptr);
@@ -2225,12 +2261,12 @@
 
 TEST_F(MoqtSessionTest, InvalidFetch) {
   // Update the state so that it expects ID > 0 next time.
-  MoqtSessionPeer::ValidateRequestId(&session_, 0);
+  MoqtSessionPeer::ValidateRequestId(&session_, 1);
   webtransport::test::MockStream control_stream;
   std::unique_ptr<MoqtControlParserVisitor> stream_input =
       MoqtSessionPeer::CreateControlStream(&session_, &control_stream);
   MoqtFetch fetch = DefaultFetch();
-  fetch.fetch_id = 0;  // Too low.
+  fetch.fetch_id = 1;  // Too low.
   EXPECT_CALL(mock_session_,
               CloseSession(static_cast<uint64_t>(MoqtError::kInvalidRequestId),
                            "Request ID not monotonically increasing"))
@@ -2286,6 +2322,7 @@
 TEST_F(MoqtSessionTest, IncomingJoiningFetch) {
   MoqtSubscribe subscribe = DefaultSubscribe();
   // Give it the latest object filter.
+  subscribe.filter_type = MoqtFilterType::kLatestObject;
   subscribe.start = std::nullopt;
   subscribe.end_group = std::nullopt;
   std::unique_ptr<MoqtControlParserVisitor> stream_input =
@@ -2295,7 +2332,7 @@
   ReceiveSubscribeSynchronousOk(track, subscribe, stream_input.get());
 
   MoqtObjectListener* subscription =
-      MoqtSessionPeer::GetSubscription(&session_, subscribe.subscribe_id);
+      MoqtSessionPeer::GetSubscription(&session_, subscribe.request_id);
   ASSERT_NE(subscription, nullptr);
   EXPECT_TRUE(
       MoqtSessionPeer::InSubscriptionWindow(subscription, Location(4, 0, 11)));
@@ -2304,6 +2341,7 @@
 
   // Joining FETCH arrives. The resulting Fetch should begin at (2, 0).
   MoqtFetch fetch = DefaultFetch();
+  fetch.fetch_id = 3;
   fetch.joining_fetch = {1, 2};
   EXPECT_CALL(*track, Fetch(Location(2, 0), 4, std::optional<uint64_t>(10), _))
       .WillOnce(Return(std::make_unique<MockFetchTask>()));
@@ -2316,7 +2354,7 @@
   MoqtFetch fetch = DefaultFetch();
   fetch.joining_fetch = {1, 2};
   MoqtFetchError expected_error = {
-      /*subscribe_id=*/2,
+      /*request_id=*/1,
       /*error_code=*/SubscribeErrorCode::kDoesNotExist,
       /*reason_phrase=*/"Joining Fetch for non-existent subscribe",
   };
@@ -2334,6 +2372,7 @@
   ReceiveSubscribeSynchronousOk(track, subscribe, stream_input.get());
 
   MoqtFetch fetch = DefaultFetch();
+  fetch.fetch_id = 3;
   fetch.joining_fetch = {1, 2};
   EXPECT_CALL(mock_session_,
               CloseSession(static_cast<uint64_t>(MoqtError::kProtocolViolation),
@@ -2349,16 +2388,19 @@
   EXPECT_CALL(mock_session_, GetStreamById(_))
       .WillRepeatedly(Return(&mock_stream_));
   MoqtSubscribe expected_subscribe = {
-      /*subscribe_id=*/0,
+      /*request_id=*/0,
       /*track_alias=*/0,
       /*full_track_name=*/FullTrackName("foo", "bar"),
       /*subscriber_priority=*/0x80,
       /*group_order=*/MoqtDeliveryOrder::kAscending,
+      /*forward=*/true,
+      /*filter_type=*/MoqtFilterType::kLatestObject,
       /*start=*/std::nullopt,
       /*end_group=*/std::nullopt,
+      VersionSpecificParameters(),
   };
   MoqtFetch expected_fetch = {
-      /*fetch_id=*/1,
+      /*fetch_id=*/2,
       /*subscriber_priority=*/0x80,
       /*group_order=*/MoqtDeliveryOrder::kAscending,
       /*joining_fetch=*/JoiningFetch(0, 1),
@@ -2391,12 +2433,12 @@
       MoqtSubscribeOk(0, quic::QuicTimeDelta::FromMilliseconds(0),
                       MoqtDeliveryOrder::kAscending, Location(2, 0),
                       VersionSpecificParameters()));
-  stream_input->OnFetchOkMessage(MoqtFetchOk(1, MoqtDeliveryOrder::kAscending,
+  stream_input->OnFetchOkMessage(MoqtFetchOk(2, MoqtDeliveryOrder::kAscending,
                                              Location(2, 0),
                                              VersionSpecificParameters()));
   // Packet arrives on FETCH stream.
   MoqtObject object = {
-      /*fetch_id=*/1,
+      /*fetch_id=*/2,
       /*group_id, object_id=*/0,
       0,
       /*publisher_priority=*/128,
@@ -2479,7 +2521,7 @@
       Location(0, 0), 4, std::nullopt, 128, std::nullopt,
       VersionSpecificParameters());
   MoqtFetchOk ok = {
-      /*subscribe_id=*/0,
+      /*request_id=*/0,
       /*group_order=*/MoqtDeliveryOrder::kAscending,
       /*largest_id=*/Location(3, 25),
       VersionSpecificParameters(),
@@ -2508,7 +2550,7 @@
       Location(0, 0), 4, std::nullopt, 128, std::nullopt,
       VersionSpecificParameters());
   MoqtFetchError error = {
-      /*subscribe_id=*/0,
+      /*request_id=*/0,
       /*error_code=*/SubscribeErrorCode::kUnauthorized,
       /*reason_phrase=*/"No username provided",
   };
@@ -2546,7 +2588,7 @@
   std::queue<quiche::QuicheBuffer> headers;
   std::queue<std::string> payloads;
   MoqtObject object = {
-      /*subscribe_id=*/0,
+      /*request_id=*/0,
       /*group_id, object_id=*/0,
       0,
       /*publisher_priority=*/128,
@@ -2578,7 +2620,7 @@
 
   // FETCH_OK arrives, objects are delivered.
   MoqtFetchOk ok = {
-      /*subscribe_id=*/0,
+      /*request_id=*/0,
       /*group_order=*/MoqtDeliveryOrder::kAscending,
       /*largest_id=*/Location(3, 25),
       VersionSpecificParameters(),
@@ -2616,7 +2658,7 @@
   std::queue<quiche::QuicheBuffer> headers;
   std::queue<std::string> payloads;
   MoqtObject object = {
-      /*subscribe_id=*/0,
+      /*request_id=*/0,
       /*group_id, object_id=*/0,
       0,
       /*publisher_priority=*/128,
@@ -2648,7 +2690,7 @@
 
   // FETCH_OK arrives, objects are available.
   MoqtFetchOk ok = {
-      /*subscribe_id=*/0,
+      /*request_id=*/0,
       /*group_order=*/MoqtDeliveryOrder::kAscending,
       /*largest_id=*/Location(3, 25),
       VersionSpecificParameters(),
@@ -2707,7 +2749,7 @@
   bool object_ready = false;
   task->SetObjectAvailableCallback([&]() { object_ready = true; });
   MoqtObject object = {
-      /*subscribe_id=*/0,
+      /*request_id=*/0,
       /*group_id, object_id=*/0,
       0,
       /*publisher_priority=*/128,
@@ -3035,6 +3077,7 @@
   EXPECT_CALL(mock_stream_,
               Writev(ControlMessageOfType(MoqtMessageType::kFetchError), _));
   MoqtFetch fetch = DefaultFetch();
+  fetch.fetch_id = 3;
   stream_input->OnFetchMessage(fetch);
   EXPECT_CALL(
       mock_stream_,
@@ -3121,7 +3164,7 @@
                                               &remote_track_visitor,
                                               VersionSpecificParameters()));
   MoqtSubscribeOk ok = {
-      /*subscribe_id=*/0,
+      /*request_id=*/0,
       /*expires=*/quic::QuicTimeDelta::FromMilliseconds(10000),
       /*group_order=*/MoqtDeliveryOrder::kAscending,
       /*largest_id=*/std::nullopt,
@@ -3179,7 +3222,7 @@
                                               &remote_track_visitor,
                                               VersionSpecificParameters()));
   MoqtSubscribeOk ok = {
-      /*subscribe_id=*/0,
+      /*request_id=*/0,
       /*expires=*/quic::QuicTimeDelta::FromMilliseconds(10000),
       /*group_order=*/MoqtDeliveryOrder::kAscending,
       /*largest_id=*/std::nullopt,
@@ -3234,7 +3277,7 @@
                                               &remote_track_visitor,
                                               VersionSpecificParameters()));
   MoqtSubscribeOk ok = {
-      /*subscribe_id=*/0,
+      /*request_id=*/0,
       /*expires=*/quic::QuicTimeDelta::FromMilliseconds(10000),
       /*group_order=*/MoqtDeliveryOrder::kAscending,
       /*largest_id=*/std::nullopt,
@@ -3297,7 +3340,7 @@
                                     MoqtObjectStatus::kNormal);
   // Update the end to fall at the last delivered object.
   MoqtSubscribeUpdate update = {
-      /*subscribe_id=*/0,
+      /*request_id=*/0,
       /*start_group=*/5,
       /*start_object=*/0,
       /*end_group=*/7,
diff --git a/quiche/quic/moqt/moqt_track.h b/quiche/quic/moqt/moqt_track.h
index a829930..0783141 100644
--- a/quiche/quic/moqt/moqt_track.h
+++ b/quiche/quic/moqt/moqt_track.h
@@ -116,7 +116,7 @@
     virtual void OnSubscribeDone(FullTrackName full_track_name) = 0;
   };
   SubscribeRemoteTrack(const MoqtSubscribe& subscribe, Visitor* visitor)
-      : RemoteTrack(subscribe.full_track_name, subscribe.subscribe_id,
+      : RemoteTrack(subscribe.full_track_name, subscribe.request_id,
                     SubscribeWindow(subscribe.start.value_or(Location()),
                                     subscribe.end_group)),
         track_alias_(subscribe.track_alias),
diff --git a/quiche/quic/moqt/moqt_track_test.cc b/quiche/quic/moqt/moqt_track_test.cc
index 9ab9078..7bca1db 100644
--- a/quiche/quic/moqt/moqt_track_test.cc
+++ b/quiche/quic/moqt/moqt_track_test.cc
@@ -54,6 +54,8 @@
       /*full_track_name=*/FullTrackName("foo", "bar"),
       /*subscriber_priority=*/128,
       /*group_order=*/std::nullopt,
+      /*forward=*/true,
+      /*filter_type=*/MoqtFilterType::kAbsoluteStart,
       /*start=*/Location(2, 0),
       std::nullopt,
       VersionSpecificParameters(),
@@ -78,7 +80,7 @@
 
 TEST_F(SubscribeRemoteTrackTest, AllowError) {
   EXPECT_TRUE(track_.ErrorIsAllowed());
-  EXPECT_EQ(track_.GetSubscribe().subscribe_id, subscribe_.subscribe_id);
+  EXPECT_EQ(track_.GetSubscribe().request_id, subscribe_.request_id);
   track_.OnObjectOrOk();
   EXPECT_FALSE(track_.ErrorIsAllowed());
 }
diff --git a/quiche/quic/moqt/test_tools/moqt_session_peer.h b/quiche/quic/moqt/test_tools/moqt_session_peer.h
index 9fa1bf3..e3ddb7b 100644
--- a/quiche/quic/moqt/test_tools/moqt_session_peer.h
+++ b/quiche/quic/moqt/test_tools/moqt_session_peer.h
@@ -91,7 +91,7 @@
                                              track.get());
     session->subscribe_by_name_.try_emplace(subscribe.full_track_name,
                                             track.get());
-    session->upstream_by_id_.try_emplace(subscribe.subscribe_id,
+    session->upstream_by_id_.try_emplace(subscribe.request_id,
                                          std::move(track));
   }
 
@@ -102,7 +102,9 @@
     MoqtSubscribe subscribe;
     subscribe.full_track_name = publisher->GetTrackName();
     subscribe.track_alias = track_alias;
-    subscribe.subscribe_id = subscribe_id;
+    subscribe.request_id = subscribe_id;
+    subscribe.forward = true;
+    subscribe.filter_type = MoqtFilterType::kAbsoluteStart;
     subscribe.start = Location(start_group, start_object);
     subscribe.subscriber_priority = 0x80;
     session->published_subscriptions_.emplace(
@@ -147,6 +149,10 @@
     session->next_request_id_ = id;
   }
 
+  static void set_next_incoming_request_id(MoqtSession* session, uint64_t id) {
+    session->next_incoming_request_id_ = id;
+  }
+
   static void set_peer_max_request_id(MoqtSession* session, uint64_t id) {
     session->peer_max_request_id_ = id;
   }
diff --git a/quiche/quic/moqt/test_tools/moqt_test_message.h b/quiche/quic/moqt/test_tools/moqt_test_message.h
index b0709ef..31c2521 100644
--- a/quiche/quic/moqt/test_tools/moqt_test_message.h
+++ b/quiche/quic/moqt/test_tools/moqt_test_message.h
@@ -465,7 +465,7 @@
 
   bool EqualFieldValues(MessageStructuredData& values) const override {
     auto cast = std::get<MoqtSubscribe>(values);
-    if (cast.subscribe_id != subscribe_.subscribe_id) {
+    if (cast.request_id != subscribe_.request_id) {
       QUIC_LOG(INFO) << "SUBSCRIBE subscribe ID mismatch";
       return false;
     }
@@ -485,6 +485,14 @@
       QUIC_LOG(INFO) << "SUBSCRIBE group order mismatch";
       return false;
     }
+    if (cast.forward != subscribe_.forward) {
+      QUIC_LOG(INFO) << "SUBSCRIBE forward mismatch";
+      return false;
+    }
+    if (cast.filter_type != subscribe_.filter_type) {
+      QUIC_LOG(INFO) << "SUBSCRIBE filter type mismatch";
+      return false;
+    }
     if (cast.start != subscribe_.start) {
       QUIC_LOG(INFO) << "SUBSCRIBE start mismatch";
       return false;
@@ -501,7 +509,7 @@
   }
 
   void ExpandVarints() override {
-    ExpandVarintsImpl("vvvv---v------vvvvv--vv-----");
+    ExpandVarintsImpl("vvvv---v-------vvvvv--vv-----");
   }
 
   MessageStructuredData structured_data() const override {
@@ -509,15 +517,16 @@
   }
 
  private:
-  uint8_t raw_packet_[31] = {
-      0x03, 0x00, 0x1c, 0x01, 0x02,  // id and alias
+  uint8_t raw_packet_[32] = {
+      0x03, 0x00, 0x1d, 0x01, 0x02,  // id and alias
       0x01, 0x03, 0x66, 0x6f, 0x6f,  // track_namespace = "foo"
       0x04, 0x61, 0x62, 0x63, 0x64,  // track_name = "abcd"
       0x20,                          // subscriber priority = 0x20
       0x02,                          // group order = descending
+      0x01,                          // forward = true
       0x03,                          // Filter type: Absolute Start
-      0x04,                          // start_group = 4 (relative previous)
-      0x01,                          // start_object = 1 (absolute)
+      0x04,                          // start_group = 4
+      0x01,                          // start_object = 1
       // No EndGroup or EndObject
       0x02,                                      // 2 parameters
       0x02, 0x67, 0x10,                          // delivery_timeout = 10000 ms
@@ -530,6 +539,8 @@
       /*full_track_name=*/FullTrackName({"foo", "abcd"}),
       /*subscriber_priority=*/0x20,
       /*group_order=*/MoqtDeliveryOrder::kDescending,
+      /*forward=*/true,
+      /*filter_type=*/MoqtFilterType::kAbsoluteStart,
       /*start=*/Location(4, 1),
       /*end_group=*/std::nullopt,
       VersionSpecificParameters(quic::QuicTimeDelta::FromMilliseconds(10000),