No public description PiperOrigin-RevId: 990542513
diff --git a/build/source_list.bzl b/build/source_list.bzl index dd72020..4825e2f 100644 --- a/build/source_list.bzl +++ b/build/source_list.bzl
@@ -1316,6 +1316,7 @@ "quic/core/http/quic_spdy_stream_test.cc", "quic/core/http/spdy_utils_test.cc", "quic/core/http/web_transport_http3_test.cc", + "quic/core/http/web_transport_stream_adapter_test.cc", "quic/core/legacy_quic_stream_id_manager_test.cc", "quic/core/packet_number_indexed_queue_test.cc", "quic/core/qpack/new_qpack_blocking_manager_test.cc",
diff --git a/build/source_list.gni b/build/source_list.gni index 7640e1a..8d1a974 100644 --- a/build/source_list.gni +++ b/build/source_list.gni
@@ -1318,6 +1318,7 @@ "src/quiche/quic/core/http/quic_spdy_stream_test.cc", "src/quiche/quic/core/http/spdy_utils_test.cc", "src/quiche/quic/core/http/web_transport_http3_test.cc", + "src/quiche/quic/core/http/web_transport_stream_adapter_test.cc", "src/quiche/quic/core/legacy_quic_stream_id_manager_test.cc", "src/quiche/quic/core/packet_number_indexed_queue_test.cc", "src/quiche/quic/core/qpack/new_qpack_blocking_manager_test.cc",
diff --git a/build/source_list.json b/build/source_list.json index 4596010..6aeb6ba 100644 --- a/build/source_list.json +++ b/build/source_list.json
@@ -1317,6 +1317,7 @@ "quiche/quic/core/http/quic_spdy_stream_test.cc", "quiche/quic/core/http/spdy_utils_test.cc", "quiche/quic/core/http/web_transport_http3_test.cc", + "quiche/quic/core/http/web_transport_stream_adapter_test.cc", "quiche/quic/core/legacy_quic_stream_id_manager_test.cc", "quiche/quic/core/packet_number_indexed_queue_test.cc", "quiche/quic/core/qpack/new_qpack_blocking_manager_test.cc",
diff --git a/quiche/quic/core/http/web_transport_stream_adapter.cc b/quiche/quic/core/http/web_transport_stream_adapter.cc index 5546930..8636cd4 100644 --- a/quiche/quic/core/http/web_transport_stream_adapter.cc +++ b/quiche/quic/core/http/web_transport_stream_adapter.cc
@@ -73,14 +73,24 @@ const absl::Status initial_check_status = CheckBeforeStreamWrite(); if (!initial_check_status.ok() && !(initial_check_status.code() == absl::StatusCode::kUnavailable && - options.buffer_unconditionally())) { + (options.buffer_unconditionally() || options.atomic_write()))) { return initial_check_status; } - size_t total_size = MemSliceSpanTotalSize(data); + const QuicByteCount total_size = MemSliceSpanTotalSize(data); + if (options.atomic_write()) { + const QuicByteCount buffered = stream_->BufferedDataBytes(); + const QuicByteCount send_window = stream_->CalculateSendWindowSize(); + if (buffered > send_window || total_size > send_window - buffered) { + return absl::ResourceExhaustedError( + "Insufficient flow control credit for atomic write"); + } + } + QuicConsumedData consumed = stream_->WriteMemSlices( data, /*fin=*/options.send_fin(), - /*buffer_unconditionally=*/options.buffer_unconditionally()); + /*buffer_unconditionally=*/options.buffer_unconditionally() || + options.atomic_write()); if (consumed.bytes_consumed == total_size) { return absl::OkStatus();
diff --git a/quiche/quic/core/http/web_transport_stream_adapter.h b/quiche/quic/core/http/web_transport_stream_adapter.h index 91192bf..ea38c6e 100644 --- a/quiche/quic/core/http/web_transport_stream_adapter.h +++ b/quiche/quic/core/http/web_transport_stream_adapter.h
@@ -41,6 +41,7 @@ ABSL_MUST_USE_RESULT ReadResult Read(std::string* output) override; absl::Status Writev(absl::Span<quiche::QuicheMemSlice> data, const webtransport::StreamWriteOptions& options) override; + bool SupportsAtomicWrites() const override { return true; } bool CanWrite() const override; size_t ReadableBytes() const override; PeekResult PeekNextReadableRegion() const override;
diff --git a/quiche/quic/core/http/web_transport_stream_adapter_test.cc b/quiche/quic/core/http/web_transport_stream_adapter_test.cc new file mode 100644 index 0000000..9b0e262 --- /dev/null +++ b/quiche/quic/core/http/web_transport_stream_adapter_test.cc
@@ -0,0 +1,165 @@ +// Copyright 2026 The Chromium Authors +// Use of this source code is governed by a BSD-style license that can be +// found in the LICENSE file. + +#include "quiche/quic/core/http/web_transport_stream_adapter.h" + +#include <array> +#include <memory> +#include <optional> + +#include "absl/status/status.h" +#include "absl/types/span.h" +#include "quiche/quic/core/crypto/null_encrypter.h" +#include "quiche/quic/core/quic_stream.h" +#include "quiche/quic/core/quic_types.h" +#include "quiche/quic/core/quic_versions.h" +#include "quiche/quic/platform/api/quic_test.h" +#include "quiche/quic/test_tools/quic_config_peer.h" +#include "quiche/quic/test_tools/quic_flow_controller_peer.h" +#include "quiche/quic/test_tools/quic_stream_peer.h" +#include "quiche/quic/test_tools/quic_test_utils.h" +#include "quiche/common/quiche_mem_slice.h" + +using testing::_; +using testing::AnyNumber; +using testing::Return; +using testing::StrictMock; + +namespace quic { +namespace test { +namespace { + +class TestStream : public QuicStream { + public: + TestStream(QuicStreamId id, QuicSession* session) + : QuicStream(id, session, /*is_static=*/false, BIDIRECTIONAL) { + sequencer()->set_level_triggered(true); + } + + MOCK_METHOD(void, OnDataAvailable, (), (override)); + MOCK_METHOD(void, OnCanWriteNewData, (), (override)); + MOCK_METHOD(void, OnWriteSideInDataRecvdState, (), (override)); + + using QuicStream::WriteOrBufferData; +}; + +class WebTransportStreamAdapterTest + : public QuicTestWithParam<ParsedQuicVersion> { + protected: + void Initialize() { + connection_ = new StrictMock<MockQuicConnection>( + &helper_, &alarm_factory_, Perspective::IS_SERVER, + ParsedQuicVersionVector{GetParam()}); + session_ = std::make_unique<MockQuicSession>(connection_); + session_->Initialize(); + connection_->SetEncrypter( + ENCRYPTION_FORWARD_SECURE, + std::make_unique<NullEncrypter>(connection_->perspective())); + QuicConfigPeer::SetReceivedInitialSessionFlowControlWindow( + session_->config(), kMinimumFlowControlSendWindow); + QuicConfigPeer::SetReceivedInitialMaxStreamDataBytesIncomingBidirectional( + session_->config(), kMinimumFlowControlSendWindow); + session_->OnConfigNegotiated(); + + stream_ = new StrictMock<TestStream>(kTestStreamId, session_.get()); + session_->ActivateStream(absl::WrapUnique(stream_)); + EXPECT_CALL(*session_, ShouldKeepConnectionAlive()) + .WillRepeatedly(Return(true)); + EXPECT_CALL(*session_, MaybeSendStopSendingFrame(kTestStreamId, _)) + .Times(AnyNumber()); + EXPECT_CALL(*session_, MaybeSendRstStreamFrame(kTestStreamId, _, _)) + .Times(AnyNumber()); + adapter_ = std::make_unique<WebTransportStreamAdapter>( + session_.get(), stream_, QuicStreamPeer::sequencer(stream_), + std::nullopt); + } + + MockQuicConnectionHelper helper_; + MockAlarmFactory alarm_factory_; + MockQuicConnection* connection_ = nullptr; + std::unique_ptr<MockQuicSession> session_; + StrictMock<TestStream>* stream_ = nullptr; + std::unique_ptr<WebTransportStreamAdapter> adapter_; + const QuicStreamId kTestStreamId = GetNthClientInitiatedBidirectionalStreamId( + GetParam().transport_version, 1); +}; + +INSTANTIATE_TEST_SUITE_P(WebTransportStreamAdapterTests, + WebTransportStreamAdapterTest, + ::testing::ValuesIn(CurrentSupportedHttp3Versions()), + ::testing::PrintToStringParamName()); + +TEST_P(WebTransportStreamAdapterTest, AtomicWriteRequiresStreamCredit) { + Initialize(); + QuicStreamPeer::SetSendWindowOffset(stream_, 4); + std::array slices = {quiche::QuicheMemSlice::Copy("12345")}; + webtransport::StreamWriteOptions options; + options.set_atomic_write(true); + + EXPECT_EQ(absl::StatusCode::kResourceExhausted, + adapter_->Writev(absl::MakeSpan(slices), options).code()); + EXPECT_EQ(0u, stream_->BufferedDataBytes()); +} + +TEST_P(WebTransportStreamAdapterTest, AtomicWriteAcceptsExactStreamCredit) { + Initialize(); + QuicStreamPeer::SetSendWindowOffset(stream_, 4); + std::array slices = {quiche::QuicheMemSlice::Copy("1234")}; + EXPECT_CALL(*session_, WritevData(kTestStreamId, 4, 0, _, _, _)) + .WillOnce(Return(QuicConsumedData(4, false))); + EXPECT_CALL(*session_, SendBlocked(_, 4)); + webtransport::StreamWriteOptions options; + options.set_atomic_write(true); + + EXPECT_TRUE(adapter_->Writev(absl::MakeSpan(slices), options).ok()); + EXPECT_EQ(0u, stream_->BufferedDataBytes()); +} + +TEST_P(WebTransportStreamAdapterTest, AtomicWriteAccountsForBufferedData) { + Initialize(); + EXPECT_CALL(*session_, WritevData(kTestStreamId, _, _, _, _, _)) + .WillOnce(Return(QuicConsumedData(0, false))); + stream_->WriteOrBufferData("12", false, nullptr); + ASSERT_EQ(2u, stream_->BufferedDataBytes()); + QuicStreamPeer::SetSendWindowOffset(stream_, 4); + std::array slices = {quiche::QuicheMemSlice::Copy("345")}; + webtransport::StreamWriteOptions options; + options.set_atomic_write(true); + + EXPECT_EQ(absl::StatusCode::kResourceExhausted, + adapter_->Writev(absl::MakeSpan(slices), options).code()); + EXPECT_EQ(2u, stream_->BufferedDataBytes()); +} + +TEST_P(WebTransportStreamAdapterTest, AtomicWriteRequiresConnectionCredit) { + Initialize(); + QuicStreamPeer::SetSendWindowOffset(stream_, 10); + QuicFlowControllerPeer::SetSendWindowOffset(session_->flow_controller(), 3); + std::array slices = {quiche::QuicheMemSlice::Copy("1234")}; + webtransport::StreamWriteOptions options; + options.set_atomic_write(true); + + EXPECT_EQ(absl::StatusCode::kResourceExhausted, + adapter_->Writev(absl::MakeSpan(slices), options).code()); + EXPECT_EQ(0u, stream_->BufferedDataBytes()); +} + +TEST_P(WebTransportStreamAdapterTest, AtomicWriteCanSendFin) { + Initialize(); + QuicStreamPeer::SetSendWindowOffset(stream_, 4); + std::array slices = {quiche::QuicheMemSlice::Copy("1234")}; + EXPECT_CALL(*session_, WritevData(kTestStreamId, 4, 0, _, _, _)) + .WillOnce(Return(QuicConsumedData(4, true))); + EXPECT_CALL(*session_, SendBlocked(_, 4)); + webtransport::StreamWriteOptions options; + options.set_atomic_write(true); + options.set_send_fin(true); + + EXPECT_TRUE(adapter_->Writev(absl::MakeSpan(slices), options).ok()); + EXPECT_TRUE(stream_->fin_sent()); +} + +} // namespace +} // namespace test +} // namespace quic
diff --git a/quiche/web_transport/encapsulated/encapsulated_web_transport.cc b/quiche/web_transport/encapsulated/encapsulated_web_transport.cc index b0eb915..a879870 100644 --- a/quiche/web_transport/encapsulated/encapsulated_web_transport.cc +++ b/quiche/web_transport/encapsulated/encapsulated_web_transport.cc
@@ -610,6 +610,9 @@ absl::Status EncapsulatedSession::InnerStream::Writev( const absl::Span<quiche::QuicheMemSlice> data, const StreamWriteOptions& options) { + if (options.atomic_write()) { + return absl::UnimplementedError("Atomic writes are not supported"); + } if (write_side_closed_) { return absl::FailedPreconditionError( "Trying to write into an already-closed stream");
diff --git a/quiche/web_transport/encapsulated/encapsulated_web_transport_test.cc b/quiche/web_transport/encapsulated/encapsulated_web_transport_test.cc index c8eddf0..fe96324 100644 --- a/quiche/web_transport/encapsulated/encapsulated_web_transport_test.cc +++ b/quiche/web_transport/encapsulated/encapsulated_web_transport_test.cc
@@ -623,6 +623,30 @@ EXPECT_FALSE(stream->CanWrite()); } +TEST_F(EncapsulatedWebTransportTest, AtomicWriteUnsupported) { + std::unique_ptr<EncapsulatedSession> session = + CreateTransport(Perspective::kClient); + DefaultHandshakeForClient(*session); + Stream* stream = session->OpenOutgoingUnidirectionalStream(); + ASSERT_TRUE(stream != nullptr); + + std::array slices = {quiche::QuicheMemSlice::Copy("test")}; + StreamWriteOptions options; + options.set_atomic_write(true); + options.set_send_fin(true); + EXPECT_FALSE(stream->SupportsAtomicWrites()); + EXPECT_THAT(stream->Writev(absl::MakeSpan(slices), options), + StatusIs(absl::StatusCode::kUnimplemented)); + + EXPECT_CALL(*this, OnCapsule(_)).WillOnce([](const Capsule& capsule) { + EXPECT_EQ(capsule.capsule_type(), CapsuleType::WT_STREAM); + EXPECT_EQ(capsule.web_transport_stream_data().data, "accepted"); + return true; + }); + EXPECT_THAT(WriteIntoStream(*stream, "accepted"), + StatusIs(absl::StatusCode::kOk)); +} + TEST_F(EncapsulatedWebTransportTest, FinOnlyWrite) { std::unique_ptr<EncapsulatedSession> session = CreateTransport(Perspective::kClient);
diff --git a/quiche/web_transport/test_tools/in_memory_stream.cc b/quiche/web_transport/test_tools/in_memory_stream.cc index ad2236e..421d42b 100644 --- a/quiche/web_transport/test_tools/in_memory_stream.cc +++ b/quiche/web_transport/test_tools/in_memory_stream.cc
@@ -60,6 +60,9 @@ absl::Status InMemoryStream::Writev( absl::Span<quiche::QuicheMemSlice> data, const webtransport::StreamWriteOptions& options) { + if (options.atomic_write()) { + return absl::UnimplementedError("Atomic writes are not supported"); + } absl::Status status = GetWriteStatusWithExtraChecks(/*is_write=*/true); if (!status.ok()) { return status;
diff --git a/quiche/web_transport/test_tools/in_memory_stream_test.cc b/quiche/web_transport/test_tools/in_memory_stream_test.cc index e905eed..1117aaa 100644 --- a/quiche/web_transport/test_tools/in_memory_stream_test.cc +++ b/quiche/web_transport/test_tools/in_memory_stream_test.cc
@@ -133,5 +133,17 @@ EXPECT_FALSE(stream.CanWrite()); } +TEST(InMemoryStreamTest, AtomicWriteUnsupportedByDefault) { + InMemoryStreamWithWriteBuffer stream(0); + std::array write_vector = {quiche::QuicheMemSlice::Copy("foo")}; + StreamWriteOptions options; + options.set_atomic_write(true); + + EXPECT_FALSE(stream.SupportsAtomicWrites()); + EXPECT_THAT(stream.Writev(absl::MakeSpan(write_vector), options), + StatusIs(absl::StatusCode::kUnimplemented)); + EXPECT_TRUE(stream.write_buffer().empty()); +} + } // namespace } // namespace webtransport::test
diff --git a/quiche/web_transport/test_tools/mock_web_transport.h b/quiche/web_transport/test_tools/mock_web_transport.h index 3172bd7..0c210e5 100644 --- a/quiche/web_transport/test_tools/mock_web_transport.h +++ b/quiche/web_transport/test_tools/mock_web_transport.h
@@ -43,6 +43,7 @@ (absl::Span<quiche::QuicheMemSlice> data, const StreamWriteOptions& options), (override)); + MOCK_METHOD(bool, SupportsAtomicWrites, (), (const, override)); MOCK_METHOD(PeekResult, PeekNextReadableRegion, (), (const, override)); MOCK_METHOD(bool, SkipBytes, (size_t bytes), (override)); MOCK_METHOD(bool, CanWrite, (), (const, override));
diff --git a/quiche/web_transport/web_transport.h b/quiche/web_transport/web_transport.h index 302780c..50ffbde 100644 --- a/quiche/web_transport/web_transport.h +++ b/quiche/web_transport/web_transport.h
@@ -179,9 +179,20 @@ buffer_unconditionally_ = value; } + // If atomic_write() is true, the write is accepted only if the ordered + // stream prefix through the end of `data` fits within the current stream and + // connection flow control windows. This is a snapshot check, not a + // reservation of connection credit. The write may bypass the normal + // buffering limit even when CanWrite() is false. If the check fails, + // Writev() returns RESOURCE_EXHAUSTED without consuming `data`. + // Implementations that do not support atomic writes return UNIMPLEMENTED. + bool atomic_write() const { return atomic_write_; } + void set_atomic_write(bool value) { atomic_write_ = value; } + private: bool send_fin_ = false; bool buffer_unconditionally_ = false; + bool atomic_write_ = false; }; inline constexpr StreamWriteOptions kDefaultStreamWriteOptions = @@ -255,6 +266,9 @@ virtual absl::Status Writev(absl::Span<quiche::QuicheMemSlice> data, const StreamWriteOptions& options) = 0; + // Indicates whether Writev() supports StreamWriteOptions::atomic_write(). + virtual bool SupportsAtomicWrites() const { return false; } + // Indicates whether it is possible to write into stream right now. virtual bool CanWrite() const = 0;