Implement draft-18 PADDING streams and datagrams. Does not included capability to send padding datagrams. PiperOrigin-RevId: 986870305
diff --git a/quiche/quic/moqt/moqt_messages.cc b/quiche/quic/moqt/moqt_messages.cc index 05ac688..6aeb9c5 100644 --- a/quiche/quic/moqt/moqt_messages.cc +++ b/quiche/quic/moqt/moqt_messages.cc
@@ -112,12 +112,8 @@ } std::string MoqtDataStreamTypeToString(MoqtDataStreamType type) { - if (type.IsPadding()) { - return "PADDING"; - } else if (type.IsFetch()) { - return "STREAM_HEADER_FETCH"; - } - return absl::StrCat("STREAM_HEADER_SUBGROUP_", type.value()); + return type.IsFetch() ? "STREAM_HEADER_FETCH" + : absl::StrCat("STREAM_HEADER_SUBGROUP_", type.value()); } std::string MoqtDatagramTypeToString(MoqtDatagramType type) {
diff --git a/quiche/quic/moqt/moqt_messages.h b/quiche/quic/moqt/moqt_messages.h index 9ed77fd..a64a33d 100644 --- a/quiche/quic/moqt/moqt_messages.h +++ b/quiche/quic/moqt/moqt_messages.h
@@ -37,10 +37,12 @@ // are not buffered by the parser). inline constexpr size_t kMaxMessageHeaderSize = 2048; +inline constexpr uint64_t kPaddingStreamType = 0x132B3E28; +inline constexpr uint64_t kPaddingDatagramType = 0x132B3E29; + class QUICHE_EXPORT MoqtDataStreamType { public: static constexpr uint64_t kFetch = 0x05; - static constexpr uint64_t kPadding = 0x26d3; static constexpr uint64_t kSubgroup = 0x10; static constexpr uint64_t kProperties = 0x01; static constexpr uint64_t kEndOfGroup = 0x08; @@ -54,7 +56,7 @@ // Factory functions. static std::optional<MoqtDataStreamType> FromValue(uint64_t value) { MoqtDataStreamType stream_type(value); - if (stream_type.IsFetch() || stream_type.IsPadding()) { + if (stream_type.IsFetch()) { return stream_type; } if (!(value & kSubgroup)) { @@ -71,7 +73,6 @@ return stream_type; } static MoqtDataStreamType Fetch() { return MoqtDataStreamType(kFetch); } - static MoqtDataStreamType Padding() { return MoqtDataStreamType(kPadding); } static MoqtDataStreamType Subgroup(uint64_t subgroup_id, uint64_t first_object_id, bool no_properties, bool default_priority, @@ -102,7 +103,6 @@ } MoqtDataStreamType(const MoqtDataStreamType& other) = default; bool IsFetch() const { return value_ == kFetch; } - bool IsPadding() const { return value_ == kPadding; } bool IsSubgroup() const { return value_ & kSubgroup; } bool IsSubgroupPresent() const { return IsSubgroup() && (value_ & kSubgroupId);
diff --git a/quiche/quic/moqt/moqt_parser.cc b/quiche/quic/moqt/moqt_parser.cc index 041b042..72dc57a 100644 --- a/quiche/quic/moqt/moqt_parser.cc +++ b/quiche/quic/moqt/moqt_parser.cc
@@ -508,7 +508,7 @@ } bool fin_read = false; std::optional<uint64_t> type = ReadMoqVarIntFromStream(*stream_, fin_read); - if (fin_read && type != MoqtDataStreamType::kPadding) { + if (fin_read && type != kPaddingStreamType) { // Besides padding streams, all other streams require some data after the // type byte. status_ = absl::InvalidArgumentError( @@ -1044,37 +1044,41 @@ visitor_.OnParsingError(MoqtError::kProtocolViolation, reason); } -std::optional<absl::string_view> ParseDatagram(absl::string_view data, - MoqtObject& object_metadata, - bool& use_default_priority) { +absl::StatusOr<absl::string_view> ParseDatagram(absl::string_view data, + MoqtObject& object_metadata, + bool& use_default_priority) { uint64_t type_raw, object_status_raw; absl::string_view properties; quic::QuicDataReader reader(data); object_metadata = MoqtObject(); - if (!reader.ReadMoqVarInt(&type_raw) || - !reader.ReadMoqVarInt(&object_metadata.track_alias) || - !reader.ReadMoqVarInt(&object_metadata.group_id)) { - return std::nullopt; + if (!reader.ReadMoqVarInt(&type_raw)) { + return absl::InvalidArgumentError("Failed to parse datagram"); } - + if (type_raw == kPaddingDatagramType) { + return absl::NotFoundError("Padding datagram"); + } std::optional<MoqtDatagramType> datagram_type = MoqtDatagramType::FromValue(type_raw); if (!datagram_type.has_value()) { - return std::nullopt; + return absl::InvalidArgumentError("Failed to parse datagram"); + } + if (!reader.ReadMoqVarInt(&object_metadata.track_alias) || + !reader.ReadMoqVarInt(&object_metadata.group_id)) { + return absl::InvalidArgumentError("Failed to parse datagram"); } if (datagram_type->end_of_group()) { object_metadata.object_status = MoqtObjectStatus::kEndOfGroup; if (datagram_type->has_status()) { QUICHE_BUG(Moqt_invalid_datagram_type) << "Invalid datagram type: " << type_raw; - return std::nullopt; + return absl::InvalidArgumentError("Failed to parse datagram"); } } else { object_metadata.object_status = MoqtObjectStatus::kNormal; } if (datagram_type->has_object_id()) { if (!reader.ReadMoqVarInt(&object_metadata.object_id)) { - return std::nullopt; + return absl::InvalidArgumentError("Failed to parse datagram"); } } else { object_metadata.object_id = 0; @@ -1083,22 +1087,22 @@ use_default_priority = datagram_type->has_default_priority(); if (!use_default_priority && !reader.ReadUInt8(&object_metadata.publisher_priority)) { - return std::nullopt; + return absl::InvalidArgumentError("Failed to parse datagram"); } if (datagram_type->has_properties()) { if (!reader.ReadStringPieceMoqVarInt(&properties)) { - return std::nullopt; + return absl::InvalidArgumentError("Failed to parse datagram"); } if (properties.empty()) { // This is a session error. - return std::nullopt; + return absl::InvalidArgumentError("Failed to parse datagram"); } object_metadata.properties = std::string(properties); } if (datagram_type->has_status()) { object_metadata.payload_length = 0; if (!reader.ReadMoqVarInt(&object_status_raw)) { - return std::nullopt; + return absl::InvalidArgumentError("Failed to parse datagram"); } object_metadata.object_status = IntegerToObjectStatus(object_status_raw); return ""; @@ -1567,10 +1571,6 @@ return; } type_ = *type; - if (type_.IsPadding()) { - next_input_ = kPadding; - return; - } if (type_.EndOfGroupInStream()) { contains_end_of_group_ = true; }
diff --git a/quiche/quic/moqt/moqt_parser.h b/quiche/quic/moqt/moqt_parser.h index cb9752c..a9c4ce5 100644 --- a/quiche/quic/moqt/moqt_parser.h +++ b/quiche/quic/moqt/moqt_parser.h
@@ -256,14 +256,16 @@ const quic::Perspective perspective_; }; -// Parses an MoQT datagram. Returns the payload bytes, or std::nullopt on error. +// Parses an MoQT datagram. Returns the payload bytes, or a status. If +// NotFoundError, it's a padding datagram and can be discarded. If it's +// InvalidArgumentError, the datagram is malformed. // The caller provides the whole datagram in `data`. The function puts the // object metadata in `object_metadata`. // If |use_default_priority| returns true, there was no reported // publisher_priority and the caller should use the default for the SUBSCRIBE. -std::optional<absl::string_view> ParseDatagram(absl::string_view data, - MoqtObject& object_metadata, - bool& use_default_priority); +absl::StatusOr<absl::string_view> ParseDatagram(absl::string_view data, + MoqtObject& object_metadata, + bool& use_default_priority); // Parser for MoQT unidirectional data stream. class QUICHE_EXPORT MoqtDataParser {
diff --git a/quiche/quic/moqt/moqt_parser_test.cc b/quiche/quic/moqt/moqt_parser_test.cc index 05d80b2..af5340b 100644 --- a/quiche/quic/moqt/moqt_parser_test.cc +++ b/quiche/quic/moqt/moqt_parser_test.cc
@@ -1298,10 +1298,10 @@ ObjectDatagramMessage message(datagram_type); MoqtObject object; bool use_default_priority; - std::optional<absl::string_view> payload = + absl::StatusOr<absl::string_view> payload = ParseDatagram(message.PacketSample(), object, use_default_priority); + QUICHE_ASSERT_OK(payload); EXPECT_EQ(use_default_priority, datagram_type.has_default_priority()); - ASSERT_TRUE(payload.has_value()); if (use_default_priority) { object.publisher_priority = message.publisher_priority(); } @@ -1309,9 +1309,9 @@ TestMessageBase::MessageStructuredData(object); EXPECT_TRUE(message.EqualFieldValues(object_metadata)); if (datagram_type.has_status()) { - EXPECT_EQ(payload, ""); + EXPECT_TRUE(payload->empty()); } else { - EXPECT_EQ(payload, "foo"); + EXPECT_EQ(*payload, "foo"); } } } @@ -1322,10 +1322,10 @@ message.ExpandVarints(); MoqtObject object; bool check_priority; - std::optional<absl::string_view> payload = + absl::StatusOr<absl::string_view> payload = ParseDatagram(message.PacketSample(), object, check_priority); + QUICHE_ASSERT_OK(payload); EXPECT_EQ(check_priority, datagram_type.has_default_priority()); - ASSERT_TRUE(payload.has_value()); if (check_priority) { object.publisher_priority = message.publisher_priority(); } @@ -1333,9 +1333,9 @@ TestMessageBase::MessageStructuredData(object); EXPECT_TRUE(message.EqualFieldValues(object_metadata)); if (datagram_type.has_status()) { - EXPECT_EQ(payload, ""); + EXPECT_TRUE(payload->empty()); } else { - EXPECT_EQ(payload, "foo"); + EXPECT_EQ(*payload, "foo"); } } } @@ -1344,9 +1344,10 @@ char payload[] = {0x33, 0x10, 0x20}; MoqtObject object; bool check_priority; - EXPECT_EQ(ParseDatagram(absl::string_view(payload, sizeof(payload)), object, - check_priority), - std::nullopt); + EXPECT_TRUE(absl::IsInvalidArgument( + ParseDatagram(absl::string_view(payload, sizeof(payload)), object, + check_priority) + .status())); } TEST_F(MoqtMessageSpecificTest, TruncatedDatagram) { @@ -1355,17 +1356,17 @@ message.set_wire_image_size(4); MoqtObject object; bool check_priority; - EXPECT_EQ(ParseDatagram(message.PacketSample(), object, check_priority), - std::nullopt); + EXPECT_TRUE(absl::IsInvalidArgument( + ParseDatagram(message.PacketSample(), object, check_priority).status())); } TEST_F(MoqtMessageSpecificTest, VeryTruncatedDatagram) { char message = 0x40; MoqtObject object; bool check_priority; - EXPECT_EQ(ParseDatagram(absl::string_view(&message, sizeof(message)), object, - check_priority), - std::nullopt); + EXPECT_TRUE(absl::IsInvalidArgument( + ParseDatagram(absl::string_view(&message, 1), object, check_priority) + .status())); } TEST_F(MoqtMessageSpecificTest, SubscribeOkInvalidDeliveryOrder) { @@ -1434,21 +1435,6 @@ HasSubstr("End object comes before start object in FETCH")); } -TEST_F(MoqtMessageSpecificTest, PaddingStream) { - MoqtParserTestVisitor visitor; - webtransport::test::InMemoryStream stream(/*stream_id=*/0); - MoqtDataParser parser(&stream, &visitor); - std::string buffer(32, '\0'); - quic::QuicDataWriter writer(buffer.size(), buffer.data()); - ASSERT_TRUE(writer.WriteMoqVarInt(MoqtDataStreamType::Padding().value())); - for (int i = 0; i < 100; ++i) { - stream.Receive(buffer, false); - parser.ReadAllData(); - ASSERT_EQ(visitor.messages_received(), 0); - ASSERT_EQ(visitor.parsing_error(), std::nullopt); - } -} - // All messages with TrackNamespace use ReadTrackNamespace. Use // PUBLISH_NAMESPACE. TEST_F(MoqtMessageSpecificTest, EmptyNamespace) { @@ -1874,10 +1860,12 @@ TEST_F(MoqtMessageSpecificTest, StreamTypeParserFinForPadding) { webtransport::test::InMemoryStream stream(/*stream_id=*/0); MoqtStreamTypeParser type_parser(&stream); - stream.Receive("\xa6\xd3", true); + char buffer[16]; + quic::QuicDataWriter writer(sizeof(buffer), buffer); + ASSERT_TRUE(writer.WriteMoqVarInt(kPaddingStreamType)); + stream.Receive(writer.data(), true); absl::StatusOr<uint64_t> type = type_parser.ReadStreamType(); - EXPECT_THAT( - type, IsOkAndHolds(static_cast<uint64_t>(MoqtDataStreamType::kPadding))); + EXPECT_THAT(type, IsOkAndHolds(kPaddingStreamType)); } TEST_F(MoqtMessageSpecificTest, StreamTypeParserMovedFrom) {
diff --git a/quiche/quic/moqt/moqt_probe_manager.cc b/quiche/quic/moqt/moqt_probe_manager.cc index cb2e53e..12af9b3 100644 --- a/quiche/quic/moqt/moqt_probe_manager.cc +++ b/quiche/quic/moqt/moqt_probe_manager.cc
@@ -75,8 +75,8 @@ } if (!header_sent_) { - absl::StatusOr<std::string> serialized_header = quiche::SerializeIntoString( - quiche::WireMoqVarInt(MoqtDataStreamType::Padding().value())); + absl::StatusOr<std::string> serialized_header = + quiche::SerializeIntoString(quiche::WireMoqVarInt(kPaddingStreamType)); if (!serialized_header.ok()) { QUICHE_BUG(ProbeStreamVisitor_HeaderFailure) << "Failed to serialize the padding stream header";
diff --git a/quiche/quic/moqt/moqt_probe_manager_test.cc b/quiche/quic/moqt/moqt_probe_manager_test.cc index 9f97832..d2b75fd 100644 --- a/quiche/quic/moqt/moqt_probe_manager_test.cc +++ b/quiche/quic/moqt/moqt_probe_manager_test.cc
@@ -40,8 +40,8 @@ using ::testing::_; using ::testing::Return; -// Two-byte varint. -constexpr size_t kProbeStreamHeaderSize = 2; +// Five-byte varint. +constexpr size_t kProbeStreamHeaderSize = 5; class MockStream : public webtransport::test::MockStream { public:
diff --git a/quiche/quic/moqt/moqt_session.cc b/quiche/quic/moqt/moqt_session.cc index d505bca..847bf23 100644 --- a/quiche/quic/moqt/moqt_session.cc +++ b/quiche/quic/moqt/moqt_session.cc
@@ -202,9 +202,12 @@ void MoqtSession::OnDatagramReceived(absl::string_view datagram) { MoqtObject message; bool use_default_priority; - std::optional<absl::string_view> payload = + absl::StatusOr<absl::string_view> payload = ParseDatagram(datagram, message, use_default_priority); - if (!payload.has_value()) { + if (!payload.ok()) { + if (absl::IsNotFound(payload.status())) { + return; // It's PADDING. + } Error(MoqtError::kProtocolViolation, "Malformed datagram received"); return; } @@ -1253,6 +1256,14 @@ temp_stream->OnCanRead(); return; } + if (*type == kPaddingStreamType) { + auto padding_stream = std::make_unique<IncomingPaddingStream>(stream_); + IncomingPaddingStream* temp_stream = padding_stream.get(); + // The line below destroys `this`. + stream_->SetVisitor(std::move(padding_stream)); + temp_stream->OnCanRead(); + return; + } auto data_stream = std::make_unique<IncomingDataStream>( std::move(parser_), session, session->callbacks_.clock); IncomingDataStream* temp_stream = data_stream.get();
diff --git a/quiche/quic/moqt/moqt_session_test.cc b/quiche/quic/moqt/moqt_session_test.cc index 94012af..715cba5 100644 --- a/quiche/quic/moqt/moqt_session_test.cc +++ b/quiche/quic/moqt/moqt_session_test.cc
@@ -20,6 +20,7 @@ #include "absl/strings/string_view.h" #include "absl/types/span.h" #include "quiche/quic/core/quic_data_reader.h" +#include "quiche/quic/core/quic_data_writer.h" #include "quiche/quic/core/quic_time.h" #include "quiche/quic/core/quic_types.h" #include "quiche/quic/moqt/moqt_bidi_stream.h" @@ -529,6 +530,40 @@ session_.OnIncomingUnidirectionalStreamAvailable(); } +TEST_F(MoqtSessionTest, IncomingPaddingStream) { + webtransport::test::InMemoryStream padding_stream(/*stream_id=*/10); + char buffer[64]; + quic::QuicDataWriter writer(sizeof(buffer), buffer); + ASSERT_TRUE(writer.WriteMoqVarInt(kPaddingStreamType)); + ASSERT_TRUE(writer.WriteStringPiece("padding_data")); + padding_stream.Receive(writer.data(), true); + EXPECT_CALL(mock_session_, AcceptIncomingUnidirectionalStream()) + .WillOnce(Return(&padding_stream)) + .WillOnce(Return(nullptr)); + session_.OnIncomingUnidirectionalStreamAvailable(); + EXPECT_EQ(padding_stream.ReadableBytes(), 0); + + padding_stream.Receive("even_more_padding", true); + EXPECT_EQ(padding_stream.ReadableBytes(), 0); +} + +TEST_F(MoqtSessionTest, PaddingDatagramDiscarded) { + EXPECT_CALL(mock_session_, CloseSession).Times(0); + EXPECT_CALL(session_callbacks_.session_terminated_callback, Call).Times(0); + EXPECT_CALL(remote_track_visitor_, OnObjectFragment).Times(0); + + char buffer[64]; + quic::QuicDataWriter writer(sizeof(buffer), buffer); + ASSERT_TRUE(writer.WriteMoqVarInt(kPaddingDatagramType)); + ASSERT_TRUE(writer.WriteStringPiece("padding_data")); + session_.OnDatagramReceived(writer.data()); + + char empty_buffer[16]; + quic::QuicDataWriter empty_writer(sizeof(empty_buffer), empty_buffer); + ASSERT_TRUE(empty_writer.WriteMoqVarInt(kPaddingDatagramType)); + session_.OnDatagramReceived(empty_writer.data()); +} + TEST_F(MoqtSessionTest, Error) { bool reported_error = false; EXPECT_CALL(
diff --git a/quiche/quic/moqt/moqt_uni_stream.cc b/quiche/quic/moqt/moqt_uni_stream.cc index 866e58d..f8776cb 100644 --- a/quiche/quic/moqt/moqt_uni_stream.cc +++ b/quiche/quic/moqt/moqt_uni_stream.cc
@@ -539,15 +539,12 @@ void IncomingDataStream::OnCanRead() { if (!parser_.stream_type().has_value()) { + // This only happens in test. parser_.ReadStreamType(); if (!parser_.stream_type().has_value()) { return; } } - if (parser_.stream_type()->IsPadding()) { - (void)stream_->SkipBytes(stream_->ReadableBytes()); - return; - } bool knew_track_alias = parser_.track_alias().has_value(); if (!knew_track_alias) { parser_.ReadTrackAlias(); @@ -606,4 +603,12 @@ session_->Error(error_code, absl::StrCat("Parse error: ", reason)); } +IncomingPaddingStream::IncomingPaddingStream( + webtransport::Stream* absl_nonnull stream) + : stream_(stream) {} + +void IncomingPaddingStream::OnCanRead() { + (void)stream_->SkipBytes(stream_->ReadableBytes()); +} + } // namespace moqt
diff --git a/quiche/quic/moqt/moqt_uni_stream.h b/quiche/quic/moqt/moqt_uni_stream.h index 28a3e4f..0e555c6 100644 --- a/quiche/quic/moqt/moqt_uni_stream.h +++ b/quiche/quic/moqt/moqt_uni_stream.h
@@ -285,6 +285,24 @@ const quic::QuicClock* absl_nonnull clock_; }; +class QUICHE_EXPORT IncomingPaddingStream : public webtransport::StreamVisitor { + public: + explicit IncomingPaddingStream(webtransport::Stream* absl_nonnull stream); + ~IncomingPaddingStream() override = default; + + // webtransport::StreamVisitor implementation. + void OnCanRead() override; + void OnCanWrite() override {} + void OnResetStreamReceived(webtransport::StreamErrorCode) override {} + void OnStopSendingReceived(webtransport::StreamErrorCode) override {} + void OnWriteSideInDataRecvdState() override {} + + webtransport::Stream* stream() const { return stream_; } + + private: + webtransport::Stream* absl_nonnull stream_; +}; + } // namespace moqt #endif // QUICHE_QUIC_MOQT_MOQT_UNI_STREAM_H_
diff --git a/quiche/quic/moqt/moqt_uni_stream_test.cc b/quiche/quic/moqt/moqt_uni_stream_test.cc index 61d7ee8..bb981dc 100644 --- a/quiche/quic/moqt/moqt_uni_stream_test.cc +++ b/quiche/quic/moqt/moqt_uni_stream_test.cc
@@ -795,6 +795,20 @@ stream_->OnCanRead(); } +TEST(IncomingPaddingStreamTest, ConsumesAllData) { + webtransport::test::InMemoryStream stream(/*stream_id=*/0); + IncomingPaddingStream padding_stream(&stream); + stream.Receive("hello world", false); + EXPECT_EQ(stream.ReadableBytes(), 11); + padding_stream.OnCanRead(); + EXPECT_EQ(stream.ReadableBytes(), 0); + + stream.Receive("more padding", true); + EXPECT_EQ(stream.ReadableBytes(), 12); + padding_stream.OnCanRead(); + EXPECT_EQ(stream.ReadableBytes(), 0); +} + } // namespace } // namespace moqt::test
diff --git a/quiche/quic/moqt/test_tools/moqt_test_message.h b/quiche/quic/moqt/test_tools/moqt_test_message.h index 3781e3b..750e1f8 100644 --- a/quiche/quic/moqt/test_tools/moqt_test_message.h +++ b/quiche/quic/moqt/test_tools/moqt_test_message.h
@@ -1670,9 +1670,6 @@ static inline std::unique_ptr<TestMessageBase> CreateTestDataStream( MoqtDataStreamType type) { - if (type.IsPadding()) { - return nullptr; - } if (type.IsFetch()) { return std::make_unique<StreamHeaderFetchMessage>(); }