Use delta-encoded group IDs in MOQT fetch streams. There is also some apparently buggy logic in diff-encoded object IDs as well, now fixed. PiperOrigin-RevId: 996662817
diff --git a/quiche/quic/moqt/moqt_fetch_stream.cc b/quiche/quic/moqt/moqt_fetch_stream.cc index 06f8dcc..5cfd9b6 100644 --- a/quiche/quic/moqt/moqt_fetch_stream.cc +++ b/quiche/quic/moqt/moqt_fetch_stream.cc
@@ -102,7 +102,10 @@ // Interactions with the task will now be mediated through the uni stream. // If the uni stream closes, so will the bidi stream. task_ = nullptr; - absl::down_cast<IncomingDataStream*>(stream)->set_fetch_task(task); + auto* data_stream = absl::down_cast<IncomingDataStream*>(stream); + data_stream->set_group_order( + const_parameters().group_order.value_or(MoqtDeliveryOrder::kAscending)); + data_stream->set_fetch_task(task); } void MoqtFetchRequestStream::OnStreamClosed(absl::Status status,
diff --git a/quiche/quic/moqt/moqt_framer.cc b/quiche/quic/moqt/moqt_framer.cc index 235b8e5..7810a07 100644 --- a/quiche/quic/moqt/moqt_framer.cc +++ b/quiche/quic/moqt/moqt_framer.cc
@@ -627,13 +627,27 @@ MoqtFetchSerialization(message, *previous_object_in_stream); } if (serialization.has_group_id()) { - group_id = message.group_id; + if (is_first_in_stream) { + group_id = message.group_id; + } else { + group_id = + (message.group_id > previous_object_in_stream->location.group) + ? (message.group_id - + previous_object_in_stream->location.group - 1) + : (previous_object_in_stream->location.group - + message.group_id - 1); + } } if (serialization.has_subgroup_id()) { subgroup_id = message.subgroup_id; } if (serialization.has_object_id()) { - object_id = message.object_id; + if (is_first_in_stream || serialization.has_group_id()) { + object_id = message.object_id; + } else { + object_id = + message.object_id - previous_object_in_stream->location.object; + } } if (serialization.has_priority()) { publisher_priority = message.publisher_priority;
diff --git a/quiche/quic/moqt/moqt_parser.cc b/quiche/quic/moqt/moqt_parser.cc index 93955e2..0c38955 100644 --- a/quiche/quic/moqt/moqt_parser.cc +++ b/quiche/quic/moqt/moqt_parser.cc
@@ -1275,7 +1275,11 @@ if (fetch_serialization_.has_object_id()) { return kObjectId; } - ++metadata_.object_id; + if (!quiche::SafeIncrementBy<uint64_t>(metadata_.object_id, 1)) { + ParseError("Integer overflow when parsing object ID"); + return kFailed; + } + last_object_id_ = metadata_.object_id; [[fallthrough]]; case kObjectId: if (fetch_serialization_.end_of_non_existent_range() || @@ -1413,17 +1417,40 @@ } case kGroupId: { - std::optional<uint64_t> value_read = ReadMoqVarIntNoFin(); - if (value_read.has_value()) { - if (type_.IsFetch() || - !fetch_serialization_.end_of_non_existent_range() || - !fetch_serialization_.end_of_unknown_range()) { - // Do not record range indicator group IDs because it will corrupt - // references to the previous object. - metadata_.group_id = *value_read; - } - next_input_ = AdvanceParserState(); + if (type_.IsFetch() && !group_order_.has_value()) { + // Check before reading the group ID to avoid wrecking parser state. + QUICHE_BUG(MoqtDataParser_no_group_order) + << "Data parser group order not set"; + return; } + std::optional<uint64_t> value_read = ReadMoqVarIntNoFin(); + if (!value_read.has_value()) { + return; + } + if (type_.IsFetch() && last_group_id_.has_value()) { + std::optional<uint64_t> new_group_id; + if (!group_order_.has_value()) { + // Exit to satisfy ClangTidy. + return; + } + if (*group_order_ == MoqtDeliveryOrder::kAscending) { + new_group_id = + quiche::SafeSum<uint64_t>({*value_read, *last_group_id_, 1}); + } else { + new_group_id = (*value_read < *last_group_id_) + ? (*last_group_id_ - *value_read - 1) + : std::optional<uint64_t>(); + } + if (!new_group_id.has_value()) { + ParseError("Integer overflow when parsing group ID"); + return; + } + metadata_.group_id = *new_group_id; + } else { + metadata_.group_id = *value_read; + } + last_group_id_ = metadata_.group_id; + next_input_ = AdvanceParserState(); return; } @@ -1447,27 +1474,28 @@ case kObjectId: { std::optional<uint64_t> value_read = ReadMoqVarIntNoFin(); - if (value_read.has_value()) { - if (type_.IsFetch() || - !fetch_serialization_.end_of_non_existent_range() || - !fetch_serialization_.end_of_unknown_range()) { - // Do not record range indicator object IDs because it will corrupt - // references to the previous object. - if (type_.IsSubgroup() && last_object_id_.has_value()) { - std::optional<uint64_t> new_object_id = - quiche::SafeSum<uint64_t>({*value_read, *last_object_id_, 1}); - if (!new_object_id.has_value()) { - ParseError("Integer overflow when parsing object ID"); - return; - } - metadata_.object_id = *new_object_id; - } else { - metadata_.object_id = *value_read; - } - } - last_object_id_ = metadata_.object_id; - next_input_ = AdvanceParserState(); + if (!value_read.has_value()) { + return; } + if (last_object_id_.has_value() && + (type_.IsSubgroup() || !fetch_serialization_.has_group_id())) { + if (type_.IsFetch() && *value_read == 0) { + ParseError("Duplicate object ID in FETCH stream"); + return; + } + std::optional<uint64_t> new_object_id = quiche::SafeSum<uint64_t>( + {*value_read, *last_object_id_, + type_.IsSubgroup() ? uint64_t{1} : uint64_t{0}}); + if (!new_object_id.has_value()) { + ParseError("Integer overflow when parsing object ID"); + return; + } + metadata_.object_id = *new_object_id; + } else { + metadata_.object_id = *value_read; + } + last_object_id_ = metadata_.object_id; + next_input_ = AdvanceParserState(); // TODO(martinduke): Report something if the fetch serialization is an end // of range indicator. return;
diff --git a/quiche/quic/moqt/moqt_parser.h b/quiche/quic/moqt/moqt_parser.h index 4fdc359..c6ede94 100644 --- a/quiche/quic/moqt/moqt_parser.h +++ b/quiche/quic/moqt/moqt_parser.h
@@ -277,6 +277,13 @@ void ReadTrackAlias(); void ReadAtMostOneObject(); + void SetGroupOrder(MoqtDeliveryOrder group_order) { + if (group_order_.has_value()) { + return; + } + group_order_ = group_order; + } + // Returns the type of the unidirectional stream, if already known. std::optional<MoqtDataStreamType> stream_type() const { if (next_input_ == kStreamType) { @@ -362,10 +369,12 @@ std::string buffered_message_; + std::optional<MoqtDeliveryOrder> group_order_; MoqtDataStreamType type_; MoqtFetchSerialization fetch_serialization_; NextInput next_input_ = kStreamType; MoqtObject metadata_; + std::optional<uint64_t> last_group_id_; std::optional<uint64_t> last_object_id_; size_t payload_length_remaining_ = 0; size_t num_objects_read_ = 0;
diff --git a/quiche/quic/moqt/moqt_parser_fuzz_test.cc b/quiche/quic/moqt/moqt_parser_fuzz_test.cc index 5da2857..ae302bd 100644 --- a/quiche/quic/moqt/moqt_parser_fuzz_test.cc +++ b/quiche/quic/moqt/moqt_parser_fuzz_test.cc
@@ -8,7 +8,9 @@ #include "absl/status/status.h" #include "absl/status/statusor.h" #include "absl/strings/string_view.h" +#include "quiche/quic/core/quic_types.h" #include "quiche/quic/moqt/moqt_parser.h" +#include "quiche/quic/moqt/moqt_priority.h" #include "quiche/quic/moqt/moqt_session_interface.h" #include "quiche/quic/moqt/test_tools/moqt_parser_test_visitor.h" #include "quiche/common/platform/api/quiche_fuzztest.h" @@ -28,6 +30,7 @@ MoqtControlMessageParser control_message_parser( kDefaultMoqtVersion, uses_web_transport, perspective); MoqtDataParser data_parser(&stream, &visitor); + data_parser.SetGroupOrder(MoqtDeliveryOrder::kAscending); if (is_data_stream) { stream.Receive(stream_data, /*fin=*/fin);
diff --git a/quiche/quic/moqt/moqt_parser_test.cc b/quiche/quic/moqt/moqt_parser_test.cc index 06859f8..e34575e 100644 --- a/quiche/quic/moqt/moqt_parser_test.cc +++ b/quiche/quic/moqt/moqt_parser_test.cc
@@ -26,6 +26,7 @@ #include "quiche/quic/moqt/moqt_key_value_pair.h" #include "quiche/quic/moqt/moqt_messages.h" #include "quiche/quic/moqt/moqt_names.h" +#include "quiche/quic/moqt/moqt_priority.h" #include "quiche/quic/moqt/moqt_session_interface.h" #include "quiche/quic/moqt/moqt_types.h" #include "quiche/quic/moqt/test_tools/moqt_framer_utils.h" @@ -156,6 +157,7 @@ // The default object has priority 0x07, so setting this will let the // parser set the correct value when absent. data_parser_.set_default_publisher_priority(0x07); + data_parser_.SetGroupOrder(MoqtDeliveryOrder::kAscending); } bool IsDataStream() const { @@ -1570,7 +1572,9 @@ class MoqtDataParserStateMachineTest : public quic::test::QuicTest { protected: MoqtDataParserStateMachineTest() - : stream_(/*stream_id=*/0), parser_(&stream_, &visitor_) {} + : stream_(/*stream_id=*/0), parser_(&stream_, &visitor_) { + parser_.SetGroupOrder(MoqtDeliveryOrder::kAscending); + } webtransport::test::InMemoryStream stream_; MoqtParserTestVisitor visitor_; @@ -1634,6 +1638,7 @@ MoqtParserTestVisitor visitor; webtransport::test::InMemoryStream stream(/*stream_id=*/0); MoqtDataParser parser(&stream, &visitor); + parser.SetGroupOrder(MoqtDeliveryOrder::kAscending); StreamHeaderFetchMessage header; StreamMiddlerFetchMessage middler(serialization); stream.Receive(header.PacketSample()); @@ -1661,6 +1666,7 @@ MoqtParserTestVisitor visitor; webtransport::test::InMemoryStream stream(/*stream_id=*/0); MoqtDataParser parser(&stream, &visitor); + parser.SetGroupOrder(MoqtDeliveryOrder::kAscending); stream.Receive(absl::string_view(data, sizeof(data))); parser.ReadStreamType(); parser.ReadAtMostOneObject(); @@ -1679,6 +1685,7 @@ MoqtParserTestVisitor visitor; webtransport::test::InMemoryStream stream(/*stream_id=*/0); MoqtDataParser parser(&stream, &visitor); + parser.SetGroupOrder(MoqtDeliveryOrder::kAscending); stream.Receive(absl::string_view(data, sizeof(data))); parser.ReadStreamType(); parser.ReadAtMostOneObject(); @@ -1788,6 +1795,79 @@ EXPECT_FALSE(visitor_.last_message()->first_object_in_subgroup.has_value()); } +TEST_F(MoqtDataParserStateMachineTest, FetchDescendingGroupOrder) { + webtransport::test::InMemoryStream stream(/*stream_id=*/0); + MoqtParserTestVisitor visitor; + MoqtDataParser parser(&stream, &visitor); + parser.SetGroupOrder(MoqtDeliveryOrder::kDescending); + + // Header has group_id = 5, object_id = 6. + StreamHeaderFetchMessage header; + stream.Receive(header.PacketSample()); + // Next object: kHasGroupId | kIsDatagram = 0x48, group delta = 1 -> group = + // 5 - (1 + 1) = 3, omitted object_id -> object_id = 7, payload = "bar". + // Third object: kHasGroupId | kIsDatagram = 0x48, group delta = 3 -> + // underflow (3 - (3 + 1) < 0). + char data[] = {0x48, 0x01, 0x03, 'b', 'a', 'r', 0x48, 0x03}; + stream.Receive(absl::string_view(data, sizeof(data))); + + parser.ReadStreamType(); + parser.ReadAtMostOneObject(); + ASSERT_EQ(visitor.messages_received(), 1); + EXPECT_EQ(visitor.last_message()->group_id, 5); + + parser.ReadAtMostOneObject(); + ASSERT_EQ(visitor.messages_received(), 2); + EXPECT_EQ(visitor.last_message()->group_id, 3); + EXPECT_EQ(visitor.last_message()->object_id, 7); + + parser.ReadAtMostOneObject(); + EXPECT_EQ(visitor.parsing_error(), "Integer overflow when parsing group ID"); +} + +TEST_F(MoqtDataParserStateMachineTest, + FetchObjectIdDeltaAfterImplicitIncrement) { + // Header has group_id = 5, object_id = 6. + StreamHeaderFetchMessage header; + stream_.Receive(header.PacketSample()); + // Object 2: kIsDatagram = 0x40 (omits group ID and object ID) -> group_id = + // 5, object_id = 7, payload = "bar". + // Object 3: kIsDatagram | kHasObjectId = 0x44, object ID delta = 3 -> + // group_id = 5, object_id = 7 + 3 = 10, payload = "baz". + char data[] = {0x40, 0x03, 'b', 'a', 'r', 0x44, 0x03, 0x03, 'b', 'a', 'z'}; + stream_.Receive(absl::string_view(data, sizeof(data)), /*fin=*/true); + + parser_.ReadAllData(); + ASSERT_EQ(visitor_.messages_received(), 3); + EXPECT_EQ(visitor_.last_message()->group_id, 5); + EXPECT_EQ(visitor_.last_message()->object_id, 10); + EXPECT_EQ(visitor_.parsing_error(), std::nullopt); +} + +TEST_F(MoqtDataParserStateMachineTest, FetchZeroObjectIdDelta) { + // Header has group_id = 5, object_id = 6. + StreamHeaderFetchMessage header; + stream_.Receive(header.PacketSample()); + // Object 2: kIsDatagram | kHasGroupId | kHasObjectId = 0x4c, group delta = 0, + // explicit object ID = 0 -> valid (group_id = 6, object_id = 0). + // Object 3: kIsDatagram | kHasObjectId = 0x44, object ID delta = 0 -> + // duplicate object ID in same group -> parse error. + char data[] = {0x4c, 0x00, 0x00, 0x03, 'b', 'a', 'r', 0x44, 0x00}; + stream_.Receive(absl::string_view(data, sizeof(data))); + + parser_.ReadStreamType(); + parser_.ReadAtMostOneObject(); + ASSERT_EQ(visitor_.messages_received(), 1); + + parser_.ReadAtMostOneObject(); + ASSERT_EQ(visitor_.messages_received(), 2); + EXPECT_EQ(visitor_.last_message()->group_id, 6); + EXPECT_EQ(visitor_.last_message()->object_id, 0); + + parser_.ReadAtMostOneObject(); + EXPECT_EQ(visitor_.parsing_error(), "Duplicate object ID in FETCH stream"); +} + TEST_F(MoqtMessageSpecificTest, StreamTypeParserToControlStream) { webtransport::test::InMemoryStream stream(/*stream_id=*/0); MoqtStreamTypeParser type_parser(&stream);
diff --git a/quiche/quic/moqt/moqt_uni_stream.h b/quiche/quic/moqt/moqt_uni_stream.h index e93b620..7b41279 100644 --- a/quiche/quic/moqt/moqt_uni_stream.h +++ b/quiche/quic/moqt/moqt_uni_stream.h
@@ -266,6 +266,9 @@ void MaybeReadOneObject(); virtual void set_fetch_task(UpstreamFetchTask* fetch_task); + void set_group_order(MoqtDeliveryOrder group_order) { + parser_.SetGroupOrder(group_order); + } private: friend class test::MoqtSessionPeer;
diff --git a/quiche/quic/moqt/test_tools/moqt_test_message.h b/quiche/quic/moqt/test_tools/moqt_test_message.h index d4db588..488bbbb 100644 --- a/quiche/quic/moqt/test_tools/moqt_test_message.h +++ b/quiche/quic/moqt/test_tools/moqt_test_message.h
@@ -531,7 +531,7 @@ size_t length = 0; raw_packet_[length++] = static_cast<uint8_t>(serialization.value()); if (serialization.has_group_id()) { - raw_packet_[length++] = 0x06; // group ID + raw_packet_[length++] = 0x00; // group ID delta = 0 (group ID = 6) object_.group_id = 6; } if (serialization.zero_subgroup_id()) { @@ -550,7 +550,11 @@ object_.subgroup_id = std::nullopt; } // If prior_subgroup_id, subgroup_id is already set properly. if (serialization.has_object_id()) { - raw_packet_[length++] = 0x0a; + if (serialization.has_group_id()) { + raw_packet_[length++] = 0x0a; // explicit object ID = 10 + } else { + raw_packet_[length++] = 0x04; // object ID delta = 4 (object ID = 10) + } object_.object_id = 10; } else { ++object_.object_id;