Always send STOP_SENDING on Reset() unless RESET has already been received. The MOQT spec says that cancellation is achieved via STOP_SENDING, not RESET. Review all instances for appropriate use of the two frames. PiperOrigin-RevId: 991668872
diff --git a/quiche/quic/moqt/moqt_bidi_stream.h b/quiche/quic/moqt/moqt_bidi_stream.h index fef0eeb..0e59ab9 100644 --- a/quiche/quic/moqt/moqt_bidi_stream.h +++ b/quiche/quic/moqt/moqt_bidi_stream.h
@@ -72,6 +72,7 @@ // webtransport::StreamVisitor implementation. void OnResetStreamReceived(webtransport::StreamErrorCode error) override { + stream_status_ = MoqtStreamErrorToStatus(error, ""); Reset(error); } void OnStopSendingReceived(webtransport::StreamErrorCode error) override { @@ -116,6 +117,9 @@ } void Reset(webtransport::StreamErrorCode error) { if (stream() != nullptr) { + if (stream_status().ok()) { // a Reset has not been received yet. + stream()->SendStopSending(error); + } stream()->ResetWithUserCode(error); } stream_status_ = MoqtStreamErrorToStatus(error, "");
diff --git a/quiche/quic/moqt/moqt_bidi_stream_test.cc b/quiche/quic/moqt/moqt_bidi_stream_test.cc index bc3feae..e4b3973 100644 --- a/quiche/quic/moqt/moqt_bidi_stream_test.cc +++ b/quiche/quic/moqt/moqt_bidi_stream_test.cc
@@ -80,6 +80,7 @@ TEST_F(MoqtBidiStreamTest, Reset) { stream_->BindStream(&mock_stream_); + EXPECT_CALL(mock_stream_, SendStopSending(1234)); EXPECT_CALL(mock_stream_, ResetWithUserCode(1234)); stream_->Reset(1234); EXPECT_TRUE(stream_->detached_); @@ -87,6 +88,7 @@ TEST_F(MoqtBidiStreamTest, IncomingReset) { stream_->BindStream(&mock_stream_); + EXPECT_CALL(mock_stream_, SendStopSending).Times(0); EXPECT_CALL(mock_stream_, ResetWithUserCode(1234)); stream_->OnResetStreamReceived(1234); EXPECT_TRUE(stream_->detached_); @@ -100,6 +102,7 @@ TEST_F(MoqtBidiStreamTest, IncomingStopSending) { stream_->BindStream(&mock_stream_); + EXPECT_CALL(mock_stream_, SendStopSending(1234)); EXPECT_CALL(mock_stream_, ResetWithUserCode(1234)); stream_->OnStopSendingReceived(1234); EXPECT_TRUE(stream_->detached_);
diff --git a/quiche/quic/moqt/moqt_session.cc b/quiche/quic/moqt/moqt_session.cc index 30006fe..b62d294 100644 --- a/quiche/quic/moqt/moqt_session.cc +++ b/quiche/quic/moqt/moqt_session.cc
@@ -176,7 +176,8 @@ // Immediately reject new requests with REQUEST_ERROR. If the stream // cannot be written, just reset it. if (!stream->CanWrite()) { - stream->ResetWithUserCode(kResetCodeSessionClosed); + stream->SendStopSending(kResetCodeGoingAway); + stream->ResetWithUserCode(kResetCodeGoingAway); continue; } webtransport::StreamWriteOptions options; @@ -185,7 +186,8 @@ quiche::QuicheMemSlice(framer_.SerializeRequestError(MoqtRequestError{ RequestErrorCode::kGoingAway, std::nullopt, ""}))}; if (!stream->Writev(absl::MakeSpan(write_vector), options).ok()) { - stream->ResetWithUserCode(kResetCodeSessionClosed); + stream->SendStopSending(kResetCodeGoingAway); + stream->ResetWithUserCode(kResetCodeGoingAway); }; continue; } @@ -949,6 +951,7 @@ 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_->SendStopSending(kResetCodeInternalError); stream_->ResetWithUserCode(kResetCodeInternalError); return; } @@ -1300,7 +1303,7 @@ return; } if (!type.ok()) { - stream_->ResetWithUserCode(kResetCodeInternalError); + stream_->SendStopSending(kResetCodeInternalError); return; } if (*type == static_cast<uint64_t>(MoqtMessageType::kSetup)) {
diff --git a/quiche/quic/moqt/moqt_session_test.cc b/quiche/quic/moqt/moqt_session_test.cc index bbed4a3..24da4bf 100644 --- a/quiche/quic/moqt/moqt_session_test.cc +++ b/quiche/quic/moqt/moqt_session_test.cc
@@ -683,6 +683,7 @@ }); bidi_wrapper_->ReceiveMessage(ok); + EXPECT_CALL(mock_bidi_stream_, SendStopSending); EXPECT_CALL(mock_bidi_stream_, ResetWithUserCode); session_.PublishNamespaceDone(TrackNamespace{"foo"}); // State is gone. @@ -1050,10 +1051,12 @@ Writev(ControlMessageOfType(MoqtMessageType::kSubscribe), _)); EXPECT_TRUE( session_.Subscribe(ftn, &remote_track_visitor_, MessageParameters())); + EXPECT_CALL(mock_bidi_stream_, SendStopSending); EXPECT_CALL(mock_bidi_stream_, ResetWithUserCode); EXPECT_CALL(remote_track_visitor_, OnPublishDone); session_.Unsubscribe(ftn); // Verify it was destroyed. + EXPECT_CALL(mock_bidi_stream_, SendStopSending).Times(0); EXPECT_CALL(mock_bidi_stream_, ResetWithUserCode).Times(0); EXPECT_CALL(remote_track_visitor_, OnPublishDone).Times(0); session_.Unsubscribe(ftn); @@ -1168,6 +1171,7 @@ }); bidi_wrapper_->ReceiveMessage(MoqtRequestOk()); EXPECT_TRUE(got_callback); + EXPECT_CALL(mock_bidi_stream_, SendStopSending); EXPECT_CALL(mock_bidi_stream_, ResetWithUserCode); } @@ -2098,6 +2102,7 @@ EXPECT_EQ(fetch_task->GetNextObject(object), MoqtFetchTask::GetNextObjectResult::kPending); // Cancel the fetch. + EXPECT_CALL(mock_bidi_stream_, SendStopSending(kResetCodeCancelled)); EXPECT_CALL(mock_bidi_stream_, ResetWithUserCode(kResetCodeCancelled)); fetch_task.reset(); } @@ -2411,6 +2416,7 @@ .WillOnce(Return(&new_request_stream_2)) .WillOnce(Return(nullptr)); EXPECT_CALL(new_request_stream_2, CanWrite()).WillOnce(Return(false)); + EXPECT_CALL(new_request_stream_2, SendStopSending); EXPECT_CALL(new_request_stream_2, ResetWithUserCode); session_.OnIncomingBidirectionalStreamAvailable(); @@ -2887,6 +2893,7 @@ std::unique_ptr<MoqtBidiStreamTestWrapper> publish_wrapper = std::make_unique<MoqtBidiStreamTestWrapper>( ResponseStream(kPublishByte, &publish_stream)); + EXPECT_CALL(mock_bidi_stream_, SendStopSending(kResetCodeCancelled)); EXPECT_CALL(mock_bidi_stream_, ResetWithUserCode(kResetCodeCancelled)); MoqtRequestOk expected_request_ok; expected_request_ok.parameters = parameters; // params from the SUBSCRIBE.