Add MoqtStreamTypeParser that can be later converted into either a data stream parser or a control message parser. Since SETUP is going to become a pair of unidirectional streams, we can no longer select the parser solely based on the stream type. PiperOrigin-RevId: 959212215
diff --git a/quiche/quic/moqt/moqt_bidi_stream.cc b/quiche/quic/moqt/moqt_bidi_stream.cc index 9509c4e..b5b7740 100644 --- a/quiche/quic/moqt/moqt_bidi_stream.cc +++ b/quiche/quic/moqt/moqt_bidi_stream.cc
@@ -23,7 +23,7 @@ namespace moqt { void MoqtBidiStreamBase::OnCanRead() { - if (stream_parser_ == nullptr) { + if (!stream_parser_.has_value()) { QUICHE_BUG(MoqtBidiStreamBase_OnCanRead_no_stream) << "OnCanRead() called when no stream is bound"; return; @@ -47,7 +47,7 @@ } void MoqtBidiStreamBase::OnCanWrite() { - if (stream_parser_ == nullptr) { + if (!stream_parser_.has_value()) { QUICHE_BUG(MoqtBidiStreamBase_OnCanWrite_no_stream) << "OnCanWrite() called when no stream is bound"; return;
diff --git a/quiche/quic/moqt/moqt_bidi_stream.h b/quiche/quic/moqt/moqt_bidi_stream.h index 68b55f7..13fc2d2 100644 --- a/quiche/quic/moqt/moqt_bidi_stream.h +++ b/quiche/quic/moqt/moqt_bidi_stream.h
@@ -54,17 +54,16 @@ ~MoqtBidiStreamBase() = default; // Binds a WebTransport stream associated with `parser` to this object. - void BindStream( - std::unique_ptr<MoqtControlStreamParser> absl_nonnull parser) { - QUICHE_DCHECK(stream_parser_ == nullptr); - stream_parser_ = std::move(parser); + void BindStream(MoqtStreamTypeParser parser) { + QUICHE_DCHECK(!stream_parser_.has_value()); + stream_parser_.emplace(std::move(parser)); outgoing_message_queue_.SetStream(stream_parser_->stream()); OnStreamBound(); } // Binds a WebTransport stream `stream` to this object. void BindStream(webtransport::Stream* absl_nonnull stream) { - QUICHE_DCHECK(stream_parser_ == nullptr); - stream_parser_ = std::make_unique<MoqtControlStreamParser>(stream); + QUICHE_DCHECK(!stream_parser_.has_value()); + stream_parser_.emplace(stream); outgoing_message_queue_.SetStream(stream); OnStreamBound(); } @@ -115,9 +114,8 @@ Detach(); } void Reset(webtransport::StreamErrorCode error) { - webtransport::Stream* stream = stream_parser_->stream(); - if (stream != nullptr) { - stream->ResetWithUserCode(error); + if (stream() != nullptr) { + stream()->ResetWithUserCode(error); } Detach(); } @@ -154,19 +152,21 @@ // Terminates the MoQT session due to a fatal error encountered. void OnFatalError(absl::Status status); - MoqtControlStreamParser* stream_parser() { return stream_parser_.get(); } + MoqtControlStreamParser* stream_parser() { + return stream_parser_.has_value() ? &*stream_parser_ : nullptr; + } const MoqtControlMessageParser& message_parser() const { return message_parser_; } webtransport::Stream* stream() const { - return stream_parser_ != nullptr ? stream_parser_->stream() : nullptr; + return stream_parser_.has_value() ? stream_parser_->stream() : nullptr; } private: friend class test::MoqtBidiStreamTestWrapper; MoqtFramer* absl_nonnull framer_; - std::unique_ptr<MoqtControlStreamParser> absl_nullable stream_parser_; + std::optional<MoqtControlStreamParser> stream_parser_; MoqtControlMessageParser message_parser_; MoqtControlMessageQueue outgoing_message_queue_; MoqtRequestUpdateQueue request_update_queue_;
diff --git a/quiche/quic/moqt/moqt_parser.cc b/quiche/quic/moqt/moqt_parser.cc index 3eec6fe..c7aa980 100644 --- a/quiche/quic/moqt/moqt_parser.cc +++ b/quiche/quic/moqt/moqt_parser.cc
@@ -488,6 +488,54 @@ return status; } +MoqtStreamTypeParser::MoqtStreamTypeParser( + MoqtStreamTypeParser&& other) noexcept + : stream_(other.stream_), type_(other.type_), status_(other.status_) { + other.status_ = absl::InternalError("Accessing a moved-from parser"); +} + +MoqtStreamTypeParser& MoqtStreamTypeParser::operator=( + MoqtStreamTypeParser&& other) noexcept { + if (this != &other) { + stream_ = other.stream_; + type_ = other.type_; + status_ = other.status_; + other.status_ = absl::InternalError("Accessing a moved-from parser"); + } + return *this; +} + +absl::StatusOr<uint64_t> MoqtStreamTypeParser::ReadStreamType() { + if (!status_.ok()) { + return status_; + } + if (type_.has_value()) { + // The type has already been read; the rest of the stream should be only + // read by the next parser. + return *type_; + } + bool fin_read = false; + std::optional<uint64_t> type = ReadMoqVarIntFromStream(*stream_, fin_read); + if (fin_read && type != MoqtDataStreamType::kPadding) { + // Besides padding streams, all other streams require some data after the + // type byte. + status_ = absl::InvalidArgumentError( + "FIN received before or immediately after the stream type"); + return status_; + } + if (!type.has_value()) { + return absl::UnavailableError("No complete message available"); + } + type_ = *type; + return *type_; +} + +MoqtControlStreamParser::MoqtControlStreamParser( + MoqtStreamTypeParser type_parser) + : stream_(*type_parser.stream()), + current_message_type_(type_parser.stream_type()), + fin_read_(false) {} + absl::StatusOr<MoqtRawControlMessage> MoqtControlStreamParser::ReadNextMessage() { if (error_encountered_ || fin_read_) { @@ -509,24 +557,6 @@ return result; } -absl::StatusOr<MoqtMessageType> -MoqtControlStreamParser::ReadFirstMessageType() { - if (first_message_type_.has_value()) { - return static_cast<MoqtMessageType>(*first_message_type_); - } - if (error_encountered_ || fin_read_) { - return absl::FailedPreconditionError( - "Trying to read from a control stream after an error or an EOF " - "occurred."); - } - absl::Status read_status = ReadMessageType(); - if (absl::IsUnavailable(read_status) && fin_read_) { - return absl::InvalidArgumentError("FIN received before any type"); - } - QUICHE_RETURN_IF_ERROR(read_status); - return static_cast<MoqtMessageType>(*first_message_type_); -} - absl::Status MoqtControlStreamParser::ReadMessageType() { if (current_message_type_.has_value()) { QUICHE_BUG(MoqtControlStreamParser_ReadMessageType_bad_state) @@ -549,9 +579,6 @@ "Unexpected FIN on a control stream (FIN received immediately after " "type)"); } - if (!first_message_type_.has_value()) { - first_message_type_ = *current_message_type_; - } return absl::OkStatus(); } @@ -1068,6 +1095,16 @@ return absl::OkStatus(); } +MoqtDataParser::MoqtDataParser(MoqtStreamTypeParser type_parser, + MoqtDataParserVisitor* visitor) + : stream_(*type_parser.stream()), visitor_(*visitor) { + if (type_parser.stream_type().has_value()) { + ProcessStreamType(*type_parser.stream_type()); + } else { + next_input_ = kStreamType; + } +} + void MoqtDataParser::ParseError(absl::string_view reason) { if (parsing_error_) { return; // Don't send multiple parse errors. @@ -1312,25 +1349,13 @@ } switch (next_input_) { case kStreamType: { + // TODO(vasilvv): Handle padding streams (which are allowed to FIN + // immediately after type, unlike all other types of streams). std::optional<uint64_t> value_read = ReadMoqVarIntNoFin(); if (!value_read.has_value()) { return; } - std::optional<MoqtDataStreamType> type = - MoqtDataStreamType::FromValue(*value_read); - if (!type.has_value()) { - ParseError("Invalid stream type supplied"); - return; - } - type_ = *type; - if (type_.IsPadding()) { - next_input_ = kPadding; - return; - } - if (type_.EndOfGroupInStream()) { - contains_end_of_group_ = true; - } - next_input_ = AdvanceParserState(); + ProcessStreamType(*value_read); return; } @@ -1604,4 +1629,22 @@ return stream_.SkipBytes(0); } +void MoqtDataParser::ProcessStreamType(uint64_t raw_type) { + std::optional<MoqtDataStreamType> type = + MoqtDataStreamType::FromValue(raw_type); + if (!type.has_value()) { + ParseError("Invalid stream type supplied"); + return; + } + type_ = *type; + if (type_.IsPadding()) { + next_input_ = kPadding; + return; + } + if (type_.EndOfGroupInStream()) { + contains_end_of_group_ = true; + } + next_input_ = AdvanceParserState(); +} + } // namespace moqt
diff --git a/quiche/quic/moqt/moqt_parser.h b/quiche/quic/moqt/moqt_parser.h index f8f685b..d28e203 100644 --- a/quiche/quic/moqt/moqt_parser.h +++ b/quiche/quic/moqt/moqt_parser.h
@@ -12,9 +12,9 @@ #include <cstdint> #include <optional> #include <string> +#include <utility> #include "absl/base/nullability.h" -#include "absl/cleanup/cleanup.h" #include "absl/status/status.h" #include "absl/status/statusor.h" #include "absl/strings/str_cat.h" @@ -63,12 +63,43 @@ virtual void OnParsingError(MoqtError code, absl::string_view reason) = 0; }; +// MoqtStreamTypeParser reads the initial varint from a WebTransport stream to +// determine its type before constructing either a control or a data stream +// parser. Note that both of those parsers can be safely constructed from the +// stream type parser even if the parser has not read the type yet. +class QUICHE_EXPORT MoqtStreamTypeParser { + public: + explicit MoqtStreamTypeParser(webtransport::Stream* absl_nonnull stream) + : stream_(stream) {} + ~MoqtStreamTypeParser() = default; + + // Move-only semantics to avoid a stream being accessed by two different + // parsers at the same time. + MoqtStreamTypeParser(const MoqtStreamTypeParser&) = delete; + MoqtStreamTypeParser& operator=(const MoqtStreamTypeParser&) = delete; + MoqtStreamTypeParser(MoqtStreamTypeParser&& other) noexcept; + MoqtStreamTypeParser& operator=(MoqtStreamTypeParser&& other) noexcept; + + // Reads the first varint from the stream. Returns kUnavailable if the type + // has not been received yet. + absl::StatusOr<uint64_t> ReadStreamType(); + + std::optional<uint64_t> stream_type() const { return type_; } + webtransport::Stream* absl_nonnull stream() const { return stream_; } + + private: + webtransport::Stream* absl_nonnull stream_; + std::optional<uint64_t> type_; + absl::Status status_ = absl::OkStatus(); +}; + // MoqtControlStreamParser unframes MoQT control messages from the control // stream without parsing the payload. class QUICHE_EXPORT MoqtControlStreamParser { public: explicit MoqtControlStreamParser(webtransport::Stream* absl_nonnull stream) : stream_(*stream) {} + explicit MoqtControlStreamParser(MoqtStreamTypeParser type_parser); // MoqtControlStreamParser is not movable, since reading from the same stream // through two different parsers would corrupt the state. @@ -81,8 +112,6 @@ // status if no complete message can be read; if FIN is read, `fin_read` will // be set to true. absl::StatusOr<MoqtRawControlMessage> ReadNextMessage(); - // Reads the type of the first message on the stream. - absl::StatusOr<MoqtMessageType> ReadFirstMessageType(); bool fin_read() const { return fin_read_; } webtransport::Stream* stream() const { return &stream_; } @@ -99,7 +128,6 @@ absl::Status ReadMessageType(); webtransport::Stream& stream_; - std::optional<uint64_t> first_message_type_; std::optional<uint64_t> current_message_type_; std::optional<absl::Span<char>> current_message_remaining_; std::string current_message_; @@ -266,6 +294,8 @@ explicit MoqtDataParser(webtransport::Stream* stream, MoqtDataParserVisitor* visitor) : stream_(*stream), visitor_(*visitor) {} + MoqtDataParser(MoqtStreamTypeParser type_parser, + MoqtDataParserVisitor* visitor); // Reads all of the available objects on the stream. void ReadAllData(); @@ -344,6 +374,7 @@ // Checks if we have encountered a FIN without data. If so, processes it and // returns true. bool CheckForFinWithoutData(); + void ProcessStreamType(uint64_t raw_type); void ParseError(absl::string_view reason);
diff --git a/quiche/quic/moqt/moqt_parser_test.cc b/quiche/quic/moqt/moqt_parser_test.cc index d598c3d..3407da4 100644 --- a/quiche/quic/moqt/moqt_parser_test.cc +++ b/quiche/quic/moqt/moqt_parser_test.cc
@@ -909,8 +909,6 @@ StatusIs(absl::StatusCode::kInvalidArgument)); EXPECT_THAT(parser.ReadNextMessage().status(), StatusIs(absl::StatusCode::kFailedPrecondition)); - EXPECT_THAT(parser.ReadFirstMessageType().status(), - StatusIs(absl::StatusCode::kFailedPrecondition)); } TEST_F(MoqtMessageSpecificTest, CannotAccessAfterError2) { @@ -921,8 +919,6 @@ StatusIs(absl::StatusCode::kInvalidArgument)); EXPECT_THAT(parser.ReadNextMessage().status(), StatusIs(absl::StatusCode::kFailedPrecondition)); - EXPECT_THAT(parser.ReadFirstMessageType(), - IsOkAndHolds(MoqtMessageType::kSubscribe)); } TEST_F(MoqtMessageSpecificTest, FinMidType) { @@ -1070,30 +1066,6 @@ HasSubstr("FIN on a control stream"))); } -TEST_F(MoqtMessageSpecificTest, ControlStreamReadType) { - webtransport::test::InMemoryStream stream(/*stream_id=*/0); - MoqtControlStreamParser parser(&stream); - stream.Receive("\x03", false); - absl::StatusOr<MoqtMessageType> type = parser.ReadFirstMessageType(); - EXPECT_THAT(type, IsOkAndHolds(MoqtMessageType::kSubscribe)); -} - -TEST_F(MoqtMessageSpecificTest, ControlStreamFinBeforeType) { - webtransport::test::InMemoryStream stream(/*stream_id=*/0); - MoqtControlStreamParser parser(&stream); - stream.Receive("", true); - absl::StatusOr<MoqtMessageType> type = parser.ReadFirstMessageType(); - EXPECT_EQ(type.status().code(), absl::StatusCode::kInvalidArgument); -} - -TEST_F(MoqtMessageSpecificTest, ControlStreamFinInTheMiddleOfType) { - webtransport::test::InMemoryStream stream(/*stream_id=*/0); - MoqtControlStreamParser parser(&stream); - stream.Receive("\xff", true); - absl::StatusOr<MoqtMessageType> type = parser.ReadFirstMessageType(); - EXPECT_EQ(type.status().code(), absl::StatusCode::kInvalidArgument); -} - TEST_F(MoqtMessageSpecificTest, InvalidObjectStatus) { webtransport::test::InMemoryStream stream(/*stream_id=*/0); MoqtParserTestVisitor data_visitor; @@ -1817,4 +1789,105 @@ EXPECT_FALSE(visitor_.last_message()->first_object_in_subgroup.has_value()); } +TEST_F(MoqtMessageSpecificTest, StreamTypeParserToControlStream) { + webtransport::test::InMemoryStream stream(/*stream_id=*/0); + MoqtStreamTypeParser type_parser(&stream); + stream.Receive("\x03", false); + absl::StatusOr<uint64_t> type = type_parser.ReadStreamType(); + EXPECT_THAT(type, + IsOkAndHolds(static_cast<uint64_t>(MoqtMessageType::kSubscribe))); + EXPECT_NE(type_parser.stream(), nullptr); + + MoqtControlStreamParser control_parser(std::move(type_parser)); + EXPECT_THAT( + type_parser.ReadStreamType(), // NOLINT(bugprone-use-after-move) + StatusIs(absl::StatusCode::kInternal, HasSubstr("moved-from parser"))); + EXPECT_EQ(control_parser.stream(), &stream); +} + +TEST_F(MoqtMessageSpecificTest, StreamTypeParserToDataStream) { + webtransport::test::InMemoryStream stream(/*stream_id=*/0); + MoqtStreamTypeParser type_parser(&stream); + stream.Receive("\x10", false); + absl::StatusOr<uint64_t> type = type_parser.ReadStreamType(); + EXPECT_THAT(type, IsOkAndHolds(0x10)); + EXPECT_THAT(type_parser.ReadStreamType(), + IsOkAndHolds(0x10)); // Second read should read the cached value. + EXPECT_NE(type_parser.stream(), nullptr); + + MoqtParserTestVisitor data_visitor; + MoqtDataParser data_parser(std::move(type_parser), &data_visitor); + EXPECT_THAT( + type_parser.ReadStreamType(), // NOLINT(bugprone-use-after-move) + StatusIs(absl::StatusCode::kInternal, HasSubstr("moved-from parser"))); + EXPECT_EQ(data_parser.stream_type(), MoqtDataStreamType::FromValue(0x10)); +} + +TEST_F(MoqtMessageSpecificTest, EmptyStreamTypeParserToDataStream) { + webtransport::test::InMemoryStream stream(/*stream_id=*/0); + MoqtStreamTypeParser type_parser(&stream); + MoqtParserTestVisitor data_visitor; + MoqtDataParser data_parser(std::move(type_parser), &data_visitor); + stream.Receive("\x10", false); + data_parser.ReadStreamType(); + EXPECT_EQ(data_parser.stream_type(), MoqtDataStreamType::FromValue(0x10)); +} + +TEST_F(MoqtMessageSpecificTest, StreamTypeParserFinBeforeType) { + webtransport::test::InMemoryStream stream(/*stream_id=*/0); + MoqtStreamTypeParser type_parser(&stream); + stream.Receive("", true); + absl::StatusOr<uint64_t> type = type_parser.ReadStreamType(); + EXPECT_THAT( + type, + StatusIs( + absl::StatusCode::kInvalidArgument, + HasSubstr( + "FIN received before or immediately after the stream type"))); +} + +TEST_F(MoqtMessageSpecificTest, StreamTypeParserFinInTheMiddleOfType) { + webtransport::test::InMemoryStream stream(/*stream_id=*/0); + MoqtStreamTypeParser type_parser(&stream); + stream.Receive("\xff", true); + absl::StatusOr<uint64_t> type = type_parser.ReadStreamType(); + EXPECT_THAT( + type, + StatusIs( + absl::StatusCode::kInvalidArgument, + HasSubstr( + "FIN received before or immediately after the stream type"))); +} + +TEST_F(MoqtMessageSpecificTest, StreamTypeParserFinAfterType) { + webtransport::test::InMemoryStream stream(/*stream_id=*/0); + MoqtStreamTypeParser type_parser(&stream); + stream.Receive("\x03", true); + absl::StatusOr<uint64_t> type = type_parser.ReadStreamType(); + EXPECT_THAT( + type, + StatusIs( + absl::StatusCode::kInvalidArgument, + HasSubstr( + "FIN received before or immediately after the stream type"))); +} + +TEST_F(MoqtMessageSpecificTest, StreamTypeParserFinForPadding) { + webtransport::test::InMemoryStream stream(/*stream_id=*/0); + MoqtStreamTypeParser type_parser(&stream); + stream.Receive("\xa6\xd3", true); + absl::StatusOr<uint64_t> type = type_parser.ReadStreamType(); + EXPECT_THAT( + type, IsOkAndHolds(static_cast<uint64_t>(MoqtDataStreamType::kPadding))); +} + +TEST_F(MoqtMessageSpecificTest, StreamTypeParserMovedFrom) { + webtransport::test::InMemoryStream stream(/*stream_id=*/0); + MoqtStreamTypeParser type_parser(&stream); + MoqtControlStreamParser control_parser(std::move(type_parser)); + EXPECT_THAT( + type_parser.ReadStreamType(), // NOLINT(bugprone-use-after-move) + StatusIs(absl::StatusCode::kInternal, HasSubstr("moved-from parser"))); +} + } // namespace moqt::test
diff --git a/quiche/quic/moqt/moqt_session.cc b/quiche/quic/moqt/moqt_session.cc index c9b54f5..909d49d 100644 --- a/quiche/quic/moqt/moqt_session.cc +++ b/quiche/quic/moqt/moqt_session.cc
@@ -194,8 +194,8 @@ void MoqtSession::OnIncomingUnidirectionalStreamAvailable() { while (webtransport::Stream* stream = session_->AcceptIncomingUnidirectionalStream()) { - stream->SetVisitor( - std::make_unique<IncomingDataStream>(stream, this, callbacks_.clock)); + stream->SetVisitor(std::make_unique<IncomingDataStream>( + MoqtStreamTypeParser(stream), this, callbacks_.clock)); stream->visitor()->OnCanRead(); } } @@ -886,24 +886,23 @@ } void MoqtSession::UnknownBidiStream::OnCanRead() { - absl::StatusOr<MoqtMessageType> message_type = - parser_->ReadFirstMessageType(); - if (absl::IsUnavailable(message_type.status())) { + absl::StatusOr<uint64_t> type = parser_.ReadStreamType(); + if (absl::IsUnavailable(type.status())) { return; } - if (absl::IsInvalidArgument(message_type.status())) { + if (absl::IsInvalidArgument(type.status())) { // Received a FIN before any type has been available, which is malformed. - session_->Error(MoqtError::kProtocolViolation, - message_type.status().message()); + session_->Error(MoqtError::kProtocolViolation, type.status().message()); return; } - if (!message_type.ok()) { + if (!type.ok()) { // The result is neither of "OK", "no type available", or "parse error". // This is unexpected; treat it as an internal error, and reset the stream. stream_->ResetWithUserCode(kResetCodeInternalError); return; } - switch (*message_type) { + MoqtMessageType message_type = static_cast<MoqtMessageType>(*type); + switch (message_type) { case MoqtMessageType::kSetup: { if (session_->control_stream_.GetIfAvailable() != nullptr) { session_->Error(MoqtError::kProtocolViolation,
diff --git a/quiche/quic/moqt/moqt_session.h b/quiche/quic/moqt/moqt_session.h index b031dc2..21ff0a6 100644 --- a/quiche/quic/moqt/moqt_session.h +++ b/quiche/quic/moqt/moqt_session.h
@@ -234,9 +234,7 @@ // responsible for calling stream->SetVisitor(). UnknownBidiStream(MoqtSession* session, webtransport::Stream* absl_nonnull stream) - : session_(session), - stream_(stream), - parser_(std::make_unique<MoqtControlStreamParser>(stream)) {} + : session_(session), stream_(stream), parser_(stream) {} ~UnknownBidiStream() {} // webtransport::StreamVisitor overrides. @@ -249,7 +247,7 @@ private: MoqtSession* session_; webtransport::Stream* stream_; - std::unique_ptr<MoqtControlStreamParser> parser_; + MoqtStreamTypeParser parser_; }; class QUICHE_EXPORT ControlStream : public MoqtBidiStreamBase {
diff --git a/quiche/quic/moqt/moqt_uni_stream.h b/quiche/quic/moqt/moqt_uni_stream.h index b681125..e1234fb 100644 --- a/quiche/quic/moqt/moqt_uni_stream.h +++ b/quiche/quic/moqt/moqt_uni_stream.h
@@ -210,13 +210,17 @@ class QUICHE_EXPORT IncomingDataStream : public webtransport::StreamVisitor, public MoqtDataParserVisitor { public: + IncomingDataStream(MoqtStreamTypeParser type_parser, + SessionToUniStreamInterface* absl_nonnull session, + const quic::QuicClock* absl_nonnull clock) + : stream_(type_parser.stream()), + parser_(std::move(type_parser), this), + session_(session), + clock_(clock) {} IncomingDataStream(webtransport::Stream* absl_nonnull stream, SessionToUniStreamInterface* absl_nonnull session, const quic::QuicClock* absl_nonnull clock) - : stream_(stream), - parser_(stream, this), - session_(session), - clock_(clock) {} + : IncomingDataStream(MoqtStreamTypeParser(stream), session, clock) {} ~IncomingDataStream(); // webtransport::StreamVisitor implementation.