Avoid sending EndOfGroup object with payload/properties to MoqtFramer. Fixes a QUICHE_BUG in interop. PiperOrigin-RevId: 998746058
diff --git a/quiche/quic/moqt/moqt_uni_stream.cc b/quiche/quic/moqt/moqt_uni_stream.cc index 4b11ed9..0918a7a 100644 --- a/quiche/quic/moqt/moqt_uni_stream.cc +++ b/quiche/quic/moqt/moqt_uni_stream.cc
@@ -31,6 +31,7 @@ #include "quiche/quic/moqt/moqt_types.h" #include "quiche/common/platform/api/quiche_bug_tracker.h" #include "quiche/common/quiche_buffer_allocator.h" +#include "quiche/common/quiche_checked_math.h" #include "quiche/common/quiche_mem_slice.h" #include "quiche/common/quiche_weak_ptr.h" #include "quiche/web_transport/stream_helpers.h" @@ -55,7 +56,18 @@ header.properties = object.metadata.properties; header.object_status = object.metadata.status; header.payload_length = object.metadata.payload_length; - + bool separate_end_of_group = false; + if (type.IsSubgroup() && !type.EndOfGroupInStream() && + object.fin_after_this && + object.metadata.status == MoqtObjectStatus::kEndOfGroup && + (object.metadata.payload_length > 0 || + !object.metadata.properties.empty())) { + // Handle the special case where an EndOfGroup is encoded in an object with + // payload or properties. It cannot be expressed that way on the wire; + // create a notional object instead. + separate_end_of_group = true; + header.object_status = MoqtObjectStatus::kNormal; + } quiche::QuicheBuffer serialized_header = framer_.SerializeObjectHeader(header, type, last_object_); std::vector<quiche::QuicheMemSlice> write_vector; @@ -64,6 +76,18 @@ for (auto& slice : object.payload) { write_vector.push_back(std::move(slice)); } + if (separate_end_of_group) { + last_object_ = object.metadata; + if (quiche::SafeIncrementBy<uint64_t>(header.object_id, 1)) { + header.object_status = MoqtObjectStatus::kEndOfGroup; + header.payload_length = 0; + header.properties.clear(); + serialized_header = + framer_.SerializeObjectHeader(header, type, last_object_); + write_vector.push_back( + quiche::QuicheMemSlice(std::move(serialized_header))); + } + } webtransport::StreamWriteOptions options; options.set_send_fin(!type.IsFetch() && object.fin_after_this); absl::Status write_status = @@ -76,8 +100,7 @@ return false; } QUIC_DVLOG(1) << "Stream " << stream_.GetStreamId() << " successfully wrote " - << object.metadata.location - << ", fin = " << object.fin_after_this; + << object.metadata.location; return true; }
diff --git a/quiche/quic/moqt/moqt_uni_stream_test.cc b/quiche/quic/moqt/moqt_uni_stream_test.cc index f6a62ce..ad4ab8a 100644 --- a/quiche/quic/moqt/moqt_uni_stream_test.cc +++ b/quiche/quic/moqt/moqt_uni_stream_test.cc
@@ -5,6 +5,7 @@ #include "quiche/quic/moqt/moqt_uni_stream.h" #include <cstdint> +#include <limits> #include <memory> #include <optional> #include <string> @@ -300,6 +301,129 @@ stream_->OnCanWrite(); } +TEST_F(OutgoingSubgroupStreamTest, EndOfGroupWithPayload) { + PublishedObject obj0 = DefaultObject(); + obj0.metadata.status = MoqtObjectStatus::kEndOfGroup; + obj0.fin_after_this = true; + EXPECT_CALL(mock_stream_, CanWrite()) + .WillOnce(Return(true)) + .WillOnce(Return(false)); + EXPECT_CALL(*track_publisher_, GetCachedObject(0, Optional(0), 0, 0)) + .WillOnce(Return(std::move(obj0))); + EXPECT_CALL(visitor_, InWindow(Location(0, 0))).WillOnce(Return(true)); + EXPECT_CALL(visitor_, object_delivery_timeout()) + .WillOnce(Return(quic::QuicTimeDelta::FromSeconds(1))); + EXPECT_CALL(visitor_, clock()).WillOnce(Return(&mock_clock_)); + EXPECT_CALL(*track_publisher_, properties()) + .WillRepeatedly(ReturnRef(track_properties_)); + { + testing::InSequence seq; + EXPECT_CALL(mock_stream_, Writev) + .WillOnce([](absl::Span<quiche::QuicheMemSlice> data, + const webtransport::StreamWriteOptions& options) { + EXPECT_EQ(data.size(), 3); + EXPECT_EQ(data[1].AsStringView(), "payload"); + EXPECT_TRUE(options.send_fin()); + return absl::OkStatus(); + }); + } + EXPECT_CALL(visitor_, OnObjectSent(Location(0, 0))); + stream_->OnCanWrite(); +} + +TEST_F(OutgoingSubgroupStreamTest, EndOfGroupWithProperties) { + PublishedObject obj0 = DefaultObject(); + obj0.metadata.status = MoqtObjectStatus::kEndOfGroup; + obj0.metadata.payload_length = 0; + obj0.payload.clear(); + obj0.metadata.properties = "prop"; + obj0.fin_after_this = true; + EXPECT_CALL(mock_stream_, CanWrite()) + .WillOnce(Return(true)) + .WillOnce(Return(false)); + EXPECT_CALL(*track_publisher_, GetCachedObject(0, Optional(0), 0, 0)) + .WillOnce(Return(std::move(obj0))); + EXPECT_CALL(visitor_, InWindow(Location(0, 0))).WillOnce(Return(true)); + EXPECT_CALL(visitor_, object_delivery_timeout()) + .WillOnce(Return(quic::QuicTimeDelta::FromSeconds(1))); + EXPECT_CALL(visitor_, clock()).WillOnce(Return(&mock_clock_)); + EXPECT_CALL(*track_publisher_, properties()) + .WillRepeatedly(ReturnRef(track_properties_)); + { + testing::InSequence seq; + EXPECT_CALL(mock_stream_, Writev) + .WillOnce([](absl::Span<quiche::QuicheMemSlice> data, + const webtransport::StreamWriteOptions& options) { + EXPECT_EQ(data.size(), 2); + EXPECT_TRUE(options.send_fin()); + return absl::OkStatus(); + }); + } + EXPECT_CALL(visitor_, OnObjectSent(Location(0, 0))); + stream_->OnCanWrite(); +} + +TEST_F(OutgoingSubgroupStreamTest, EndOfGroupEmptyPayloadAndProperties) { + PublishedObject obj0 = DefaultObject(); + obj0.metadata.status = MoqtObjectStatus::kEndOfGroup; + obj0.metadata.payload_length = 0; + obj0.payload.clear(); + obj0.fin_after_this = true; + EXPECT_CALL(mock_stream_, CanWrite()) + .WillOnce(Return(true)) + .WillOnce(Return(false)); + EXPECT_CALL(*track_publisher_, GetCachedObject(0, Optional(0), 0, 0)) + .WillOnce(Return(std::move(obj0))); + EXPECT_CALL(visitor_, InWindow(Location(0, 0))).WillOnce(Return(true)); + EXPECT_CALL(visitor_, object_delivery_timeout()) + .WillOnce(Return(quic::QuicTimeDelta::FromSeconds(1))); + EXPECT_CALL(visitor_, clock()).WillOnce(Return(&mock_clock_)); + EXPECT_CALL(*track_publisher_, properties()) + .WillRepeatedly(ReturnRef(track_properties_)); + EXPECT_CALL(mock_stream_, Writev) + .WillOnce([](absl::Span<quiche::QuicheMemSlice> data, + const webtransport::StreamWriteOptions& options) { + EXPECT_EQ(data.size(), 1); + EXPECT_TRUE(options.send_fin()); + return absl::OkStatus(); + }); + EXPECT_CALL(visitor_, OnObjectSent(Location(0, 0))); + stream_->OnCanWrite(); +} + +TEST_F(OutgoingSubgroupStreamTest, EndOfGroupWithPayloadMaxObjectId) { + constexpr uint64_t kMaxUint64 = std::numeric_limits<uint64_t>::max(); + EXPECT_CALL(visitor_, OnDataStreamDestroyed(index_)); + CreateStream(kMaxUint64); + + PublishedObject obj = DefaultObject(); + obj.metadata.location = Location(0, kMaxUint64); + obj.metadata.status = MoqtObjectStatus::kEndOfGroup; + obj.fin_after_this = true; + EXPECT_CALL(mock_stream_, CanWrite()) + .WillOnce(Return(true)) + .WillOnce(Return(false)); + EXPECT_CALL(*track_publisher_, GetCachedObject(0, Optional(0), kMaxUint64, 0)) + .WillOnce(Return(std::move(obj))); + EXPECT_CALL(visitor_, InWindow(Location(0, kMaxUint64))) + .WillOnce(Return(true)); + EXPECT_CALL(visitor_, object_delivery_timeout()) + .WillOnce(Return(quic::QuicTimeDelta::FromSeconds(1))); + EXPECT_CALL(visitor_, clock()).WillOnce(Return(&mock_clock_)); + EXPECT_CALL(*track_publisher_, properties()) + .WillRepeatedly(ReturnRef(track_properties_)); + EXPECT_CALL(mock_stream_, Writev) + .WillOnce([](absl::Span<quiche::QuicheMemSlice> data, + const webtransport::StreamWriteOptions& options) { + EXPECT_EQ(data.size(), 2); + EXPECT_EQ(data[1].AsStringView(), "payload"); + EXPECT_TRUE(options.send_fin()); + return absl::OkStatus(); + }); + EXPECT_CALL(visitor_, OnObjectSent(Location(0, kMaxUint64))); + stream_->OnCanWrite(); +} + class OutgoingFetchStreamTest : public quic::test::QuicTest { public: OutgoingFetchStreamTest()