Publisher-side MoQT FETCH handling. Also: a. Enforce that subscribe_id is monotonically increasing. b. Use the cached object publisher_priority instead of the track-level pub priority. PiperOrigin-RevId: 698869614
diff --git a/quiche/quic/moqt/moqt_failed_fetch.h b/quiche/quic/moqt/moqt_failed_fetch.h index a1e93bf..40e8d67 100644 --- a/quiche/quic/moqt/moqt_failed_fetch.h +++ b/quiche/quic/moqt/moqt_failed_fetch.h
@@ -8,6 +8,7 @@ #include <utility> #include "absl/status/status.h" +#include "quiche/quic/moqt/moqt_messages.h" #include "quiche/quic/moqt/moqt_publisher.h" namespace moqt { @@ -23,6 +24,7 @@ absl::Status GetStatus() override { return status_; } void SetObjectAvailableCallback( ObjectsAvailableCallback /*callback*/) override {} + FullSequence GetLargestId() const override { return FullSequence(); } private: absl::Status status_;
diff --git a/quiche/quic/moqt/moqt_outgoing_queue.h b/quiche/quic/moqt/moqt_outgoing_queue.h index 2bc46ee..83d7773 100644 --- a/quiche/quic/moqt/moqt_outgoing_queue.h +++ b/quiche/quic/moqt/moqt_outgoing_queue.h
@@ -102,6 +102,7 @@ GetNextObjectResult GetNextObject(PublishedObject&) override; absl::Status GetStatus() override { return status_; } + FullSequence GetLargestId() const override { return objects_.back(); } void SetObjectAvailableCallback( ObjectsAvailableCallback /*callback*/) override {
diff --git a/quiche/quic/moqt/moqt_publisher.h b/quiche/quic/moqt/moqt_publisher.h index a0897f4..010db7f 100644 --- a/quiche/quic/moqt/moqt_publisher.h +++ b/quiche/quic/moqt/moqt_publisher.h
@@ -84,8 +84,9 @@ // Returns the error if fetch has completely failed, and OK otherwise. virtual absl::Status GetStatus() = 0; - // TODO: expose the largest sequence and the end of track bit returned in - // the FETCH_OK. + // Returns the highest sequence number that will be delivered by the fetch. + // It is the minimum of the end of the fetch range and the live edge. + virtual FullSequence GetLargestId() const = 0; }; // MoqtTrackPublisher is an application-side API for an MoQT publisher
diff --git a/quiche/quic/moqt/moqt_session.cc b/quiche/quic/moqt/moqt_session.cc index 0f8fad7..ec7a867 100644 --- a/quiche/quic/moqt/moqt_session.cc +++ b/quiche/quic/moqt/moqt_session.cc
@@ -339,6 +339,46 @@ return Subscribe(message, visitor); } +void MoqtSession::PublishedFetch::FetchStreamVisitor::OnCanWrite() { + std::shared_ptr<PublishedFetch> fetch = fetch_.lock(); + if (fetch == nullptr) { + return; + } + PublishedObject object; + while (stream_->CanWrite()) { + MoqtFetchTask::GetNextObjectResult result = + fetch->fetch_task()->GetNextObject(object); + switch (result) { + case MoqtFetchTask::GetNextObjectResult::kSuccess: + // Skip ObjectDoesNotExist in FETCH. + if (object.status == MoqtObjectStatus::kObjectDoesNotExist) { + continue; + } + if (fetch->session_->WriteObjectToStream( + stream_, fetch->fetch_id_, object, + MoqtDataStreamType::kStreamHeaderFetch, !stream_header_written_, + /*fin=*/false)) { + stream_header_written_ = true; + } + break; + case MoqtFetchTask::GetNextObjectResult::kPending: + return; + case MoqtFetchTask::GetNextObjectResult::kEof: + // TODO(martinduke): Either prefetch the next object, or alter the API + // so that we're not sending FIN in a separate frame. + if (!quiche::SendFinOnStream(*stream_).ok()) { + QUIC_DVLOG(1) << "Sending FIN onStream " << stream_->GetStreamId() + << " failed"; + } + return; + case MoqtFetchTask::GetNextObjectResult::kError: + stream_->ResetWithUserCode(static_cast<webtransport::StreamErrorCode>( + fetch->fetch_task()->GetStatus().code())); + return; + } + } +} + bool MoqtSession::SubscribeIsDone(uint64_t subscribe_id, SubscribeDoneCode code, absl::string_view reason_phrase) { auto it = published_subscriptions_.find(subscribe_id); @@ -446,13 +486,36 @@ return new_stream; } +bool MoqtSession::OpenDataStream(std::shared_ptr<PublishedFetch> fetch) { + webtransport::Stream* new_stream = + session_->OpenOutgoingUnidirectionalStream(); + if (new_stream == nullptr) { + QUICHE_BUG(MoqtSession_OpenDataStream_blocked) + << "OpenDataStream called when creation of new streams is blocked."; + return false; + } + fetch->SetStreamId(new_stream->GetStreamId()); + new_stream->SetVisitor( + std::make_unique<PublishedFetch::FetchStreamVisitor>(fetch, new_stream)); + if (new_stream->CanWrite()) { + new_stream->visitor()->OnCanWrite(); + } + return true; +} + void MoqtSession::OnCanCreateNewOutgoingUnidirectionalStream() { while (!subscribes_with_queued_outgoing_data_streams_.empty() && session_->CanOpenNextOutgoingUnidirectionalStream()) { auto next = subscribes_with_queued_outgoing_data_streams_.rbegin(); auto subscription = published_subscriptions_.find(next->subscription_id); if (subscription == published_subscriptions_.end()) { - // Subscription no longer exists; delete the entry. + auto fetch = incoming_fetches_.find(next->subscription_id); + // Create the stream if the fetch still exists. + if (fetch != incoming_fetches_.end() && !OpenDataStream(fetch->second)) { + return; // A QUIC_BUG has fired because this shouldn't happen. + } + // FETCH needs only one stream, and can be deleted from the queue. Or, + // there is no subscribe and no fetch; the entry in the queue is invalid. subscribes_with_queued_outgoing_data_streams_.erase((++next).base()); continue; } @@ -535,6 +598,28 @@ return std::make_pair(track.full_track_name(), track.visitor()); } +bool MoqtSession::ValidateSubscribeId(uint64_t subscribe_id) { + if (peer_role_ == MoqtRole::kPublisher) { + QUIC_DLOG(INFO) << ENDPOINT << "Publisher peer sent SUBSCRIBE"; + Error(MoqtError::kProtocolViolation, "Received SUBSCRIBE from publisher"); + return false; + } + if (subscribe_id > local_max_subscribe_id_) { + QUIC_DLOG(INFO) << ENDPOINT << "Received SUBSCRIBE with too large ID"; + Error(MoqtError::kTooManySubscribes, + "Received SUBSCRIBE with too large ID"); + return false; + } + if (subscribe_id < next_incoming_subscribe_id_) { + QUIC_DLOG(INFO) << ENDPOINT << "Subscribe ID not monotonically increasing"; + Error(MoqtError::kProtocolViolation, + "Subscribe ID not monotonically increasing"); + return false; + } + next_incoming_subscribe_id_ = subscribe_id + 1; + return true; +} + template <class Parser> static void ForwardStreamDataToParser(webtransport::Stream& stream, Parser& parser) { @@ -647,18 +732,19 @@ session_->framer_.SerializeSubscribeError(subscribe_error)); } +void MoqtSession::ControlStream::SendFetchError( + uint64_t subscribe_id, SubscribeErrorCode error_code, + absl::string_view reason_phrase) { + MoqtFetchError fetch_error; + fetch_error.subscribe_id = subscribe_id; + fetch_error.error_code = error_code; + fetch_error.reason_phrase = reason_phrase; + SendOrBufferMessage(session_->framer_.SerializeFetchError(fetch_error)); +} + void MoqtSession::ControlStream::OnSubscribeMessage( const MoqtSubscribe& message) { - if (session_->peer_role_ == MoqtRole::kPublisher) { - QUIC_DLOG(INFO) << ENDPOINT << "Publisher peer sent SUBSCRIBE"; - session_->Error(MoqtError::kProtocolViolation, - "Received SUBSCRIBE from publisher"); - return; - } - if (message.subscribe_id > session_->local_max_subscribe_id_) { - QUIC_DLOG(INFO) << ENDPOINT << "Received SUBSCRIBE with too large ID"; - session_->Error(MoqtError::kTooManySubscribes, - "Received SUBSCRIBE with too large ID"); + if (!session_->ValidateSubscribeId(message.subscribe_id)) { return; } QUIC_DLOG(INFO) << ENDPOINT << "Received a SUBSCRIBE for " @@ -883,6 +969,66 @@ session_->peer_max_subscribe_id_ = message.max_subscribe_id; } +void MoqtSession::ControlStream::OnFetchMessage(const MoqtFetch& message) { + if (!session_->ValidateSubscribeId(message.subscribe_id)) { + return; + } + QUIC_DLOG(INFO) << ENDPOINT << "Received a FETCH for " + << message.full_track_name; + + const FullTrackName& track_name = message.full_track_name; + absl::StatusOr<std::shared_ptr<MoqtTrackPublisher>> track_publisher = + session_->publisher_->GetTrack(track_name); + if (!track_publisher.ok()) { + QUIC_DLOG(INFO) << ENDPOINT << "FETCH for " << track_name + << " rejected by the application: " + << track_publisher.status(); + SendFetchError(message.subscribe_id, SubscribeErrorCode::kTrackDoesNotExist, + track_publisher.status().message()); + return; + } + std::unique_ptr<MoqtFetchTask> fetch = + (*track_publisher) + ->Fetch(message.start_object, message.end_group, message.end_object, + message.group_order.value_or( + (*track_publisher)->GetDeliveryOrder())); + if (!fetch->GetStatus().ok()) { + QUIC_DLOG(INFO) << ENDPOINT << "FETCH for " << track_name + << " could not initialize the task"; + SendFetchError(message.subscribe_id, SubscribeErrorCode::kInvalidRange, + fetch->GetStatus().message()); + return; + } + auto published_fetch = std::make_unique<PublishedFetch>( + message.subscribe_id, session_, std::move(fetch)); + auto result = session_->incoming_fetches_.emplace(message.subscribe_id, + std::move(published_fetch)); + if (!result.second) { // Emplace failed. + QUIC_DLOG(INFO) << ENDPOINT << "FETCH for " << track_name + << " could not be added to the session"; + SendFetchError(message.subscribe_id, SubscribeErrorCode::kInternalError, + "Could not initialize FETCH state"); + return; + } + MoqtFetchOk fetch_ok; + fetch_ok.subscribe_id = message.subscribe_id; + fetch_ok.group_order = + message.group_order.value_or((*track_publisher)->GetDeliveryOrder()); + fetch_ok.largest_id = result.first->second->fetch_task()->GetLargestId(); + SendOrBufferMessage(session_->framer_.SerializeFetchOk(fetch_ok)); + if (!session_->session()->CanOpenNextOutgoingUnidirectionalStream() || + !session_->OpenDataStream(result.first->second)) { + // Put the FETCH in the queue for a new stream. + session_->UpdateQueuedSendOrder( + message.subscribe_id, std::nullopt, + SendOrderForStream(message.subscriber_priority, + (*track_publisher)->GetPublisherPriority(), + /*group_id=*/0, + message.group_order.value_or( + (*track_publisher)->GetDeliveryOrder()))); + } +} + void MoqtSession::ControlStream::OnParsingError(MoqtError error_code, absl::string_view reason) { session_->Error(error_code, absl::StrCat("Parse error: ", reason)); @@ -1290,7 +1436,38 @@ << "Writing FIN failed despite CanWrite() being true."; return; } - SendNextObject(subscription, *std::move(object)); + QUICHE_DCHECK(next_object_ <= object->sequence); + MoqtTrackPublisher& publisher = subscription.publisher(); + QUICHE_DCHECK(DoesTrackStatusImplyHavingData(*publisher.GetTrackStatus())); + MoqtForwardingPreference forwarding_preference = + publisher.GetForwardingPreference(); + UpdateSendOrder(subscription); + switch (forwarding_preference) { + case MoqtForwardingPreference::kTrack: + if (object->status == MoqtObjectStatus::kEndOfGroup || + object->status == MoqtObjectStatus::kGroupDoesNotExist) { + ++next_object_.group; + next_object_.object = 0; + } else { + next_object_.object = object->sequence.object + 1; + } + break; + + case MoqtForwardingPreference::kSubgroup: + next_object_.object = object->sequence.object + 1; + break; + + case MoqtForwardingPreference::kDatagram: + QUICHE_NOTREACHED(); + break; + } + if (session_->WriteObjectToStream( + stream_, subscription.track_alias(), *object, + GetMessageTypeForForwardingPreference(forwarding_preference), + !stream_header_written_, object->fin_after_this)) { + stream_header_written_ = true; + subscription.OnObjectSent(object->sequence); + } } } @@ -1305,78 +1482,41 @@ << "Writing pure FIN failed."; } -void MoqtSession::OutgoingDataStream::SendNextObject( - PublishedSubscription& subscription, PublishedObject object) { - QUICHE_DCHECK(next_object_ <= object.sequence); - QUICHE_DCHECK(stream_->CanWrite()); - - MoqtTrackPublisher& publisher = subscription.publisher(); - QUICHE_DCHECK(DoesTrackStatusImplyHavingData(*publisher.GetTrackStatus())); - MoqtForwardingPreference forwarding_preference = - publisher.GetForwardingPreference(); - - UpdateSendOrder(subscription); - +bool MoqtSession::WriteObjectToStream(webtransport::Stream* stream, uint64_t id, + const PublishedObject& object, + MoqtDataStreamType type, + bool is_first_on_stream, bool fin) { + QUICHE_DCHECK(stream->CanWrite()); MoqtObject header; - header.track_alias = subscription.track_alias(); + header.track_alias = id; header.group_id = object.sequence.group; + header.subgroup_id = object.sequence.subgroup; header.object_id = object.sequence.object; - header.publisher_priority = publisher.GetPublisherPriority(); + header.publisher_priority = object.publisher_priority; header.object_status = object.status; - header.forwarding_preference = forwarding_preference; - // TODO(martinduke): send values other than 0. - header.subgroup_id = - (forwarding_preference == MoqtForwardingPreference::kSubgroup) - ? 0 - : std::optional<uint64_t>(); header.payload_length = object.payload.length(); quiche::QuicheBuffer serialized_header = - session_->framer_.SerializeObjectHeader( - header, GetMessageTypeForForwardingPreference(forwarding_preference), - !stream_header_written_); - switch (forwarding_preference) { - case MoqtForwardingPreference::kTrack: - if (object.status == MoqtObjectStatus::kEndOfGroup || - object.status == MoqtObjectStatus::kGroupDoesNotExist) { - ++next_object_.group; - next_object_.object = 0; - } else { - next_object_.object = header.object_id + 1; - } - break; - - case MoqtForwardingPreference::kSubgroup: - next_object_.object = header.object_id + 1; - break; - - case MoqtForwardingPreference::kDatagram: - QUICHE_NOTREACHED(); - break; - } - + framer_.SerializeObjectHeader(header, type, is_first_on_stream); // TODO(vasilvv): add a version of WebTransport write API that accepts // memslices so that we can avoid a copy here. std::array<absl::string_view, 2> write_vector = { serialized_header.AsStringView(), object.payload.AsStringView()}; quiche::StreamWriteOptions options; - options.set_send_fin(object.fin_after_this); - absl::Status write_status = stream_->Writev(write_vector, options); + options.set_send_fin(fin); + absl::Status write_status = stream->Writev(write_vector, options); if (!write_status.ok()) { - QUICHE_BUG(MoqtSession_SendNextObject_write_failed) + QUICHE_BUG(MoqtSession_WriteObjectToStream_write_failed) << "Writing into MoQT stream failed despite CanWrite() being true " "before; status: " << write_status; - session_->Error(MoqtError::kInternalError, "Data stream write error"); - return; + Error(MoqtError::kInternalError, "Data stream write error"); + return false; } - QUIC_DVLOG(1) << "Stream " << stream_->GetStreamId() << " successfully wrote " - << object.sequence << ", fin = " << object.fin_after_this - << ", next: " << next_object_; - - stream_header_written_ = true; - subscription.OnObjectSent(object.sequence); + QUIC_DVLOG(1) << "Stream " << stream->GetStreamId() << " successfully wrote " + << object.sequence << ", fin = " << fin; + return true; } void MoqtSession::PublishedSubscription::SendDatagram(FullSequence sequence) {
diff --git a/quiche/quic/moqt/moqt_session.h b/quiche/quic/moqt/moqt_session.h index 04972b1..ffae8d7 100644 --- a/quiche/quic/moqt/moqt_session.h +++ b/quiche/quic/moqt/moqt_session.h
@@ -172,6 +172,8 @@ private: friend class test::MoqtSessionPeer; + struct Empty {}; + class QUICHE_EXPORT ControlStream : public webtransport::StreamVisitor, public MoqtControlParserVisitor { public: @@ -213,7 +215,7 @@ void OnUnsubscribeAnnouncesMessage( const MoqtUnsubscribeAnnounces& message) override {} void OnMaxSubscribeIdMessage(const MoqtMaxSubscribeId& message) override; - void OnFetchMessage(const MoqtFetch& message) override {} + void OnFetchMessage(const MoqtFetch& message) override; void OnFetchCancelMessage(const MoqtFetchCancel& message) override {} void OnFetchOkMessage(const MoqtFetchOk& message) override {} void OnFetchErrorMessage(const MoqtFetchError& message) override {} @@ -244,6 +246,8 @@ SubscribeErrorCode error_code, absl::string_view reason_phrase, uint64_t track_alias); + void SendFetchError(uint64_t subscribe_id, SubscribeErrorCode error_code, + absl::string_view reason_phrase); MoqtSession* session_; webtransport::Stream* stream_; @@ -418,11 +422,6 @@ // the stream and returns nullptr. PublishedSubscription* GetSubscriptionIfValid(); - // Actually sends an object on the stream; the object MUST have object ID - // of at least `next_object_`. - void SendNextObject(PublishedSubscription& subscription, - PublishedObject object); - MoqtSession* session_; webtransport::Stream* stream_; uint64_t subscription_id_; @@ -436,6 +435,54 @@ std::weak_ptr<void> session_liveness_; }; + class QUICHE_EXPORT PublishedFetch { + public: + PublishedFetch(uint64_t fetch_id, MoqtSession* session, + std::unique_ptr<MoqtFetchTask> fetch) + : session_(session), fetch_(std::move(fetch)), fetch_id_(fetch_id) {} + + class FetchStreamVisitor : public webtransport::StreamVisitor { + public: + FetchStreamVisitor(std::shared_ptr<PublishedFetch> fetch, + webtransport::Stream* stream) + : fetch_(fetch), stream_(stream) { + fetch->fetch_task()->SetObjectAvailableCallback( + [this]() { this->OnCanWrite(); }); + } + ~FetchStreamVisitor() { + std::shared_ptr<PublishedFetch> fetch = fetch_.lock(); + if (fetch != nullptr) { + fetch->session()->incoming_fetches_.erase(fetch->fetch_id_); + } + } + // webtransport::StreamVisitor implementation. + void OnCanRead() override {} // Write-only stream. + void OnCanWrite() override; + void OnResetStreamReceived(webtransport::StreamErrorCode error) override { + } // Write-only stream + void OnStopSendingReceived(webtransport::StreamErrorCode error) override { + } + void OnWriteSideInDataRecvdState() override {} + + private: + std::weak_ptr<PublishedFetch> fetch_; + webtransport::Stream* stream_; + bool stream_header_written_ = false; + }; + + MoqtFetchTask* fetch_task() { return fetch_.get(); } + MoqtSession* session() { return session_; } + uint64_t fetch_id() const { return fetch_id_; } + void SetStreamId(webtransport::StreamId id) { stream_id_ = id; } + + private: + MoqtSession* session_; + std::unique_ptr<MoqtFetchTask> fetch_; + uint64_t fetch_id_; + // Store the stream ID in case a FETCH_CANCEL requires a reset. + std::optional<webtransport::StreamId> stream_id_; + }; + // Private members of MoqtSession. // Returns true if SUBSCRIBE_DONE was sent. @@ -458,12 +505,26 @@ // blocked. webtransport::Stream* OpenDataStream(PublishedSubscription& subscription, FullSequence first_object); + // Returns false if creation failed. + [[nodiscard]] bool OpenDataStream(std::shared_ptr<PublishedFetch> fetch); // Get FullTrackName and visitor for a subscribe_id and track_alias. Returns // an empty FullTrackName tuple and nullptr if not present. std::pair<FullTrackName, RemoteTrack::Visitor*> TrackPropertiesFromAlias( const MoqtObject& message); + // Checks that a subscribe ID from a SUBSCRIBE or FETCH is valid, and throws + // a session error if is not. + bool ValidateSubscribeId(uint64_t subscribe_id); + + // Actually sends an object on |stream| with track alias or fetch ID |id| + // and metadata in |object|. Not for use with datagrams. Returns |true| if + // the write was successful. + bool WriteObjectToStream(webtransport::Stream* stream, uint64_t id, + const PublishedObject& object, + MoqtDataStreamType type, bool is_first_on_stream, + bool fin); + // Sends an OBJECT_ACK message for a specific subscribe ID. void SendObjectAck(uint64_t subscribe_id, uint64_t group_id, uint64_t object_id, @@ -515,6 +576,12 @@ absl::flat_hash_set<uint64_t> used_track_aliases_; uint64_t next_local_track_alias_ = 0; + // Incoming FETCHes, indexed by fetch ID. There will be other pointers to + // PublishedFetch, so storing a shared_ptr in the map provides pointer + // stability for the value. + absl::flat_hash_map<uint64_t, std::shared_ptr<PublishedFetch>> + incoming_fetches_; + // Indexed by subscribe_id. struct ActiveSubscribe { MoqtSubscribe message; @@ -547,10 +614,12 @@ uint64_t peer_max_subscribe_id_ = 0; // The maximum subscribe ID sent to the peer. uint64_t local_max_subscribe_id_ = 0; + // The minimum subscribe ID the peer can use that is monotonically increasing. + uint64_t next_incoming_subscribe_id_ = 0; // Must be last. Token used to make sure that the streams do not call into // the session when the session has already been destroyed. - struct Empty {}; + std::shared_ptr<Empty> liveness_token_; };
diff --git a/quiche/quic/moqt/moqt_session_test.cc b/quiche/quic/moqt/moqt_session_test.cc index bc31674..5e40775 100644 --- a/quiche/quic/moqt/moqt_session_test.cc +++ b/quiche/quic/moqt/moqt_session_test.cc
@@ -42,6 +42,7 @@ using ::quic::test::MemSliceFromString; using ::testing::_; using ::testing::AnyNumber; +using ::testing::Invoke; using ::testing::Return; using ::testing::StrictMock; @@ -75,6 +76,20 @@ } // namespace +class MockFetchTask : public MoqtFetchTask { + public: + MOCK_METHOD(MoqtFetchTask::GetNextObjectResult, GetNextObject, + (PublishedObject & output), (override)); + MOCK_METHOD(absl::Status, GetStatus, (), (override)); + MOCK_METHOD(FullSequence, GetLargestId, (), (const, override)); + + void SetObjectAvailableCallback(ObjectsAvailableCallback callback) override { + callback_ = std::move(callback); + } + + ObjectsAvailableCallback callback_; +}; + class MoqtSessionPeer { public: static std::unique_ptr<MoqtControlParserVisitor> CreateControlStream( @@ -169,6 +184,35 @@ static void set_peer_max_subscribe_id(MoqtSession* session, uint64_t id) { session->peer_max_subscribe_id_ = id; } + + static MockFetchTask* AddFetch(MoqtSession* session, uint64_t fetch_id) { + auto fetch_task = std::make_unique<MockFetchTask>(); + MockFetchTask* return_ptr = fetch_task.get(); + auto published_fetch = std::make_unique<MoqtSession::PublishedFetch>( + fetch_id, session, std::move(fetch_task)); + session->incoming_fetches_.emplace(fetch_id, std::move(published_fetch)); + // Add the fetch to the pending stream queue. + session->UpdateQueuedSendOrder(fetch_id, std::nullopt, 0); + return return_ptr; + } + + static MoqtSession::PublishedFetch* GetFetch(MoqtSession* session, + uint64_t fetch_id) { + auto it = session->incoming_fetches_.find(fetch_id); + if (it == session->incoming_fetches_.end()) { + return nullptr; + } + return it->second.get(); + } + + static void ValidateSubscribeId(MoqtSession* session, uint64_t id) { + session->ValidateSubscribeId(id); + } + + static FullSequence LargestSentForSubscription(MoqtSession* session, + uint64_t subscribe_id) { + return *session->published_subscriptions_[subscribe_id]->largest_sent(); + } }; class MoqtSessionTest : public quic::test::QuicTest { @@ -356,6 +400,7 @@ EXPECT_EQ(*ExtractMessageType(data[0]), MoqtMessageType::kSubscribeOk); return absl::OkStatus(); }); + request.subscribe_id = 2; stream_input->OnSubscribeMessage(request); EXPECT_TRUE(correct_message); } @@ -626,6 +671,47 @@ stream_input->OnSubscribeMessage(request); } +TEST_F(MoqtSessionTest, SubscribeIdNotIncreasing) { + // Peer subscribes to (0, 0) + MoqtSubscribe request = { + /*subscribe_id=*/1, + /*track_alias=*/2, + /*full_track_name=*/FullTrackName({"foo", "bar"}), + /*subscriber_priority=*/0x80, + /*group_order=*/std::nullopt, + /*start_group=*/0, + /*start_object=*/0, + /*end_group=*/std::nullopt, + /*end_object=*/std::nullopt, + /*parameters=*/MoqtSubscribeParameters(), + }; + webtransport::test::MockStream mock_stream; + std::unique_ptr<MoqtControlParserVisitor> stream_input = + MoqtSessionPeer::CreateControlStream(&session_, &mock_stream); + // Request for track returns SUBSCRIBE_ERROR. + bool correct_message = false; + EXPECT_CALL(mock_stream, Writev(_, _)) + .WillOnce([&](absl::Span<const absl::string_view> data, + const quiche::StreamWriteOptions& options) { + correct_message = true; + EXPECT_EQ(*ExtractMessageType(data[0]), + MoqtMessageType::kSubscribeError); + return absl::OkStatus(); + }); + stream_input->OnSubscribeMessage(request); + EXPECT_TRUE(correct_message); + + // Second request is a protocol violation. + request.subscribe_id = 0; + request.track_alias = 3; + request.full_track_name = FullTrackName({"dead", "beef"}); + EXPECT_CALL(mock_session_, + CloseSession(static_cast<uint64_t>(MoqtError::kProtocolViolation), + "Subscribe ID not monotonically increasing")) + .Times(1); + stream_input->OnSubscribeMessage(request); +} + TEST_F(MoqtSessionTest, TooManySubscribes) { MoqtSessionPeer::set_next_subscribe_id(&session_, kDefaultInitialMaxSubscribeId); @@ -1148,7 +1234,7 @@ control_stream->OnSubscribeOkMessage(ok); } -TEST_F(MoqtSessionTest, CreateIncomingDataStreamAndSend) { +TEST_F(MoqtSessionTest, CreateOutgoingDataStreamAndSend) { FullTrackName ftn("foo", "bar"); auto track = SetupPublisher(ftn, MoqtForwardingPreference::kSubgroup, FullSequence(4, 2)); @@ -1177,7 +1263,7 @@ // Verify first six message fields are sent correctly bool correct_message = false; - const std::string kExpectedMessage = {0x04, 0x02, 0x05, 0x00, 0x00}; + const std::string kExpectedMessage = {0x04, 0x02, 0x05, 0x00, 0x7f}; EXPECT_CALL(mock_stream, Writev(_, _)) .WillOnce([&](absl::Span<const absl::string_view> data, const quiche::StreamWriteOptions& options) { @@ -1195,6 +1281,8 @@ subscription->OnNewObjectAvailable(FullSequence(5, 0)); EXPECT_TRUE(correct_message); EXPECT_FALSE(fin); + EXPECT_EQ(MoqtSessionPeer::LargestSentForSubscription(&session_, 0), + FullSequence(5, 0)); } TEST_F(MoqtSessionTest, FinDataStreamFromCache) { @@ -1224,9 +1312,9 @@ EXPECT_CALL(mock_session_, GetStreamById(kOutgoingUniStreamId)) .WillRepeatedly(Return(&mock_stream)); - // Verify first six message fields are sent correctly + // Verify first five message fields are sent correctly bool correct_message = false; - const std::string kExpectedMessage = {0x04, 0x02, 0x05, 0x00, 0x00}; + const std::string kExpectedMessage = {0x04, 0x02, 0x05, 0x00, 0x7f}; EXPECT_CALL(mock_stream, Writev(_, _)) .WillOnce([&](absl::Span<const absl::string_view> data, const quiche::StreamWriteOptions& options) { @@ -1275,7 +1363,7 @@ // Verify first six message fields are sent correctly bool correct_message = false; - const std::string kExpectedMessage = {0x04, 0x02, 0x05, 0x00, 0x00}; + const std::string kExpectedMessage = {0x04, 0x02, 0x05, 0x00, 0x7f}; EXPECT_CALL(mock_stream, Writev(_, _)) .WillOnce([&](absl::Span<const absl::string_view> data, const quiche::StreamWriteOptions& options) { @@ -1327,7 +1415,7 @@ // Verify first six message fields are sent correctly bool correct_message = false; - const std::string kExpectedMessage = {0x04, 0x02, 0x05, 0x00, 0x00}; + const std::string kExpectedMessage = {0x04, 0x02, 0x05, 0x00, 0x7f}; EXPECT_CALL(mock_stream, Writev(_, _)) .WillOnce([&](absl::Span<const absl::string_view> data, const quiche::StreamWriteOptions& options) { @@ -1994,6 +2082,316 @@ session_.OnCanCreateNewOutgoingUnidirectionalStream(); } +TEST_F(MoqtSessionTest, FetchReturnsOk) { + webtransport::test::MockStream control_stream; + std::unique_ptr<MoqtControlParserVisitor> stream_input = + MoqtSessionPeer::CreateControlStream(&session_, &control_stream); + FullTrackName ftn("foo", "bar"); + MoqtFetch request = { + /*subscribe_id=*/0, + /*full_track_name=*/ftn, + /*subscriber_priority=*/0x80, + /*group_order=*/std::nullopt, + /*start=*/FullSequence(0, 0), + /*end_group=*/1, + /*end_object=*/std::nullopt, + /*parameters=*/MoqtSubscribeParameters(), + }; + bool correct_message = false; + auto track = std::make_shared<MockTrackPublisher>(ftn); + publisher_.Add(track); + + auto fetch_task_ptr = std::make_unique<MockFetchTask>(); + MockFetchTask* fetch_task = fetch_task_ptr.get(); + EXPECT_CALL(*track, Fetch(_, _, _, _)) + .WillOnce(Return(std::move(fetch_task_ptr))); + // Compose and send the FETCH_OK. + EXPECT_CALL(*track, GetDeliveryOrder()) + .WillRepeatedly(Return(MoqtDeliveryOrder::kAscending)); + EXPECT_CALL(*fetch_task, GetLargestId()).WillOnce(Return(FullSequence(0, 0))); + EXPECT_CALL(control_stream, Writev(_, _)) + .WillOnce([&](absl::Span<const absl::string_view> data, + const quiche::StreamWriteOptions& options) { + correct_message = true; + EXPECT_EQ(*ExtractMessageType(data[0]), MoqtMessageType::kFetchOk); + return absl::OkStatus(); + }); + // Stream can't open yet. + EXPECT_CALL(mock_session_, CanOpenNextOutgoingUnidirectionalStream) + .WillOnce(Return(false)); + stream_input->OnFetchMessage(request); + EXPECT_TRUE(correct_message); +} + +TEST_F(MoqtSessionTest, FetchReturnsOkImmediateOpen) { + webtransport::test::MockStream control_stream; + std::unique_ptr<MoqtControlParserVisitor> stream_input = + MoqtSessionPeer::CreateControlStream(&session_, &control_stream); + FullTrackName ftn("foo", "bar"); + MoqtFetch request = { + /*subscribe_id=*/0, + /*full_track_name=*/ftn, + /*subscriber_priority=*/0x80, + /*group_order=*/std::nullopt, + /*start=*/FullSequence(0, 0), + /*end_group=*/1, + /*end_object=*/std::nullopt, + /*parameters=*/MoqtSubscribeParameters(), + }; + bool correct_message = false; + auto track = std::make_shared<MockTrackPublisher>(ftn); + publisher_.Add(track); + + auto fetch_task_ptr = std::make_unique<MockFetchTask>(); + MockFetchTask* fetch_task = fetch_task_ptr.get(); + EXPECT_CALL(*track, Fetch(_, _, _, _)) + .WillOnce(Return(std::move(fetch_task_ptr))); + // Compose and send the FETCH_OK. + EXPECT_CALL(*track, GetDeliveryOrder()) + .WillRepeatedly(Return(MoqtDeliveryOrder::kAscending)); + EXPECT_CALL(*fetch_task, GetLargestId()).WillOnce(Return(FullSequence(0, 0))); + EXPECT_CALL(control_stream, Writev(_, _)) + .WillOnce([&](absl::Span<const absl::string_view> data, + const quiche::StreamWriteOptions& options) { + correct_message = true; + EXPECT_EQ(*ExtractMessageType(data[0]), MoqtMessageType::kFetchOk); + return absl::OkStatus(); + }); + // Open stream immediately. + EXPECT_CALL(mock_session_, CanOpenNextOutgoingUnidirectionalStream) + .WillOnce(Return(true)); + webtransport::test::MockStream data_stream; + EXPECT_CALL(mock_session_, OpenOutgoingUnidirectionalStream()) + .WillOnce(Return(&data_stream)); + std::unique_ptr<webtransport::StreamVisitor> stream_visitor; + EXPECT_CALL(data_stream, SetVisitor(_)) + .WillOnce( + Invoke([&](std::unique_ptr<webtransport::StreamVisitor> visitor) { + stream_visitor = std::move(visitor); + })); + EXPECT_CALL(data_stream, CanWrite()).WillRepeatedly(Return(true)); + EXPECT_CALL(data_stream, visitor()).WillOnce(Invoke([&]() { + return stream_visitor.get(); + })); + EXPECT_CALL(*fetch_task, GetNextObject(_)) + .WillOnce(Return(MoqtFetchTask::GetNextObjectResult::kPending)); + stream_input->OnFetchMessage(request); + EXPECT_TRUE(correct_message); + + // Signal the stream that pending object is now available. + correct_message = false; + EXPECT_CALL(data_stream, CanWrite()).WillRepeatedly(Return(true)); + EXPECT_CALL(*fetch_task, GetNextObject(_)) + .WillOnce(Invoke([](PublishedObject& output) { + output.sequence = FullSequence(0, 0, 0); + output.status = MoqtObjectStatus::kNormal; + output.publisher_priority = 128; + output.payload = MemSliceFromString("foo"); + output.fin_after_this = true; + return MoqtFetchTask::GetNextObjectResult::kSuccess; + })) + .WillOnce(Invoke([](PublishedObject& /*output*/) { + return MoqtFetchTask::GetNextObjectResult::kPending; + })); + EXPECT_CALL(data_stream, Writev(_, _)) + .WillOnce([&](absl::Span<const absl::string_view> data, + const quiche::StreamWriteOptions& options) { + correct_message = true; + quic::QuicDataReader reader(data[0]); + uint64_t type; + EXPECT_TRUE(reader.ReadVarInt62(&type)); + EXPECT_EQ(type, static_cast<uint64_t>( + MoqtDataStreamType::kStreamHeaderFetch)); + return absl::OkStatus(); + }); + fetch_task->callback_(); + EXPECT_TRUE(correct_message); +} + +TEST_F(MoqtSessionTest, InvalidFetch) { + // Update the state so that it expects ID > 0 next time. + MoqtSessionPeer::ValidateSubscribeId(&session_, 0); + webtransport::test::MockStream control_stream; + std::unique_ptr<MoqtControlParserVisitor> stream_input = + MoqtSessionPeer::CreateControlStream(&session_, &control_stream); + FullTrackName ftn("foo", "bar"); + MoqtFetch request = { + /*subscribe_id=*/0, // Subscribe ID is too low. + /*full_track_name=*/ftn, + /*subscriber_priority=*/0x80, + /*group_order=*/std::nullopt, + /*start=*/FullSequence(0, 0), + /*end_group=*/1, + /*end_object=*/std::nullopt, + /*parameters=*/MoqtSubscribeParameters(), + }; + EXPECT_CALL(mock_session_, + CloseSession(static_cast<uint64_t>(MoqtError::kProtocolViolation), + "Subscribe ID not monotonically increasing")) + .Times(1); + stream_input->OnFetchMessage(request); +} + +TEST_F(MoqtSessionTest, FetchFails) { + webtransport::test::MockStream control_stream; + std::unique_ptr<MoqtControlParserVisitor> stream_input = + MoqtSessionPeer::CreateControlStream(&session_, &control_stream); + FullTrackName ftn("foo", "bar"); + MoqtFetch request = { + /*subscribe_id=*/0, + /*full_track_name=*/ftn, + /*subscriber_priority=*/0x80, + /*group_order=*/std::nullopt, + /*start=*/FullSequence(0, 0), + /*end_group=*/1, + /*end_object=*/std::nullopt, + /*parameters=*/MoqtSubscribeParameters(), + }; + bool correct_message = false; + auto track = std::make_shared<MockTrackPublisher>(ftn); + publisher_.Add(track); + + auto fetch_task_ptr = std::make_unique<MockFetchTask>(); + MockFetchTask* fetch_task = fetch_task_ptr.get(); + EXPECT_CALL(*track, Fetch(_, _, _, _)) + .WillOnce(Return(std::move(fetch_task_ptr))); + EXPECT_CALL(*fetch_task, GetStatus()) + .WillRepeatedly(Return(absl::Status(absl::StatusCode::kInternal, "foo"))); + EXPECT_CALL(control_stream, Writev(_, _)) + .WillOnce([&](absl::Span<const absl::string_view> data, + const quiche::StreamWriteOptions& options) { + correct_message = true; + EXPECT_EQ(*ExtractMessageType(data[0]), MoqtMessageType::kFetchError); + return absl::OkStatus(); + }); + stream_input->OnFetchMessage(request); + EXPECT_TRUE(correct_message); +} + +TEST_F(MoqtSessionTest, FetchDelivery) { + constexpr uint64_t kFetchId = 0; + MockFetchTask* fetch = MoqtSessionPeer::AddFetch(&session_, kFetchId); + // Stream creation started out blocked. Allow its creation, but data is + // blocked. + webtransport::test::MockStream data_stream; + EXPECT_CALL(mock_session_, CanOpenNextOutgoingUnidirectionalStream()) + .WillRepeatedly(Return(true)); + EXPECT_CALL(mock_session_, OpenOutgoingUnidirectionalStream()) + .WillOnce(Return(&data_stream)); + std::unique_ptr<webtransport::StreamVisitor> stream_visitor; + EXPECT_CALL(data_stream, GetStreamId()).WillOnce(Return(4)); + EXPECT_CALL(data_stream, SetVisitor(_)) + .WillOnce( + Invoke([&](std::unique_ptr<webtransport::StreamVisitor> visitor) { + stream_visitor = std::move(visitor); + })); + EXPECT_CALL(data_stream, CanWrite()).WillOnce(Return(false)); + session_.OnCanCreateNewOutgoingUnidirectionalStream(); + // Unblock the stream. Provide one object and an EOF. + EXPECT_CALL(data_stream, CanWrite()).WillRepeatedly(Return(true)); + EXPECT_CALL(*fetch, GetNextObject(_)) + .WillOnce(Invoke([](PublishedObject& output) { + output.sequence = FullSequence(0, 0, 0); + output.status = MoqtObjectStatus::kNormal; + output.publisher_priority = 128; + output.payload = MemSliceFromString("foo"); + output.fin_after_this = true; + return MoqtFetchTask::GetNextObjectResult::kSuccess; + })) + .WillOnce(Invoke([](PublishedObject& /*output*/) { + return MoqtFetchTask::GetNextObjectResult::kEof; + })); + + int objects_received = 0; + EXPECT_CALL(data_stream, Writev(_, _)) + .WillOnce(Invoke([&](absl::Span<const absl::string_view> data, + const quiche::StreamWriteOptions& options) { + ++objects_received; + quic::QuicDataReader reader(data[0]); + uint64_t type; + EXPECT_TRUE(reader.ReadVarInt62(&type)); + EXPECT_EQ(type, static_cast<uint64_t>( + MoqtDataStreamType::kStreamHeaderFetch)); + EXPECT_FALSE(options.send_fin()); // fin_after_this is ignored. + return absl::OkStatus(); + })) + .WillOnce(Invoke([&](absl::Span<const absl::string_view> data, + const quiche::StreamWriteOptions& options) { + ++objects_received; + EXPECT_TRUE(data.empty()); + EXPECT_TRUE(options.send_fin()); + return absl::OkStatus(); + })); + stream_visitor->OnCanWrite(); + EXPECT_EQ(objects_received, 2); +} + +TEST_F(MoqtSessionTest, FetchNonNormalObjects) { + constexpr uint64_t kFetchId = 0; + MockFetchTask* fetch = MoqtSessionPeer::AddFetch(&session_, kFetchId); + // Stream creation started out blocked. Allow its creation, but data is + // blocked. + webtransport::test::MockStream data_stream; + EXPECT_CALL(mock_session_, CanOpenNextOutgoingUnidirectionalStream()) + .WillRepeatedly(Return(true)); + EXPECT_CALL(mock_session_, OpenOutgoingUnidirectionalStream()) + .WillOnce(Return(&data_stream)); + std::unique_ptr<webtransport::StreamVisitor> stream_visitor; + EXPECT_CALL(data_stream, SetVisitor(_)) + .WillOnce( + Invoke([&](std::unique_ptr<webtransport::StreamVisitor> visitor) { + stream_visitor = std::move(visitor); + })); + EXPECT_CALL(data_stream, CanWrite()).WillOnce(Return(false)); + session_.OnCanCreateNewOutgoingUnidirectionalStream(); + // Unblock the stream. Provide one object and an EOF. + EXPECT_CALL(data_stream, CanWrite()).WillRepeatedly(Return(true)); + EXPECT_CALL(*fetch, GetNextObject(_)) + .WillOnce(Invoke([](PublishedObject& output) { + // DoesNotExist will be skipped. + output.sequence = FullSequence(0, 0, 0); + output.status = MoqtObjectStatus::kObjectDoesNotExist; + output.publisher_priority = 128; + output.payload = MemSliceFromString(""); + output.fin_after_this = true; + return MoqtFetchTask::GetNextObjectResult::kSuccess; + })) + .WillOnce(Invoke([](PublishedObject& output) { + output.sequence = FullSequence(0, 0, 1); + output.status = MoqtObjectStatus::kEndOfGroup; + output.publisher_priority = 128; + output.payload = MemSliceFromString(""); + output.fin_after_this = true; + return MoqtFetchTask::GetNextObjectResult::kSuccess; + })) + .WillOnce(Invoke([](PublishedObject& /*output*/) { + return MoqtFetchTask::GetNextObjectResult::kEof; + })); + + int objects_received = 0; + EXPECT_CALL(data_stream, Writev(_, _)) + .WillOnce(Invoke([&](absl::Span<const absl::string_view> data, + const quiche::StreamWriteOptions& options) { + ++objects_received; + quic::QuicDataReader reader(data[0]); + uint64_t type; + EXPECT_TRUE(reader.ReadVarInt62(&type)); + EXPECT_EQ(type, static_cast<uint64_t>( + MoqtDataStreamType::kStreamHeaderFetch)); + EXPECT_FALSE(options.send_fin()); + return absl::OkStatus(); + })) + .WillOnce(Invoke([&](absl::Span<const absl::string_view> data, + const quiche::StreamWriteOptions& options) { + ++objects_received; + EXPECT_TRUE(data.empty()); + EXPECT_TRUE(options.send_fin()); + return absl::OkStatus(); + })); + stream_visitor->OnCanWrite(); + EXPECT_EQ(objects_received, 2); +} + // TODO: re-enable this test once this behavior is re-implemented. #if 0 TEST_F(MoqtSessionTest, SubscribeUpdateClosesSubscription) {