Use a pair of unidirectional streams for control messages. PiperOrigin-RevId: 972222230
diff --git a/quiche/quic/moqt/moqt_integration_test.cc b/quiche/quic/moqt/moqt_integration_test.cc index 2e239ee..8650fc8 100644 --- a/quiche/quic/moqt/moqt_integration_test.cc +++ b/quiche/quic/moqt/moqt_integration_test.cc
@@ -748,9 +748,10 @@ } TEST_F(MoqtIntegrationTest, DeliveryTimeout) { - CreateDefaultEndpoints(); + EstablishSession(); + // The loss is added after the handshake, to ensure that the connection has + // enough congestion control window to work with for the further tests. WireUpEndpointsWithLoss(/*lose_every_n=*/4); - ConnectEndpoints(); FullTrackName full_track_name("foo", "bar"); MoqtKnownTrackPublisher publisher; @@ -936,7 +937,7 @@ EXPECT_EQ(annotation.moqt_subgroup_stream().subgroup_id(), 0); } } - EXPECT_EQ(control_streams, 1); + EXPECT_EQ(control_streams, 2); EXPECT_EQ(subgroup_streams, 1); int objects_enqueued = 0;
diff --git a/quiche/quic/moqt/moqt_session.cc b/quiche/quic/moqt/moqt_session.cc index a762b7b..00565a0 100644 --- a/quiche/quic/moqt/moqt_session.cc +++ b/quiche/quic/moqt/moqt_session.cc
@@ -114,7 +114,7 @@ } void MoqtSession::SendControlMessage(quiche::QuicheBuffer message) { - ControlStream* control_stream = GetControlStream(); + OutgoingControlStream* control_stream = GetOutgoingControlStream(); if (control_stream == nullptr) { QUICHE_LOG(DFATAL) << "Trying to send a message on the control stream " "while it does not exist"; @@ -131,27 +131,23 @@ "MOQT peer chose wrong subprotocol"); return; } - if (parameters_.perspective == Perspective::IS_SERVER) { - return; - } - auto control_stream = std::make_unique<ControlStream>(this); - if (!session_->CanOpenNextOutgoingBidirectionalStream()) { + if (!session_->CanOpenNextOutgoingUnidirectionalStream()) { Error(MoqtError::kControlMessageTimeout, "Unable to open a control stream"); return; } - webtransport::Stream* stream = session_->OpenOutgoingBidirectionalStream(); + webtransport::Stream* stream = session_->OpenOutgoingUnidirectionalStream(); if (stream == nullptr) { Error(MoqtError::kInternalError, "Unable to open a control stream"); return; } - control_stream_ = control_stream->GetWeakPtr(); - control_stream->BindStream(stream); + auto control_stream = std::make_unique<OutgoingControlStream>(this, stream); + outgoing_control_stream_ = control_stream->GetWeakPtr(); trace_recorder_.RecordControlStreamCreated(stream->GetStreamId()); stream->SetVisitor(std::move(control_stream)); MoqtSetup setup; parameters_.ToSetupParameters(setup.parameters); SendControlMessage(framer_.SerializeSetup(setup)); - QUIC_DLOG(INFO) << ENDPOINT << "Send CLIENT_SETUP"; + QUIC_DLOG(INFO) << ENDPOINT << "Send SETUP"; } void MoqtSession::OnSessionClosed(webtransport::SessionErrorCode, @@ -168,6 +164,9 @@ } void MoqtSession::OnIncomingBidirectionalStreamAvailable() { + if (!peer_setup_received_) { + return; + } while (webtransport::Stream* stream = session_->AcceptIncomingBidirectionalStream()) { if (sent_goaway_) { @@ -196,8 +195,7 @@ void MoqtSession::OnIncomingUnidirectionalStreamAvailable() { while (webtransport::Stream* stream = session_->AcceptIncomingUnidirectionalStream()) { - stream->SetVisitor(std::make_unique<IncomingDataStream>( - MoqtStreamTypeParser(stream), this, callbacks_.clock)); + stream->SetVisitor(std::make_unique<UnknownUniStream>(this, stream)); stream->visitor()->OnCanRead(); } } @@ -918,24 +916,6 @@ } MoqtMessageType message_type = static_cast<MoqtMessageType>(*type); switch (message_type) { - case MoqtMessageType::kSetup: { - if (session_->control_stream_.GetIfAvailable() != nullptr) { - session_->Error(MoqtError::kProtocolViolation, - "Multiple control streams"); - return; - } - auto control_stream = std::make_unique<ControlStream>(session_); - control_stream->BindStream(std::move(parser_)); - // Store a reference to the stream context when the current context is - // destroyed below. - ControlStream* temp_stream = control_stream.get(); - session_->control_stream_ = temp_stream->GetWeakPtr(); - // Deletes the UnknownBidiStream object; no class access after this - // point. - stream_->SetVisitor(std::move(control_stream)); - temp_stream->OnCanRead(); - break; - } case MoqtMessageType::kSubscribeNamespace: { auto namespace_stream = std::make_unique<MoqtSubscribeNamespaceResponseStream>( @@ -1165,42 +1145,157 @@ } } -void MoqtSession::ControlStream::OnStreamBound() { - stream()->SetPriority( - webtransport::StreamPriority{/*send_group_id=*/kMoqtSendGroupId, - /*send_order=*/kMoqtControlStreamSendOrder}); +void MoqtSession::UnknownUniStream::OnCanRead() { + MoqtSession* session = MoqtSessionFromWeakPtr(session_); + if (session == nullptr || session->is_closing_) { + return; + } + absl::StatusOr<uint64_t> type = parser_.ReadStreamType(); + if (absl::IsUnavailable(type.status())) { + return; + } + if (absl::IsInvalidArgument(type.status())) { + // Received a FIN before any type has been available, which is malformed. + session->Error(MoqtError::kProtocolViolation, type.status().message()); + return; + } + if (!type.ok()) { + stream_->ResetWithUserCode(kResetCodeInternalError); + return; + } + if (*type == static_cast<uint64_t>(MoqtMessageType::kSetup)) { + if (session->incoming_control_stream_.GetIfAvailable() != nullptr) { + session->Error(MoqtError::kProtocolViolation, "Multiple control streams"); + return; + } + session->trace_recorder().RecordControlStreamCreated( + stream_->GetStreamId()); + auto control_stream = + std::make_unique<IncomingControlStream>(session, std::move(parser_)); + IncomingControlStream* temp_stream = control_stream.get(); + session->incoming_control_stream_ = temp_stream->GetWeakPtr(); + // The line below destroys `this`. + stream_->SetVisitor(std::move(control_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(); + // The line below destroys `this`. + stream_->SetVisitor(std::move(data_stream)); + temp_stream->OnCanRead(); } -absl::Status MoqtSession::ControlStream::OnRawControlMessage( - const MoqtRawControlMessage& message) { - return ControlMessageDispatcher::DispatchControlMessage( - *session_, message_parser(), message, "control"); +MoqtSession::IncomingControlStream::IncomingControlStream( + MoqtSession* absl_nonnull session, MoqtStreamTypeParser type_parser) + : session_(session->GetWeakPtr()), + parser_(std::move(type_parser)), + weak_ptr_factory_(this) {} + +void MoqtSession::IncomingControlStream::OnCanRead() { + quiche::QuicheWeakPtr<IncomingControlStream> weak_this = + weak_ptr_factory_.Create(); + MoqtSession* session = MoqtSessionFromWeakPtr(session_); + if (session == nullptr || session->is_closing_) { + return; + } + while (true) { + absl::StatusOr<MoqtRawControlMessage> message = parser_.ReadNextMessage(); + if (absl::IsUnavailable(message.status())) { + return; + } + if (!message.ok()) { + std::optional<MoqtError> error_code = + GetMoqtErrorForStatus(message.status()); + session->Error(error_code.value_or(MoqtError::kProtocolViolation), + message.status().message()); + return; + } + + absl::Status status = ControlMessageDispatcher::DispatchControlMessage( + *session, session->ControlMessageParser(), *message, "control"); + // `DispatchControlMessage` might have closed the session by itself, + // resulting in the stream and/or the session object being deleted. + if (!weak_this.IsValid() || !session_.IsValid() || session->is_closing_) { + return; + } + if (!status.ok()) { + std::optional<MoqtError> error_code = GetMoqtErrorForStatus(status); + session->Error(error_code.value_or(MoqtError::kProtocolViolation), + status.message()); + return; + } + } +} + +void MoqtSession::IncomingControlStream::OnResetStreamReceived( + webtransport::StreamErrorCode /*error*/) { + MoqtSession* session = MoqtSessionFromWeakPtr(session_); + if (session != nullptr) { + session->Error(MoqtError::kProtocolViolation, + "Control stream reset received"); + } +} + +MoqtSession::OutgoingControlStream::OutgoingControlStream( + MoqtSession* absl_nonnull session, + webtransport::Stream* absl_nonnull stream) + : session_(session->GetWeakPtr()), + outgoing_message_queue_(stream), + weak_ptr_factory_(this) { + if (stream != nullptr) { + stream->SetPriority(webtransport::StreamPriority{ + /*send_group_id=*/kMoqtSendGroupId, + /*send_order=*/kMoqtControlStreamSendOrder}); + } +} + +void MoqtSession::OutgoingControlStream::OnCanWrite() { + CheckStatus(outgoing_message_queue_.OnCanWrite()); +} + +void MoqtSession::OutgoingControlStream::OnStopSendingReceived( + webtransport::StreamErrorCode /*error*/) { + MoqtSession* session = MoqtSessionFromWeakPtr(session_); + if (session != nullptr) { + session->Error(MoqtError::kProtocolViolation, + "Control stream stop sending received"); + } +} + +void MoqtSession::OutgoingControlStream::CheckStatus(absl::Status status) { + MoqtSession* session = MoqtSessionFromWeakPtr(session_); + if (session == nullptr) { + return; + } + if (!status.ok() && !session->is_closing_) { + std::optional<MoqtError> error_code = GetMoqtErrorForStatus(status); + session->Error(error_code.value_or(MoqtError::kInternalError), + status.message()); + } } absl::Status MoqtSession::OnControlMessage(const MoqtSetup& message) { - if (parameters_.perspective == Perspective::IS_SERVER) { - peer_supports_object_ack_ = message.parameters.support_object_acks.value_or( - kDefaultSupportObjectAcks); - peer_max_request_id_ = - message.parameters.max_request_id.value_or(kDefaultMaxRequestId); - QUICHE_DLOG(INFO) << "Received CLIENT_SETUP"; - MoqtSetup response; - parameters_.ToSetupParameters(response.parameters); - SendControlMessage(framer_.SerializeSetup(response)); - QUICHE_DLOG(INFO) << "Sent SERVER_SETUP"; - // TODO: handle path. - std::move(callbacks_.session_established_callback)(); - return absl::OkStatus(); - } else { - peer_supports_object_ack_ = message.parameters.support_object_acks.value_or( - kDefaultSupportObjectAcks); - QUIC_DLOG(INFO) << ENDPOINT << "Received the SETUP message"; - // TODO: handle path. - peer_max_request_id_ = - message.parameters.max_request_id.value_or(kDefaultMaxRequestId); - std::move(callbacks_.session_established_callback)(); - return absl::OkStatus(); + if (peer_setup_received_) { + return absl::InvalidArgumentError("Duplicate SETUP message"); } + peer_setup_received_ = true; + peer_supports_object_ack_ = message.parameters.support_object_acks.value_or( + kDefaultSupportObjectAcks); + peer_max_request_id_ = + message.parameters.max_request_id.value_or(kDefaultMaxRequestId); + QUIC_DLOG(INFO) << ENDPOINT << "Received the SETUP message"; + // TODO: handle path. + if (callbacks_.session_established_callback != nullptr) { + MoqtSessionEstablishedCallback callback = + std::move(callbacks_.session_established_callback); + callbacks_.session_established_callback = nullptr; + std::move(callback)(); + } + // Drain streams that were potentially stalled due to a missing SETUP. + OnIncomingBidirectionalStreamAvailable(); + return absl::OkStatus(); } absl::Status MoqtSession::OnControlMessage(const MoqtRequestOk& message) { @@ -1542,7 +1637,7 @@ it->second->Destroy(); // This is only called from the callback where UpstreamFetchTask has been // destroyed, so there is no need to notify the application. - ControlStream* stream = GetControlStream(); + OutgoingControlStream* stream = GetOutgoingControlStream(); if (stream == nullptr) { return; }
diff --git a/quiche/quic/moqt/moqt_session.h b/quiche/quic/moqt/moqt_session.h index 7e4904f..4593bff 100644 --- a/quiche/quic/moqt/moqt_session.h +++ b/quiche/quic/moqt/moqt_session.h
@@ -24,6 +24,7 @@ #include "quiche/quic/core/quic_time.h" #include "quiche/quic/core/quic_types.h" #include "quiche/quic/moqt/moqt_bidi_stream.h" +#include "quiche/quic/moqt/moqt_control_message_queue.h" #include "quiche/quic/moqt/moqt_error.h" #include "quiche/quic/moqt/moqt_fetch_task.h" #include "quiche/quic/moqt/moqt_framer.h" @@ -38,7 +39,6 @@ #include "quiche/quic/moqt/moqt_session_callbacks.h" #include "quiche/quic/moqt/moqt_session_interface.h" #include "quiche/quic/moqt/moqt_trace_recorder.h" -#include "quiche/quic/moqt/moqt_track_status_stream.h" #include "quiche/quic/moqt/moqt_types.h" #include "quiche/quic/moqt/moqt_uni_stream.h" #include "quiche/quic/moqt/session_namespace_tree.h" @@ -54,7 +54,8 @@ namespace test { class MoqtSessionPeer; -} +class MoqtBidiStreamTestWrapper; +} // namespace test inline constexpr quic::QuicTimeDelta kDefaultGoAwayTimeout = quic::QuicTime::Delta::FromSeconds(10); @@ -202,7 +203,7 @@ void set_publisher(MoqtPublisher* publisher) { publisher_ = publisher; } bool support_object_acks() const { return parameters_.support_object_acks; } void set_support_object_acks(bool value) { - QUICHE_DCHECK(!control_stream_.IsValid()) + QUICHE_DCHECK(!outgoing_control_stream_.IsValid()) << "support_object_acks needs to be set before handshake"; parameters_.support_object_acks = value; } @@ -228,6 +229,7 @@ private: friend class ControlMessageDispatcher; friend class test::MoqtSessionPeer; + friend class test::MoqtBidiStreamTestWrapper; struct Empty {}; @@ -254,51 +256,93 @@ MoqtStreamTypeParser parser_; }; - class QUICHE_EXPORT ControlStream : public MoqtBidiStreamBase { + // UnknownUniStream is the initial handler for all incoming unidirectional + // streams; it reads the type tag from the wire, and creates an appropriate + // handler based on that. + class QUICHE_EXPORT UnknownUniStream : public webtransport::StreamVisitor { public: - explicit ControlStream(MoqtSession* session) - : MoqtBidiStreamBase( - &session->framer_, session->ControlMessageParser(), - [session](MoqtError code, absl::string_view reason) { - session->control_stream_ = - quiche::QuicheWeakPtr<ControlStream>(); - if (!session->is_closing_) { - session->Error(code, reason); - } - }), - session_(session), - weak_ptr_factory_(this) {} + UnknownUniStream(MoqtSession* absl_nonnull session, + webtransport::Stream* absl_nonnull stream) + : session_(session->GetWeakPtr()), stream_(stream), parser_(stream) {} - void OnStreamBound() override; - absl::Status OnRawControlMessage( - const MoqtRawControlMessage& message) override; + // webtransport::StreamVisitor overrides. + void OnResetStreamReceived(webtransport::StreamErrorCode error) override {} + void OnStopSendingReceived(webtransport::StreamErrorCode error) override {} + void OnWriteSideInDataRecvdState() override {} + void OnCanRead() override; + void OnCanWrite() override {} - // webtransport::StreamVisitor overrides - void OnResetStreamReceived(webtransport::StreamErrorCode error) override { - session_->Error(MoqtError::kProtocolViolation, - "Control stream reset received"); - } + private: + quiche::QuicheWeakPtr<MoqtSessionInterface> session_; + webtransport::Stream* absl_nonnull stream_; + MoqtStreamTypeParser parser_; + }; + + class QUICHE_EXPORT IncomingControlStream + : public webtransport::StreamVisitor { + public: + IncomingControlStream(MoqtSession* absl_nonnull session, + MoqtStreamTypeParser parser); + + void OnCanRead() override; + void OnCanWrite() override {} + void OnResetStreamReceived(webtransport::StreamErrorCode error) override; void OnStopSendingReceived(webtransport::StreamErrorCode error) override { - session_->Error(MoqtError::kProtocolViolation, - "Control stream stop sending received"); + // Impossible for QUIC incoming unidirectional streams. } + void OnWriteSideInDataRecvdState() override {} - quic::Perspective perspective() const { - return session_->parameters_.perspective; - } - quiche::QuicheWeakPtr<ControlStream> GetWeakPtr() { + quiche::QuicheWeakPtr<IncomingControlStream> GetWeakPtr() { return weak_ptr_factory_.Create(); } - void Detach() override { - session_->Error(MoqtError::kProtocolViolation, "Control stream closed"); - } private: friend class test::MoqtSessionPeer; + friend class test::MoqtBidiStreamTestWrapper; - MoqtSession* session_; + quiche::QuicheWeakPtr<MoqtSessionInterface> session_; + MoqtControlStreamParser parser_; // Must be last. - quiche::QuicheWeakPtrFactory<ControlStream> weak_ptr_factory_; + quiche::QuicheWeakPtrFactory<IncomingControlStream> weak_ptr_factory_; + }; + + class QUICHE_EXPORT OutgoingControlStream + : public webtransport::StreamVisitor { + public: + OutgoingControlStream(MoqtSession* absl_nonnull session, + webtransport::Stream* absl_nonnull stream); + + void OnCanRead() override {} + void OnCanWrite() override; + void OnResetStreamReceived(webtransport::StreamErrorCode error) override { + // Impossible for QUIC incoming unidirectional streams. + } + void OnStopSendingReceived(webtransport::StreamErrorCode error) override; + void OnWriteSideInDataRecvdState() override {} + + absl::Status SendOrBufferMessage(quiche::QuicheBuffer message, + bool fin = false) { + return outgoing_message_queue_.SendOrBufferMessage(std::move(message), + fin); + } + void SendOrBufferMessageOrFatal(quiche::QuicheBuffer message, + bool fin = false) { + CheckStatus(SendOrBufferMessage(std::move(message), fin)); + } + void CheckStatus(absl::Status status); + + quiche::QuicheWeakPtr<OutgoingControlStream> GetWeakPtr() { + return weak_ptr_factory_.Create(); + } + + private: + friend class test::MoqtSessionPeer; + friend class test::MoqtBidiStreamTestWrapper; + + quiche::QuicheWeakPtr<MoqtSessionInterface> session_; + MoqtControlMessageQueue outgoing_message_queue_; + // Must be last. + quiche::QuicheWeakPtrFactory<OutgoingControlStream> weak_ptr_factory_; }; class QUICHE_EXPORT PublishedFetch { @@ -332,8 +376,14 @@ MoqtSession* session_; }; - // Returns the pointer to the control stream, or nullptr if none is present. - ControlStream* GetControlStream() { return control_stream_.GetIfAvailable(); } + // Returns the pointer to the outgoing control stream, or nullptr if none is + // present. + OutgoingControlStream* GetOutgoingControlStream() { + return outgoing_control_stream_.GetIfAvailable(); + } + IncomingControlStream* GetIncomingControlStream() { + return incoming_control_stream_.GetIfAvailable(); + } // Sends a message on the control stream; QUICHE_DCHECKs if no control stream // is present. void SendControlMessage(quiche::QuicheBuffer message); @@ -421,11 +471,14 @@ MoqtSessionCallbacks callbacks_; MoqtFramer framer_; - quiche::QuicheWeakPtr<ControlStream> control_stream_ = - quiche::QuicheWeakPtr<ControlStream>(); + quiche::QuicheWeakPtr<IncomingControlStream> incoming_control_stream_ = + quiche::QuicheWeakPtr<IncomingControlStream>(); + quiche::QuicheWeakPtr<OutgoingControlStream> outgoing_control_stream_ = + quiche::QuicheWeakPtr<OutgoingControlStream>(); quiche::QuicheCircularDeque<std::unique_ptr<MoqtBidiStreamBase>> pending_bidi_streams_; bool peer_supports_object_ack_ = false; + bool peer_setup_received_ = false; std::string error_; bool sent_goaway_ = false;
diff --git a/quiche/quic/moqt/moqt_session_test.cc b/quiche/quic/moqt/moqt_session_test.cc index cbd80cc..61a52ae 100644 --- a/quiche/quic/moqt/moqt_session_test.cc +++ b/quiche/quic/moqt/moqt_session_test.cc
@@ -143,6 +143,7 @@ session_.set_publisher(&publisher_); MoqtSessionPeer::set_peer_max_request_id(&session_, kDefaultInitialMaxRequestId); + MoqtSessionPeer::set_peer_setup_received(&session_, true); ON_CALL(mock_session_, GetStreamById) .WillByDefault(Return(&mock_bidi_stream_)); EXPECT_EQ(MoqtSessionPeer::GetImplementationString(&session_), @@ -286,7 +287,7 @@ .WillOnce(Return(stream)) .WillOnce(Return(nullptr)); EXPECT_CALL(*stream, SetVisitor(_)) - .WillOnce( + .WillRepeatedly( [&](std::unique_ptr<webtransport::StreamVisitor> new_visitor) { visitor = std::move(new_visitor); }); @@ -338,41 +339,46 @@ EXPECT_EQ(session_.perspective(), quic::Perspective::IS_CLIENT); } -// Verify the session sends CLIENT_SETUP on the control stream. +// Verify the session sends SETUP on the control stream. TEST_F(MoqtSessionTest, OnSessionReady) { + MoqtSessionPeer::set_peer_setup_received(&session_, false); EXPECT_CALL(mock_session_, GetNegotiatedSubprotocol) .WillOnce(Return(std::optional<std::string>(kDefaultMoqtVersion))); - EXPECT_CALL(mock_session_, CanOpenNextOutgoingBidirectionalStream()) + EXPECT_CALL(mock_session_, CanOpenNextOutgoingUnidirectionalStream()) .WillOnce(Return(true)); - EXPECT_CALL(mock_session_, OpenOutgoingBidirectionalStream()) - .WillOnce(Return(&mock_bidi_stream_)); - EXPECT_CALL(mock_bidi_stream_, CanWrite).WillRepeatedly(Return(true)); + EXPECT_CALL(mock_session_, OpenOutgoingUnidirectionalStream()) + .WillOnce(Return(&mock_uni_stream_)); + EXPECT_CALL(mock_uni_stream_, CanWrite).WillRepeatedly(Return(true)); std::unique_ptr<webtransport::StreamVisitor> visitor; - // Save a reference to MoqtSession::Stream - EXPECT_CALL(mock_bidi_stream_, SetVisitor(_)) + EXPECT_CALL(mock_uni_stream_, SetVisitor(_)) .WillOnce([&](std::unique_ptr<webtransport::StreamVisitor> new_visitor) { visitor = std::move(new_visitor); }); - EXPECT_CALL(mock_bidi_stream_, GetStreamId()) + EXPECT_CALL(mock_uni_stream_, GetStreamId()) .WillRepeatedly(Return(webtransport::StreamId(4))); - EXPECT_CALL(mock_bidi_stream_, + EXPECT_CALL(mock_uni_stream_, Writev(ControlMessageOfType(MoqtMessageType::kSetup), _)); session_.OnSessionReady(); - // Receive SERVER_SETUP - bidi_wrapper_ = - MoqtSessionPeer::FetchParserVisitorFromWebtransportStreamVisitor( - std::move(visitor)); - // Handle the server setup - MoqtSetup setup; // No fields are set. + // Receive SERVER_SETUP on an incoming unidirectional stream + webtransport::test::InMemoryStreamWithWriteBuffer in_memory_stream(0); + MoqtFramer framer(true, quic::Perspective::IS_SERVER); + MoqtSetup setup; + quiche::QuicheBuffer buffer = framer.SerializeSetup(setup); + in_memory_stream.Receive(absl::string_view(buffer.data(), buffer.size()), + /*fin=*/false); + + EXPECT_CALL(mock_session_, AcceptIncomingUnidirectionalStream()) + .WillOnce(Return(&in_memory_stream)) + .WillOnce(Return(nullptr)); EXPECT_CALL(session_callbacks_.session_established_callback, Call()).Times(1); - bidi_wrapper_->ReceiveMessage(setup); + session_.OnIncomingUnidirectionalStreamAvailable(); } TEST_F(MoqtSessionTest, OnSessionReadyNoControlStream) { EXPECT_CALL(mock_session_, GetNegotiatedSubprotocol) .WillOnce(Return(std::optional<std::string>(kDefaultMoqtVersion))); - EXPECT_CALL(mock_session_, CanOpenNextOutgoingBidirectionalStream) + EXPECT_CALL(mock_session_, CanOpenNextOutgoingUnidirectionalStream) .WillOnce(Return(false)); EXPECT_CALL(session_callbacks_.session_terminated_callback, Call); session_.OnSessionReady(); @@ -383,6 +389,7 @@ &mock_session_, MoqtSessionParameters(quic::Perspective::IS_SERVER), std::make_unique<quic::test::TestAlarmFactory>(), session_callbacks_.AsSessionCallbacks()); + MoqtSessionPeer::set_peer_setup_received(&server_session, true); EXPECT_CALL(mock_session_, AcceptIncomingBidirectionalStream()) .WillOnce(Return(&mock_bidi_stream_)) .WillOnce(Return(nullptr)); @@ -404,7 +411,7 @@ MoqtSession server_session(&mock_session_, session_parameters, std::make_unique<quic::test::TestAlarmFactory>(), session_callbacks_.AsSessionCallbacks()); - // Load a CLIENT_SETUP message into an in-memory stream. + // Load a SETUP message into an in-memory stream. webtransport::test::InMemoryStreamWithWriteBuffer in_memory_stream(0); MoqtFramer framer(session_parameters.using_webtrans, quic::Perspective::IS_CLIENT); @@ -414,14 +421,82 @@ in_memory_stream.Receive(absl::string_view(buffer.data(), buffer.size()), /*fin=*/false); - EXPECT_CALL(mock_session_, AcceptIncomingBidirectionalStream()) + EXPECT_CALL(mock_session_, AcceptIncomingUnidirectionalStream()) .WillOnce(Return(&in_memory_stream)) .WillOnce(Return(nullptr)); EXPECT_CALL(session_callbacks_.session_established_callback, Call()); - server_session.OnIncomingBidirectionalStreamAvailable(); - EXPECT_EQ(PeekControlMessageType(in_memory_stream.write_buffer()), - MoqtMessageType::kSetup); - EXPECT_NE(MoqtSessionPeer::GetControlStream(&server_session), nullptr); + server_session.OnIncomingUnidirectionalStreamAvailable(); + EXPECT_NE(MoqtSessionPeer::GetIncomingControlStream(&server_session), + nullptr); +} + +TEST_F(MoqtSessionTest, DuplicateSetup) { + MoqtSessionParameters session_parameters(quic::Perspective::IS_SERVER); + MoqtSession server_session(&mock_session_, session_parameters, + std::make_unique<quic::test::TestAlarmFactory>(), + session_callbacks_.AsSessionCallbacks()); + webtransport::test::InMemoryStreamWithWriteBuffer in_memory_stream(0); + MoqtFramer framer(session_parameters.using_webtrans, + quic::Perspective::IS_CLIENT); + MoqtSetup setup; + session_parameters.ToSetupParameters(setup.parameters); + quiche::QuicheBuffer buffer = framer.SerializeSetup(setup); + in_memory_stream.Receive(absl::string_view(buffer.data(), buffer.size()), + /*fin=*/false); + + EXPECT_CALL(mock_session_, AcceptIncomingUnidirectionalStream()) + .WillOnce(Return(&in_memory_stream)) + .WillOnce(Return(nullptr)); + EXPECT_CALL(session_callbacks_.session_established_callback, Call()); + server_session.OnIncomingUnidirectionalStreamAvailable(); + EXPECT_NE(MoqtSessionPeer::GetIncomingControlStream(&server_session), + nullptr); + + // Send a duplicate SETUP message on the control stream. + EXPECT_CALL(mock_session_, + CloseSession(static_cast<uint64_t>(MoqtError::kProtocolViolation), + "Duplicate SETUP message")); + EXPECT_CALL(session_callbacks_.session_terminated_callback, + Call(absl::string_view("Duplicate SETUP message"))); + in_memory_stream.Receive(absl::string_view(buffer.data(), buffer.size()), + /*fin=*/false); +} + +TEST_F(MoqtSessionTest, TwoStreamsStartWithSetup) { + MoqtSessionParameters session_parameters(quic::Perspective::IS_SERVER); + MoqtSession server_session(&mock_session_, session_parameters, + std::make_unique<quic::test::TestAlarmFactory>(), + session_callbacks_.AsSessionCallbacks()); + webtransport::test::InMemoryStreamWithWriteBuffer stream1(0); + webtransport::test::InMemoryStreamWithWriteBuffer stream2(1); + + MoqtFramer framer(session_parameters.using_webtrans, + quic::Perspective::IS_CLIENT); + MoqtSetup setup; + session_parameters.ToSetupParameters(setup.parameters); + quiche::QuicheBuffer buffer = framer.SerializeSetup(setup); + + stream1.Receive(absl::string_view(buffer.data(), buffer.size()), + /*fin=*/false); + EXPECT_CALL(mock_session_, AcceptIncomingUnidirectionalStream()) + .WillOnce(Return(&stream1)) + .WillOnce(Return(nullptr)); + EXPECT_CALL(session_callbacks_.session_established_callback, Call()); + server_session.OnIncomingUnidirectionalStreamAvailable(); + EXPECT_NE(MoqtSessionPeer::GetIncomingControlStream(&server_session), + nullptr); + + stream2.Receive(absl::string_view(buffer.data(), buffer.size()), + /*fin=*/false); + EXPECT_CALL(mock_session_, AcceptIncomingUnidirectionalStream()) + .WillOnce(Return(&stream2)) + .WillOnce(Return(nullptr)); + EXPECT_CALL(mock_session_, + CloseSession(static_cast<uint64_t>(MoqtError::kProtocolViolation), + "Multiple control streams")); + EXPECT_CALL(session_callbacks_.session_terminated_callback, + Call(absl::string_view("Multiple control streams"))); + server_session.OnIncomingUnidirectionalStreamAvailable(); } TEST_F(MoqtSessionTest, OnSessionClosed) { @@ -2429,6 +2504,54 @@ EXPECT_TRUE(reported_error); } +TEST_F(MoqtSessionTest, IncomingTrackStatusBeforeSetup) { + MoqtSessionParameters session_parameters(quic::Perspective::IS_SERVER); + MoqtSession server_session(&mock_session_, session_parameters, + std::make_unique<quic::test::TestAlarmFactory>(), + session_callbacks_.AsSessionCallbacks()); + server_session.set_publisher(&publisher_); + MockTrackPublisher* track = CreateTrackPublisher(); + + // Receive TRACK_STATUS on an incoming bidirectional stream. + webtransport::test::InMemoryStreamWithWriteBuffer bidi_stream(0); + MoqtFramer client_framer(session_parameters.using_webtrans, + quic::Perspective::IS_CLIENT); + MoqtTrackStatus track_status = DefaultSubscribe(); + quiche::QuicheBuffer serialized_track_status = + client_framer.SerializeTrackStatus(track_status); + bidi_stream.Receive(serialized_track_status.AsStringView(), + /*fin=*/false); + + // Before SETUP is received, the incoming bidirectional stream is not + // accepted. + EXPECT_CALL(mock_session_, AcceptIncomingBidirectionalStream).Times(0); + EXPECT_CALL(*track, AddObjectListener).Times(0); + server_session.OnIncomingBidirectionalStreamAvailable(); + + // Receive CLIENT_SETUP on an incoming unidirectional stream. + webtransport::test::InMemoryStreamWithWriteBuffer control_stream(1); + MoqtSetup setup; + session_parameters.ToSetupParameters(setup.parameters); + quiche::QuicheBuffer setup_buffer = client_framer.SerializeSetup(setup); + control_stream.Receive( + absl::string_view(setup_buffer.data(), setup_buffer.size()), + /*fin=*/false); + + MoqtObjectListener* listener = nullptr; + EXPECT_CALL(session_callbacks_.session_established_callback, Call); + EXPECT_CALL(mock_session_, AcceptIncomingUnidirectionalStream) + .WillOnce(Return(&control_stream)) + .WillOnce(Return(nullptr)); + EXPECT_CALL(mock_session_, AcceptIncomingBidirectionalStream) + .WillOnce(Return(&bidi_stream)) + .WillOnce(Return(nullptr)); + EXPECT_CALL(*track, AddObjectListener) + .WillOnce(testing::SaveArg<0>(&listener)); + + server_session.OnIncomingUnidirectionalStreamAvailable(); + EXPECT_NE(listener, nullptr); +} + TEST_F(MoqtSessionTest, IncomingTrackStatusThenSynchronousOk) { bidi_wrapper_ = std::make_unique<MoqtBidiStreamTestWrapper>( ResponseStream(kTrackStatusByte)); @@ -2663,18 +2786,6 @@ session_.OnSessionReady(); } - -TEST_F(MoqtSessionTest, ClientSetupNotAllowedOnControlStream) { - // While technically on the Control stream, when it arrives, it's an - // UnknownBidiStream - bidi_wrapper_ = - MoqtSessionPeer::CreateControlStream(&session_, &mock_bidi_stream_); - EXPECT_CALL(mock_session_, CloseSession); - EXPECT_CALL(session_callbacks_.session_terminated_callback, Call); - bidi_wrapper_->ReceiveMessage( - MoqtSetup(SetupParameters("/", "example.com", 0))); -} - TEST_F(MoqtSessionTest, NamespaceNotAllowedOnControlStream) { bidi_wrapper_ = MoqtSessionPeer::CreateControlStream(&session_, &mock_bidi_stream_); @@ -2854,9 +2965,9 @@ webtransport::test::InMemoryStreamWithWriteBuffer control_stream(0); EXPECT_CALL(mock_session_, GetNegotiatedSubprotocol) .WillOnce(Return(std::string(kDefaultMoqtVersion))); - EXPECT_CALL(mock_session_, CanOpenNextOutgoingBidirectionalStream) + EXPECT_CALL(mock_session_, CanOpenNextOutgoingUnidirectionalStream) .WillOnce(Return(true)); - EXPECT_CALL(mock_session_, OpenOutgoingBidirectionalStream) + EXPECT_CALL(mock_session_, OpenOutgoingUnidirectionalStream) .WillOnce(Return(&control_stream)); session_.OnSessionReady(); control_stream.write_buffer().clear();
diff --git a/quiche/quic/moqt/test_tools/moqt_session_peer.h b/quiche/quic/moqt/test_tools/moqt_session_peer.h index 65dce82..db84f18 100644 --- a/quiche/quic/moqt/test_tools/moqt_session_peer.h +++ b/quiche/quic/moqt/test_tools/moqt_session_peer.h
@@ -21,6 +21,7 @@ #include "quiche/quic/core/quic_alarm_factory.h" #include "quiche/quic/core/quic_time.h" #include "quiche/quic/moqt/moqt_bidi_stream.h" +#include "quiche/quic/moqt/moqt_error.h" #include "quiche/quic/moqt/moqt_fetch_task.h" #include "quiche/quic/moqt/moqt_key_value_pair.h" #include "quiche/quic/moqt/moqt_live_publisher.h" @@ -60,7 +61,28 @@ std::unique_ptr<MoqtBidiStreamBase> absl_nonnull stream) : stream_(std::move(stream)) {} - MoqtBidiStreamBase& stream() { return *stream_; } + MoqtBidiStreamTestWrapper( + std::unique_ptr<MoqtSession::OutgoingControlStream> absl_nonnull + outgoing_stream, + std::unique_ptr<MoqtSession::IncomingControlStream> absl_nonnull + incoming_stream) + : outgoing_control_stream_(std::move(outgoing_stream)), + incoming_control_stream_(std::move(incoming_stream)) {} + + MoqtBidiStreamBase& stream() { + QUICHE_DCHECK(stream_ != nullptr); + return *stream_; + } + + MoqtSession::OutgoingControlStream& outgoing_control_stream() { + QUICHE_DCHECK(outgoing_control_stream_ != nullptr); + return *outgoing_control_stream_; + } + + MoqtSession::IncomingControlStream& incoming_control_stream() { + QUICHE_DCHECK(incoming_control_stream_ != nullptr); + return *incoming_control_stream_; + } // Simulates receiving the specified control message on the bidi stream. void ReceiveMessage(const AnyMoqtControlMessage& message) { @@ -69,14 +91,31 @@ uint64_t raw_type; ASSERT_TRUE(reader.ReadMoqVarInt(&raw_type)); ASSERT_TRUE(reader.Seek(2)); - absl::Status status = stream_->OnRawControlMessage(MoqtRawControlMessage{ + MoqtRawControlMessage raw_message{ .type = static_cast<MoqtMessageType>(raw_type), - .payload = std::string(reader.ReadRemainingPayload())}); - stream_->CheckStatus(status); + .payload = std::string(reader.ReadRemainingPayload())}; + if (stream_ != nullptr) { + absl::Status status = stream_->OnRawControlMessage(raw_message); + stream_->CheckStatus(status); + return; + } + QUICHE_DCHECK(incoming_control_stream_ != nullptr); + MoqtSession* session = + MoqtSessionFromWeakPtr(incoming_control_stream_->session_); + QUICHE_DCHECK(session != nullptr); + absl::Status status = ControlMessageDispatcher::DispatchControlMessage( + *session, session->ControlMessageParser(), raw_message, "control"); + if (!status.ok()) { + std::optional<MoqtError> error_code = GetMoqtErrorForStatus(status); + session->Error(error_code.value_or(MoqtError::kProtocolViolation), + status.message()); + } } private: - std::unique_ptr<MoqtBidiStreamBase> absl_nonnull stream_; + std::unique_ptr<MoqtBidiStreamBase> stream_; + std::unique_ptr<MoqtSession::OutgoingControlStream> outgoing_control_stream_; + std::unique_ptr<MoqtSession::IncomingControlStream> incoming_control_stream_; }; class OutgoingSubgroupStreamPeer { @@ -92,13 +131,17 @@ static std::unique_ptr<MoqtBidiStreamTestWrapper> CreateControlStream( MoqtSession* session, webtransport::test::MockStream* stream) { - auto new_stream = std::make_unique<MoqtSession::ControlStream>(session); - session->control_stream_ = new_stream->GetWeakPtr(); - new_stream->BindStream(stream); + auto outgoing = + std::make_unique<MoqtSession::OutgoingControlStream>(session, stream); + session->outgoing_control_stream_ = outgoing->GetWeakPtr(); + auto incoming = std::make_unique<MoqtSession::IncomingControlStream>( + session, MoqtStreamTypeParser(stream)); + session->incoming_control_stream_ = incoming->GetWeakPtr(); ON_CALL(*stream, visitor()) - .WillByDefault(::testing::Return(new_stream.get())); + .WillByDefault(::testing::Return(outgoing.get())); ON_CALL(*stream, CanWrite).WillByDefault(::testing::Return(true)); - return std::make_unique<MoqtBidiStreamTestWrapper>(std::move(new_stream)); + return std::make_unique<MoqtBidiStreamTestWrapper>(std::move(outgoing), + std::move(incoming)); } static std::unique_ptr<webtransport::StreamVisitor> @@ -114,20 +157,6 @@ return session->published_subscriptions_.contains(request_id); } - // In the test OnSessionReady, the session creates a stream and then passes - // its unique_ptr to the mock webtransport stream. This function casts - // that unique_ptr into a MoqtSession::Stream*, which is a private class of - // MoqtSession, and then casts again into MoqtParserVisitor so that the test - // can inject packets into that stream. - // This function is useful for any test that wants to inject packets on a - // stream created by the MoqtSession. - static std::unique_ptr<MoqtBidiStreamTestWrapper> - FetchParserVisitorFromWebtransportStreamVisitor( - std::unique_ptr<webtransport::StreamVisitor> visitor) { - return std::make_unique<MoqtBidiStreamTestWrapper>(absl::WrapUnique( - absl::down_cast<MoqtSession::ControlStream*>(visitor.release()))); - } - static void set_next_request_id(MoqtSession* session, uint64_t id) { session->next_request_id_ = id; } @@ -136,6 +165,10 @@ session->peer_max_request_id_ = id; } + static void set_peer_setup_received(MoqtSession* session, bool value) { + session->peer_setup_received_ = value; + } + static MoqtSession::PublishedFetch* GetFetch(MoqtSession* session, uint64_t fetch_id) { auto it = session->incoming_fetches_.find(fetch_id); @@ -178,8 +211,17 @@ return session->parameters_.moqt_implementation; } - static MoqtSession::ControlStream* GetControlStream(MoqtSession* session) { - return session->control_stream_.GetIfAvailable(); + static MoqtSession::OutgoingControlStream* GetOutgoingControlStream( + MoqtSession* session) { + return session->outgoing_control_stream_.GetIfAvailable(); + } + static MoqtSession::IncomingControlStream* GetIncomingControlStream( + MoqtSession* session) { + return session->incoming_control_stream_.GetIfAvailable(); + } + static MoqtSession::OutgoingControlStream* GetControlStream( + MoqtSession* session) { + return session->outgoing_control_stream_.GetIfAvailable(); } static const MoqtSessionParameters& GetParameters(MoqtSession* session) {
diff --git a/quiche/quic/moqt/test_tools/moqt_simulator_harness.cc b/quiche/quic/moqt/test_tools/moqt_simulator_harness.cc index c282858..e6199bc 100644 --- a/quiche/quic/moqt/test_tools/moqt_simulator_harness.cc +++ b/quiche/quic/moqt/test_tools/moqt_simulator_harness.cc
@@ -31,7 +31,7 @@ namespace { MoqtSessionParameters CreateParameters(quic::Perspective perspective, absl::string_view version) { - MoqtSessionParameters parameters(perspective, "", ""); + MoqtSessionParameters parameters(perspective, "/foo", "example.com"); parameters.version = version; parameters.deliver_partial_objects = false; return parameters;