Account for partial object delivery in MoqtRelayTrackPublisher::group.next_object. For EndOfGroup streams, this is triggering malformed tracks because there's a Malformed Track check where it's the end of the group and group.next_object > the object location. PiperOrigin-RevId: 1001156696
diff --git a/quiche/quic/moqt/moqt_relay_track_publisher.cc b/quiche/quic/moqt/moqt_relay_track_publisher.cc index 0bc1e11..ed3000f 100644 --- a/quiche/quic/moqt/moqt_relay_track_publisher.cc +++ b/quiche/quic/moqt/moqt_relay_track_publisher.cc
@@ -224,6 +224,11 @@ if (!duplicate_object->Append(offset, object)) { return; } + if ((duplicate_object->payload_received() == + duplicate_object->metadata().payload_length) && + group.next_object == metadata.location.object) { + group.next_object = metadata.location.object + 1; + } // Data added to the object. Notify listeners. for (MoqtObjectListener* listener : listeners_) { listener->OnNewObjectAvailable(metadata.location, metadata.subgroup, @@ -242,7 +247,9 @@ } next_location_ = metadata.location.Next(); } - if (metadata.location.object >= group.next_object) { + if (metadata.location.object > group.next_object || + (metadata.location.object == group.next_object && + offset + object.length() == metadata.payload_length)) { group.next_object = metadata.location.object + 1; } switch (metadata.status) {
diff --git a/quiche/quic/moqt/moqt_relay_track_publisher_test.cc b/quiche/quic/moqt/moqt_relay_track_publisher_test.cc index 46146a5..0cafdeb 100644 --- a/quiche/quic/moqt/moqt_relay_track_publisher_test.cc +++ b/quiche/quic/moqt/moqt_relay_track_publisher_test.cc
@@ -461,7 +461,7 @@ TEST_F(MoqtRelayTrackPublisherTest, ObjectArrivalInFragments) { SubscribeAndOk(); - Location location = kLargestLocation.Next(); + Location location(4, 0); uint64_t subgroup = 0; // Total size is 15 bytes. PublishedObjectMetadata metadata = {location, subgroup, @@ -489,7 +489,7 @@ } EXPECT_EQ(payload, "frag1frag2"); - // Fragment 3 arrives. + // Fragment 3 arrives, completing (4, 0) and advancing next_object to 1. EXPECT_CALL(listener_, OnNewObjectAvailable(location, Optional(subgroup), 128)); publisher_.OnObjectFragment(kTrackName, metadata, "frag3", 10); @@ -503,6 +503,30 @@ payload += std::string(slice.AsStringView()); } EXPECT_EQ(payload, "frag3"); + + // Next object (4, 1) arrives in two fragments on an EndOfGroup stream: + // first fragment has kNormal, final fragment has kEndOfGroup. + location = location.Next(); + metadata = {location, subgroup, "", MoqtObjectStatus::kNormal, + 128, false, 10}; + EXPECT_CALL(listener_, + OnNewObjectAvailable(location, Optional(subgroup), 128)); + publisher_.OnObjectFragment(kTrackName, metadata, "frag4", 0); + + metadata.status = MoqtObjectStatus::kEndOfGroup; + EXPECT_CALL(listener_, + OnNewObjectAvailable(location, Optional(subgroup), 128)); + publisher_.OnObjectFragment(kTrackName, metadata, "frag5", 5); + EXPECT_FALSE(track_deleted_); + + // Verify next_object was updated to 2 by sending an EndOfGroup at (4, 1) on + // another subgroup, which now fails the object < next_object check. + metadata.subgroup = 1; + metadata.payload_length = 0; + EXPECT_CALL(listener_, OnTrackPublisherGone); + EXPECT_CALL(*session_, Unsubscribe(kTrackName)); + publisher_.OnObjectFragment(kTrackName, metadata, "", 0); + EXPECT_TRUE(track_deleted_); } TEST_F(MoqtRelayTrackPublisherTest, IncompleteDatagram) {
diff --git a/quiche/quic/moqt/moqt_uni_stream.cc b/quiche/quic/moqt/moqt_uni_stream.cc index 0918a7a..afcd0bc 100644 --- a/quiche/quic/moqt/moqt_uni_stream.cc +++ b/quiche/quic/moqt/moqt_uni_stream.cc
@@ -78,15 +78,7 @@ } 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))); - } + SendNotionalEndOfGroup(type, write_vector); } webtransport::StreamWriteOptions options; options.set_send_fin(!type.IsFetch() && object.fin_after_this); @@ -104,6 +96,30 @@ return true; } +void OutgoingUniStream::SendNotionalEndOfGroup( + MoqtDataStreamType type, + std::vector<quiche::QuicheMemSlice>& write_vector) { + QUICHE_DCHECK(last_object_.has_value()); + if (!last_object_.has_value()) { + return; + } + MoqtObject header; + header.track_alias = track_identifier_; + header.group_id = last_object_->location.group; + header.subgroup_id = last_object_->subgroup; + header.object_id = last_object_->location.object; + header.publisher_priority = last_object_->publisher_priority; + if (quiche::SafeIncrementBy<uint64_t>(header.object_id, 1)) { + header.object_status = MoqtObjectStatus::kEndOfGroup; + header.payload_length = 0; + header.properties.clear(); + quiche::QuicheBuffer serialized_header = + framer_.SerializeObjectHeader(header, type, last_object_); + write_vector.push_back( + quiche::QuicheMemSlice(std::move(serialized_header))); + } +} + OutgoingSubgroupStream::OutgoingSubgroupStream( MoqtFramer framer, webtransport::Stream* absl_nonnull stream, DataStreamIndex index, uint64_t first_object, @@ -217,6 +233,10 @@ << "Empty payload for partial object " << object->metadata.location; return; } + if (!type_.EndOfGroupInStream() && object->fin_after_this && + object->metadata.status == MoqtObjectStatus::kEndOfGroup) { + SendNotionalEndOfGroup(type_, object->payload); + } webtransport::StreamWriteOptions options; options.set_send_fin(object->fin_after_this); absl::Status write_status =
diff --git a/quiche/quic/moqt/moqt_uni_stream.h b/quiche/quic/moqt/moqt_uni_stream.h index 8ac3646..eae17b4 100644 --- a/quiche/quic/moqt/moqt_uni_stream.h +++ b/quiche/quic/moqt/moqt_uni_stream.h
@@ -10,6 +10,7 @@ #include <optional> #include <string> #include <utility> +#include <vector> #include "absl/base/nullability.h" #include "absl/status/status.h" @@ -31,8 +32,8 @@ #include "quiche/quic/moqt/moqt_types.h" #include "quiche/common/platform/api/quiche_export.h" #include "quiche/common/quiche_callbacks.h" +#include "quiche/common/quiche_mem_slice.h" #include "quiche/common/quiche_weak_ptr.h" -#include "quiche/web_transport/stream_helpers.h" #include "quiche/web_transport/web_transport.h" namespace moqt { @@ -80,6 +81,12 @@ // caller should reset the stream if that happens. bool WriteObjectToStream(PublishedObject& object, MoqtDataStreamType type); + // Appends a notional empty EndOfGroup object header to `write_vector`, + // incrementing the object ID from `last_object_`. + void SendNotionalEndOfGroup( + MoqtDataStreamType type, + std::vector<quiche::QuicheMemSlice>& write_vector); + private: webtransport::Stream& stream_; // Always valid because it owns this object. webtransport::StreamPriority priority_;
diff --git a/quiche/quic/moqt/moqt_uni_stream_test.cc b/quiche/quic/moqt/moqt_uni_stream_test.cc index ad4ab8a..90a7f0c 100644 --- a/quiche/quic/moqt/moqt_uni_stream_test.cc +++ b/quiche/quic/moqt/moqt_uni_stream_test.cc
@@ -301,6 +301,57 @@ stream_->OnCanWrite(); } +TEST_F(OutgoingSubgroupStreamTest, SendFragmentedObjectEndOfGroup) { + PublishedObject obj0 = DefaultObject(); + obj0.metadata.payload_length = 15; + obj0.payload.clear(); + obj0.payload.push_back(quiche::QuicheMemSlice::Copy("part1")); + obj0.payload.push_back(quiche::QuicheMemSlice::Copy("part2")); + obj0.fin_after_this = false; + EXPECT_CALL(*track_publisher_, GetCachedObject(0, Optional(0), 0, 0)) + .WillOnce(Return(std::move(obj0))); + EXPECT_CALL(visitor_, InWindow).WillRepeatedly(Return(true)); + EXPECT_CALL(mock_stream_, CanWrite).WillRepeatedly(Return(true)); + EXPECT_CALL(visitor_, object_delivery_timeout()) + .WillRepeatedly(Return(quic::QuicTimeDelta::FromSeconds(1))); + EXPECT_CALL(visitor_, subgroup_delivery_timeout()) + .WillRepeatedly(Return(quic::QuicTimeDelta::FromSeconds(1))); + EXPECT_CALL(visitor_, clock()).WillRepeatedly(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(), 3); + EXPECT_EQ(data[1].AsStringView(), "part1"); + EXPECT_EQ(data[2].AsStringView(), "part2"); + EXPECT_FALSE(options.send_fin()); + return absl::OkStatus(); + }); + EXPECT_CALL(visitor_, OnObjectSent(Location(0, 0))); + stream_->OnCanWrite(); + PublishedObject obj1 = DefaultObject(); + obj1.metadata.payload_length = 15; + obj1.metadata.status = MoqtObjectStatus::kEndOfGroup; + obj1.payload.clear(); + obj1.payload.push_back(quiche::QuicheMemSlice::Copy("part3")); + obj1.fin_after_this = true; + EXPECT_CALL(*track_publisher_, GetCachedObject(0, Optional(0), 0, 10)) + .WillOnce(Return(std::move(obj1))); + EXPECT_CALL(*track_publisher_, GetCachedObject(0, Optional(0), 1, 0)) + .WillOnce(Return(std::nullopt)); + EXPECT_CALL(mock_stream_, Writev) + .WillOnce([&](absl::Span<quiche::QuicheMemSlice> data, + const webtransport::StreamWriteOptions& options) { + EXPECT_EQ(data.size(), 2); + EXPECT_EQ(data[0].AsStringView(), "part3"); + EXPECT_TRUE(options.send_fin()); + return absl::OkStatus(); + }); + EXPECT_CALL(visitor_, OnObjectSent).Times(0); + stream_->OnCanWrite(); +} + TEST_F(OutgoingSubgroupStreamTest, EndOfGroupWithPayload) { PublishedObject obj0 = DefaultObject(); obj0.metadata.status = MoqtObjectStatus::kEndOfGroup;