Eliminate Track Alias negotiation in MoQT, per draft-12. PiperOrigin-RevId: 788590992
diff --git a/quiche/quic/moqt/moqt_framer.cc b/quiche/quic/moqt/moqt_framer.cc index 3505cf4..33ff375 100644 --- a/quiche/quic/moqt/moqt_framer.cc +++ b/quiche/quic/moqt/moqt_framer.cc
@@ -406,50 +406,40 @@ << "Serializing invalid MoQT parameters"; return quiche::QuicheBuffer(); } + std::optional<uint64_t> start_group, start_object, end_group; switch (message.filter_type) { case MoqtFilterType::kNextGroupStart: case MoqtFilterType::kLatestObject: - return SerializeControlMessage( - MoqtMessageType::kSubscribe, WireVarInt62(message.request_id), - WireVarInt62(message.track_alias), - WireFullTrackName(message.full_track_name), - WireUint8(message.subscriber_priority), - WireDeliveryOrder(message.group_order), WireBoolean(message.forward), - WireVarInt62(message.filter_type), WireKeyValuePairList(parameters)); - case MoqtFilterType::kAbsoluteStart: - if (!message.start.has_value()) { - return quiche::QuicheBuffer(); - }; - return SerializeControlMessage( - MoqtMessageType::kSubscribe, WireVarInt62(message.request_id), - WireVarInt62(message.track_alias), - WireFullTrackName(message.full_track_name), - WireUint8(message.subscriber_priority), - WireDeliveryOrder(message.group_order), WireBoolean(message.forward), - WireVarInt62(message.filter_type), WireVarInt62(message.start->group), - WireVarInt62(message.start->object), - WireKeyValuePairList(parameters)); + break; case MoqtFilterType::kAbsoluteRange: - if (!message.start.has_value() || !message.end_group.has_value()) { - return quiche::QuicheBuffer(); - } - if (*message.end_group < message.start->group) { + if (!message.end_group.has_value() || !message.start.has_value() || + *message.end_group < message.start->group) { QUICHE_BUG(MoqtFramer_invalid_end_group) << "Invalid object range"; return quiche::QuicheBuffer(); } - return SerializeControlMessage( - MoqtMessageType::kSubscribe, WireVarInt62(message.request_id), - WireVarInt62(message.track_alias), - WireFullTrackName(message.full_track_name), - WireUint8(message.subscriber_priority), - WireDeliveryOrder(message.group_order), WireBoolean(message.forward), - WireVarInt62(message.filter_type), WireVarInt62(message.start->group), - WireVarInt62(message.start->object), WireVarInt62(*message.end_group), - WireKeyValuePairList(parameters)); + end_group = *message.end_group; + [[fallthrough]]; + case MoqtFilterType::kAbsoluteStart: + if (!message.start.has_value()) { + QUICHE_BUG(MoqtFramer_invalid_start) << "Filter requires start"; + return quiche::QuicheBuffer(); + } + start_group = message.start->group; + start_object = message.start->object; + break; default: QUICHE_BUG(MoqtFramer_end_group_missing) << "Subscribe framing error."; return quiche::QuicheBuffer(); } + return SerializeControlMessage( + MoqtMessageType::kSubscribe, WireVarInt62(message.request_id), + WireFullTrackName(message.full_track_name), + WireUint8(message.subscriber_priority), + WireDeliveryOrder(message.group_order), WireBoolean(message.forward), + WireVarInt62(message.filter_type), + WireOptional<WireVarInt62>(start_group), + WireOptional<WireVarInt62>(start_object), + WireOptional<WireVarInt62>(end_group), WireKeyValuePairList(parameters)); } quiche::QuicheBuffer MoqtFramer::SerializeSubscribeOk( @@ -465,6 +455,7 @@ if (message.largest_location.has_value()) { return SerializeControlMessage( MoqtMessageType::kSubscribeOk, WireVarInt62(message.request_id), + WireVarInt62(message.track_alias), WireVarInt62(message.expires.ToMilliseconds()), WireDeliveryOrder(message.group_order), WireUint8(1), WireVarInt62(message.largest_location->group), @@ -473,6 +464,7 @@ } return SerializeControlMessage( MoqtMessageType::kSubscribeOk, WireVarInt62(message.request_id), + WireVarInt62(message.track_alias), WireVarInt62(message.expires.ToMilliseconds()), WireDeliveryOrder(message.group_order), WireUint8(0), WireKeyValuePairList(parameters)); @@ -483,8 +475,7 @@ return SerializeControlMessage( MoqtMessageType::kSubscribeError, WireVarInt62(message.request_id), WireVarInt62(message.error_code), - WireStringWithVarInt62Length(message.reason_phrase), - WireVarInt62(message.track_alias)); + WireStringWithVarInt62Length(message.reason_phrase)); } quiche::QuicheBuffer MoqtFramer::SerializeUnsubscribe(
diff --git a/quiche/quic/moqt/moqt_framer_test.cc b/quiche/quic/moqt/moqt_framer_test.cc index c3780db..d88673e 100644 --- a/quiche/quic/moqt/moqt_framer_test.cc +++ b/quiche/quic/moqt/moqt_framer_test.cc
@@ -393,7 +393,6 @@ MoqtFilterType::kAbsoluteStart, MoqtFilterType::kAbsoluteRange}) { MoqtSubscribe subscribe = { /*subscribe_id=*/3, - /*track_alias=*/4, /*full_track_name=*/FullTrackName({"foo", "abcd"}), /*subscriber_priority=*/0x20, /*group_order=*/std::nullopt, @@ -412,7 +411,6 @@ TEST_F(MoqtFramerSimpleTest, SubscribeEndBeforeStart) { MoqtSubscribe subscribe = { /*subscribe_id=*/3, - /*track_alias=*/4, /*full_track_name=*/FullTrackName({"foo", "abcd"}), /*subscriber_priority=*/0x20, /*group_order=*/std::nullopt, @@ -428,6 +426,42 @@ EXPECT_EQ(buffer.size(), 0); } +TEST_F(MoqtFramerSimpleTest, AbsoluteRangeStartMissing) { + MoqtSubscribe subscribe = { + /*subscribe_id=*/3, + /*full_track_name=*/FullTrackName({"foo", "abcd"}), + /*subscriber_priority=*/0x20, + /*group_order=*/std::nullopt, + /*forward=*/true, + /*filter_type=*/MoqtFilterType::kAbsoluteRange, + /*start=*/std::nullopt, + /*end_group=*/std::make_optional<uint64_t>(3ULL), + VersionSpecificParameters(AuthTokenType::kOutOfBand, "bar"), + }; + quiche::QuicheBuffer buffer; + EXPECT_QUICHE_BUG(buffer = framer_.SerializeSubscribe(subscribe), + "Invalid object range"); + EXPECT_EQ(buffer.size(), 0); +} + +TEST_F(MoqtFramerSimpleTest, AbsoluteRangeEndMissing) { + MoqtSubscribe subscribe = { + /*subscribe_id=*/3, + /*full_track_name=*/FullTrackName({"foo", "abcd"}), + /*subscriber_priority=*/0x20, + /*group_order=*/std::nullopt, + /*forward=*/true, + /*filter_type=*/MoqtFilterType::kAbsoluteRange, + /*start=*/std::make_optional<Location>(4, 3), + /*end_group=*/std::nullopt, + VersionSpecificParameters(AuthTokenType::kOutOfBand, "bar"), + }; + quiche::QuicheBuffer buffer; + EXPECT_QUICHE_BUG(buffer = framer_.SerializeSubscribe(subscribe), + "Invalid object range"); + EXPECT_EQ(buffer.size(), 0); +} + TEST_F(MoqtFramerSimpleTest, FetchEndBeforeStart) { MoqtFetch fetch = { /*subscribe_id =*/1,
diff --git a/quiche/quic/moqt/moqt_messages.cc b/quiche/quic/moqt/moqt_messages.cc index 8600f62..5b5ee77 100644 --- a/quiche/quic/moqt/moqt_messages.cc +++ b/quiche/quic/moqt/moqt_messages.cc
@@ -134,7 +134,6 @@ return absl::StatusCode::kNotFound; case RequestErrorCode::kInvalidJoiningSubscribeId: case RequestErrorCode::kMalformedAuthToken: - case RequestErrorCode::kUnknownAuthTokenAlias: return absl::StatusCode::kInvalidArgument; case RequestErrorCode::kExpiredAuthToken: return absl::StatusCode::kUnauthenticated;
diff --git a/quiche/quic/moqt/moqt_messages.h b/quiche/quic/moqt/moqt_messages.h index 232c8ab..69beaa9 100644 --- a/quiche/quic/moqt/moqt_messages.h +++ b/quiche/quic/moqt/moqt_messages.h
@@ -337,11 +337,9 @@ kNamespacePrefixUnknown = 0x4, // SUBSCRIBE_ANNOUNCES_ERROR only. kInvalidRange = 0x5, // SUBSCRIBE_ERROR and FETCH_ERROR only. kNamespacePrefixOverlap = 0x5, // SUBSCRIBE_ANNOUNCES_ERROR only. - kRetryTrackAlias = 0x6, // SUBSCRIBE_ERROR only. kNoObjects = 0x6, // FETCH_ERROR only. kInvalidJoiningSubscribeId = 0x7, // FETCH_ERROR only. kMalformedAuthToken = 0x10, - kUnknownAuthTokenAlias = 0x11, kExpiredAuthToken = 0x12, }; @@ -619,7 +617,6 @@ struct QUICHE_EXPORT MoqtSubscribe { uint64_t request_id; - uint64_t track_alias; FullTrackName full_track_name; MoqtPriority subscriber_priority; std::optional<MoqtDeliveryOrder> group_order; @@ -632,6 +629,7 @@ struct QUICHE_EXPORT MoqtSubscribeOk { uint64_t request_id; + uint64_t track_alias; // The message uses ms, but expires is in us. quic::QuicTimeDelta expires = quic::QuicTimeDelta::FromMilliseconds(0); MoqtDeliveryOrder group_order; @@ -644,7 +642,6 @@ uint64_t request_id; RequestErrorCode error_code; std::string reason_phrase; - uint64_t track_alias; }; struct QUICHE_EXPORT MoqtUnsubscribe {
diff --git a/quiche/quic/moqt/moqt_parser.cc b/quiche/quic/moqt/moqt_parser.cc index 75dcc5f..d381d04 100644 --- a/quiche/quic/moqt/moqt_parser.cc +++ b/quiche/quic/moqt/moqt_parser.cc
@@ -370,7 +370,6 @@ uint64_t filter, group, object; uint8_t group_order, forward; if (!reader.ReadVarInt62(&subscribe.request_id) || - !reader.ReadVarInt62(&subscribe.track_alias) || !ReadFullTrackName(reader, subscribe.full_track_name) || !reader.ReadUInt8(&subscribe.subscriber_priority) || !reader.ReadUInt8(&group_order) || !reader.ReadUInt8(&forward) || @@ -436,6 +435,7 @@ uint8_t group_order; uint8_t content_exists; if (!reader.ReadVarInt62(&subscribe_ok.request_id) || + !reader.ReadVarInt62(&subscribe_ok.track_alias) || !reader.ReadVarInt62(&milliseconds) || !reader.ReadUInt8(&group_order) || !reader.ReadUInt8(&content_exists)) { return 0; @@ -479,8 +479,7 @@ uint64_t error_code; if (!reader.ReadVarInt62(&subscribe_error.request_id) || !reader.ReadVarInt62(&error_code) || - !reader.ReadStringVarInt62(subscribe_error.reason_phrase) || - !reader.ReadVarInt62(&subscribe_error.track_alias)) { + !reader.ReadStringVarInt62(subscribe_error.reason_phrase)) { return 0; } subscribe_error.error_code = static_cast<RequestErrorCode>(error_code);
diff --git a/quiche/quic/moqt/moqt_parser_test.cc b/quiche/quic/moqt/moqt_parser_test.cc index 4f6535a..7e4f5ce 100644 --- a/quiche/quic/moqt/moqt_parser_test.cc +++ b/quiche/quic/moqt/moqt_parser_test.cc
@@ -580,8 +580,8 @@ webtransport::test::InMemoryStream stream(/*stream_id=*/0); MoqtControlParser parser(kWebTrans, &stream, visitor_); char subscribe[] = { - 0x03, 0x00, 0x1b, 0x01, 0x02, - 0x01, 0x03, 0x66, 0x6f, 0x6f, // track_namespace = "foo" + 0x03, 0x00, 0x1a, 0x01, 0x01, + 0x03, 0x66, 0x6f, 0x6f, // track_namespace = "foo" 0x04, 0x61, 0x62, 0x63, 0x64, // track_name = "abcd" 0x20, 0x02, 0x01, // priority, order, forward 0x02, // filter_type = kLatestObject @@ -598,8 +598,8 @@ webtransport::test::InMemoryStream stream(/*stream_id=*/0); MoqtControlParser parser(kRawQuic, &stream, visitor_); char subscribe[] = { - 0x03, 0x00, 0x17, 0x01, 0x02, - 0x01, 0x03, 0x66, 0x6f, 0x6f, // track_namespace = "foo" + 0x03, 0x00, 0x16, 0x01, 0x01, + 0x03, 0x66, 0x6f, 0x6f, // track_namespace = "foo" 0x04, 0x61, 0x62, 0x63, 0x64, // track_name = "abcd" 0x20, 0x02, 0x01, // priority, order, forward 0x02, // filter_type = kLatestObject @@ -618,8 +618,8 @@ webtransport::test::InMemoryStream stream(/*stream_id=*/0); MoqtControlParser parser(kRawQuic, &stream, visitor_); char subscribe[] = { - 0x03, 0x00, 0x17, 0x01, 0x02, - 0x01, 0x03, 0x66, 0x6f, 0x6f, // track_namespace = "foo" + 0x03, 0x00, 0x16, 0x01, 0x01, + 0x03, 0x66, 0x6f, 0x6f, // track_namespace = "foo" 0x04, 0x61, 0x62, 0x63, 0x64, // track_name = "abcd" 0x20, 0x02, 0x01, // priority, order, forward 0x02, // filter_type = kLatestObject @@ -638,8 +638,8 @@ webtransport::test::InMemoryStream stream(/*stream_id=*/0); MoqtControlParser parser(kRawQuic, &stream, visitor_); char subscribe[] = { - 0x03, 0x00, 0x15, 0x01, 0x02, - 0x01, 0x03, 0x66, 0x6f, 0x6f, // track_namespace = "foo" + 0x03, 0x00, 0x14, 0x01, 0x01, + 0x03, 0x66, 0x6f, 0x6f, // track_namespace = "foo" 0x04, 0x61, 0x62, 0x63, 0x64, // track_name = "abcd" 0x20, 0x02, 0x01, // priority, order, forward 0x02, // filter_type = kLatestObject @@ -657,8 +657,8 @@ webtransport::test::InMemoryStream stream(/*stream_id=*/0); MoqtControlParser parser(kRawQuic, &stream, visitor_); char subscribe[] = { - 0x03, 0x00, 0x19, 0x01, 0x02, 0x01, 0x03, 0x66, - 0x6f, 0x6f, // track_namespace = "foo" + 0x03, 0x00, 0x18, 0x01, 0x01, 0x03, 0x66, 0x6f, + 0x6f, // track_namespace = "foo" 0x04, 0x61, 0x62, 0x63, 0x64, // track_name = "abcd" 0x20, 0x02, 0x01, // priority, order, forward 0x02, // filter_type = kLatestObject @@ -676,8 +676,8 @@ webtransport::test::InMemoryStream stream(/*stream_id=*/0); MoqtControlParser parser(kRawQuic, &stream, visitor_); char subscribe[] = { - 0x03, 0x00, 0x15, 0x01, 0x02, - 0x01, 0x03, 0x66, 0x6f, 0x6f, // track_namespace = "foo" + 0x03, 0x00, 0x14, 0x01, 0x01, + 0x03, 0x66, 0x6f, 0x6f, // track_namespace = "foo" 0x04, 0x61, 0x62, 0x63, 0x64, // track_name = "abcd" 0x20, 0x02, 0x01, // priority, order, forward 0x02, // filter_type = kLatestObject @@ -696,8 +696,8 @@ webtransport::test::InMemoryStream stream(/*stream_id=*/0); MoqtControlParser parser(kRawQuic, &stream, visitor_); char subscribe[] = { - 0x03, 0x00, 0x15, 0x01, 0x02, - 0x01, 0x03, 0x66, 0x6f, 0x6f, // track_namespace = "foo" + 0x03, 0x00, 0x14, 0x01, 0x01, + 0x03, 0x66, 0x6f, 0x6f, // track_namespace = "foo" 0x04, 0x61, 0x62, 0x63, 0x64, // track_name = "abcd" 0x20, 0x02, 0x01, // priority, order, forward 0x02, // filter_type = kLatestObject @@ -716,8 +716,8 @@ webtransport::test::InMemoryStream stream(/*stream_id=*/0); MoqtControlParser parser(kRawQuic, &stream, visitor_); char subscribe[] = { - 0x03, 0x00, 0x17, 0x01, 0x02, 0x01, - 0x03, 0x66, 0x6f, 0x6f, // track_namespace = "foo" + 0x03, 0x00, 0x16, 0x01, 0x01, 0x03, + 0x66, 0x6f, 0x6f, // track_namespace = "foo" 0x04, 0x61, 0x62, 0x63, 0x64, // track_name = "abcd" 0x20, 0x02, 0x01, // priority, order, forward 0x02, // filter_type = kLatestObject @@ -735,19 +735,38 @@ webtransport::test::InMemoryStream stream(/*stream_id=*/0); MoqtControlParser parser(kRawQuic, &stream, visitor_); char subscribe[] = { - 0x03, 0x00, 0x1d, 0x01, 0x02, // id and alias - 0x01, 0x03, 0x66, 0x6f, 0x6f, // track_namespace = "foo" - 0x04, 0x61, 0x62, 0x63, 0x64, // track_name = "abcd" - 0x20, // subscriber priority = 0x20 - 0x03, // group order = invalid - 0x01, // forward = true - 0x03, // Filter type: Absolute Start - 0x04, // start_group = 4 (relative previous) - 0x01, // start_object = 1 (absolute) + 0x03, + 0x00, + 0x1c, + 0x01, // id + 0x01, + 0x03, + 0x66, + 0x6f, + 0x6f, // track_namespace = "foo" + 0x04, + 0x61, + 0x62, + 0x63, + 0x64, // track_name = "abcd" + 0x20, // subscriber priority = 0x20 + 0x03, // group order = invalid + 0x01, // forward = true + 0x03, // Filter type: Absolute Start + 0x04, // start_group = 4 (relative previous) + 0x01, // start_object = 1 (absolute) // No EndGroup or EndObject - 0x02, // 2 parameters - 0x02, 0x67, 0x10, // delivery_timeout = 10000 ms - 0x03, 0x05, 0x03, 0x00, 0x62, 0x61, 0x72, // authorization_tag = "bar" + 0x02, // 2 parameters + 0x02, + 0x67, + 0x10, // delivery_timeout = 10000 ms + 0x03, + 0x05, + 0x03, + 0x00, + 0x62, + 0x61, + 0x72, // authorization_tag = "bar" }; stream.Receive(absl::string_view(subscribe, sizeof(subscribe)), false); parser.ReadAndDispatchMessages(); @@ -760,19 +779,38 @@ webtransport::test::InMemoryStream stream(/*stream_id=*/0); MoqtControlParser parser(kRawQuic, &stream, visitor_); char subscribe[] = { - 0x03, 0x00, 0x1d, 0x01, 0x02, // id and alias - 0x01, 0x03, 0x66, 0x6f, 0x6f, // track_namespace = "foo" - 0x04, 0x61, 0x62, 0x63, 0x64, // track_name = "abcd" - 0x20, // subscriber priority = 0x20 - 0x02, // group order = descending - 0x02, // forward = invalid - 0x03, // Filter type: Absolute Start - 0x04, // start_group = 4 (relative previous) - 0x01, // start_object = 1 (absolute) + 0x03, + 0x00, + 0x1c, + 0x01, // id + 0x01, + 0x03, + 0x66, + 0x6f, + 0x6f, // track_namespace = "foo" + 0x04, + 0x61, + 0x62, + 0x63, + 0x64, // track_name = "abcd" + 0x20, // subscriber priority = 0x20 + 0x02, // group order = descending + 0x02, // forward = invalid + 0x03, // Filter type: Absolute Start + 0x04, // start_group = 4 (relative previous) + 0x01, // start_object = 1 (absolute) // No EndGroup or EndObject - 0x02, // 2 parameters - 0x02, 0x67, 0x10, // delivery_timeout = 10000 ms - 0x03, 0x05, 0x03, 0x00, 0x62, 0x61, 0x72, // authorization_tag = "bar" + 0x02, // 2 parameters + 0x02, + 0x67, + 0x10, // delivery_timeout = 10000 ms + 0x03, + 0x05, + 0x03, + 0x00, + 0x62, + 0x61, + 0x72, // authorization_tag = "bar" }; stream.Receive(absl::string_view(subscribe, sizeof(subscribe)), false); parser.ReadAndDispatchMessages(); @@ -785,19 +823,38 @@ webtransport::test::InMemoryStream stream(/*stream_id=*/0); MoqtControlParser parser(kRawQuic, &stream, visitor_); char subscribe[] = { - 0x03, 0x00, 0x1d, 0x01, 0x02, // id and alias - 0x01, 0x03, 0x66, 0x6f, 0x6f, // track_namespace = "foo" - 0x04, 0x61, 0x62, 0x63, 0x64, // track_name = "abcd" - 0x20, // subscriber priority = 0x20 - 0x02, // group order = descending - 0x01, // forward = true - 0x05, // Filter type: Absolute Start - 0x04, // start_group = 4 (relative previous) - 0x01, // start_object = 1 (absolute) + 0x03, + 0x00, + 0x1c, + 0x01, // id + 0x01, + 0x03, + 0x66, + 0x6f, + 0x6f, // track_namespace = "foo" + 0x04, + 0x61, + 0x62, + 0x63, + 0x64, // track_name = "abcd" + 0x20, // subscriber priority = 0x20 + 0x02, // group order = descending + 0x01, // forward = true + 0x05, // Filter type: Absolute Start + 0x04, // start_group = 4 (relative previous) + 0x01, // start_object = 1 (absolute) // No EndGroup or EndObject - 0x02, // 2 parameters - 0x02, 0x67, 0x10, // delivery_timeout = 10000 ms - 0x03, 0x05, 0x03, 0x00, 0x62, 0x61, 0x72, // authorization_tag = "bar" + 0x02, // 2 parameters + 0x02, + 0x67, + 0x10, // delivery_timeout = 10000 ms + 0x03, + 0x05, + 0x03, + 0x00, + 0x62, + 0x61, + 0x72, // authorization_tag = "bar" }; stream.Receive(absl::string_view(subscribe, sizeof(subscribe)), false); parser.ReadAndDispatchMessages(); @@ -810,8 +867,8 @@ webtransport::test::InMemoryStream stream(/*stream_id=*/0); MoqtControlParser parser(kWebTrans, &stream, visitor_); char subscribe_ok[] = { - 0x04, 0x00, 0x11, 0x01, 0x03, // subscribe_id = 1, expires = 3 - 0x02, 0x01, // group_order = 2, content exists + 0x04, 0x00, 0x12, 0x01, 0x02, 0x03, // subscribe_id, alias, expires = 3 + 0x02, 0x01, // group_order = 2, content exists 0x0c, 0x14, // largest_group_id = 12, largest_object_id = 20, 0x02, // 2 parameters 0x02, 0x67, 0x10, // delivery_timeout = 10000 @@ -997,7 +1054,7 @@ webtransport::test::InMemoryStream stream(/*stream_id=*/0); MoqtControlParser parser(kRawQuic, &stream, visitor_); char subscribe[] = { - 0x03, 0x00, 0x18, 0x01, 0x02, // id and alias + 0x03, 0x00, 0x17, 0x01, // id 0x01, 0x03, 0x66, 0x6f, 0x6f, // track_namespace = "foo" 0x04, 0x61, 0x62, 0x63, 0x64, // track_name = "abcd" 0x20, 0x02, 0x01, // priority = 0x20, group order, forward @@ -1019,7 +1076,7 @@ webtransport::test::InMemoryStream stream(/*stream_id=*/0); MoqtControlParser parser(kRawQuic, &stream, visitor_); char subscribe[] = { - 0x03, 0x00, 0x18, 0x01, 0x02, // id and alias + 0x03, 0x00, 0x17, 0x01, // id 0x01, 0x03, 0x66, 0x6f, 0x6f, // track_namespace = "foo" 0x04, 0x61, 0x62, 0x63, 0x64, // track_name = "abcd" 0x20, 0x08, 0x01, // priority, invalid order, forward @@ -1037,7 +1094,7 @@ webtransport::test::InMemoryStream stream(/*stream_id=*/0); MoqtControlParser parser(kRawQuic, &stream, visitor_); char subscribe[] = { - 0x03, 0x00, 0x1a, 0x01, 0x02, // id and alias + 0x03, 0x00, 0x19, 0x01, // id 0x01, 0x03, 0x66, 0x6f, 0x6f, // track_namespace = "foo" 0x04, 0x61, 0x62, 0x63, 0x64, // track_name = "abcd" 0x20, 0x02, 0x01, // priority, order, forward @@ -1062,7 +1119,7 @@ webtransport::test::InMemoryStream stream(/*stream_id=*/0); MoqtControlParser parser(kRawQuic, &stream, visitor_); char subscribe[] = { - 0x03, 0x00, 0x1b, 0x01, 0x02, // id and alias + 0x03, 0x00, 0x1a, 0x01, // id 0x01, 0x03, 0x66, 0x6f, 0x6f, // track_namespace = "foo" 0x04, 0x61, 0x62, 0x63, 0x64, // track_name = "abcd" 0x20, 0x02, 0x01, // priority, order, forward @@ -1088,7 +1145,7 @@ webtransport::test::InMemoryStream stream(/*stream_id=*/0); MoqtControlParser parser(kRawQuic, &stream, visitor_); char subscribe[] = { - 0x03, 0x00, 0x19, 0x01, 0x02, // id and alias + 0x03, 0x00, 0x18, 0x01, // id 0x01, 0x03, 0x66, 0x6f, 0x6f, // track_namespace = "foo" 0x04, 0x61, 0x62, 0x63, 0x64, // track_name = "abcd" 0x20, 0x02, 0x01, // priority, order, forward @@ -1109,7 +1166,7 @@ webtransport::test::InMemoryStream stream(/*stream_id=*/0); MoqtControlParser parser(kRawQuic, &stream, visitor_); char subscribe[] = { - 0x03, 0x00, 0x14, 0x01, 0x02, // id and alias + 0x03, 0x00, 0x13, 0x01, // id 0x01, 0x03, 0x66, 0x6f, 0x6f, // track_namespace = "foo" 0x04, 0x61, 0x62, 0x63, 0x64, // track_name = "abcd" 0x20, 0x02, 0x01, // priority, order, forward
diff --git a/quiche/quic/moqt/moqt_session.cc b/quiche/quic/moqt/moqt_session.cc index eb11c26..ed75729 100644 --- a/quiche/quic/moqt/moqt_session.cc +++ b/quiche/quic/moqt/moqt_session.cc
@@ -581,7 +581,7 @@ subscribe.start = std::nullopt; subscribe.end_group = std::nullopt; subscribe.parameters = parameters; - if (!Subscribe(subscribe, visitor, std::nullopt)) { + if (!Subscribe(subscribe, visitor)) { return false; } MoqtFetch fetch; @@ -707,12 +707,13 @@ void MoqtSession::DestroySubscription(SubscribeRemoteTrack* subscribe) { subscribe->visitor()->OnSubscribeDone(subscribe->full_track_name()); subscribe_by_name_.erase(subscribe->full_track_name()); - subscribe_by_alias_.erase(subscribe->track_alias()); + if (subscribe->track_alias().has_value()) { + subscribe_by_alias_.erase(*subscribe->track_alias()); + } } bool MoqtSession::Subscribe(MoqtSubscribe& message, - SubscribeRemoteTrack::Visitor* visitor, - std::optional<uint64_t> provided_track_alias) { + SubscribeRemoteTrack::Visitor* visitor) { // TODO(martinduke): support authorization info if (next_request_id_ >= peer_max_request_id_) { if (!last_requests_blocked_sent_.has_value() || @@ -734,24 +735,12 @@ << " which is already subscribed"; return false; } - if (provided_track_alias.has_value() && - subscribe_by_alias_.contains(*provided_track_alias)) { - Error(MoqtError::kProtocolViolation, "Provided track alias already in use"); - return false; - } if (received_goaway_ || sent_goaway_) { QUIC_DLOG(INFO) << ENDPOINT << "Tried to send SUBSCRIBE after GOAWAY"; return false; } message.request_id = next_request_id_; next_request_id_ += 2; - if (provided_track_alias.has_value()) { - message.track_alias = *provided_track_alias; - next_remote_track_alias_ = - std::max(next_remote_track_alias_, *provided_track_alias) + 1; - } else { - message.track_alias = next_remote_track_alias_++; - } if (SupportsObjectAck() && visitor != nullptr) { // Since we do not expose subscribe IDs directly in the API, instead wrap // the session and subscribe ID in a callback. @@ -768,7 +757,6 @@ << message.full_track_name; auto track = std::make_unique<SubscribeRemoteTrack>(message, visitor); subscribe_by_name_.emplace(message.full_track_name, track.get()); - subscribe_by_alias_.emplace(message.track_alias, track.get()); upstream_by_id_.emplace(message.request_id, std::move(track)); return true; } @@ -1012,12 +1000,11 @@ void MoqtSession::ControlStream::SendSubscribeError( uint64_t request_id, RequestErrorCode error_code, - absl::string_view reason_phrase, uint64_t track_alias) { + absl::string_view reason_phrase) { MoqtSubscribeError subscribe_error; subscribe_error.request_id = request_id; subscribe_error.error_code = error_code; subscribe_error.reason_phrase = reason_phrase; - subscribe_error.track_alias = track_alias; SendOrBufferMessage( session_->framer_.SerializeSubscribeError(subscribe_error)); } @@ -1042,7 +1029,7 @@ if (session_->sent_goaway_) { QUIC_DLOG(INFO) << ENDPOINT << "Received a SUBSCRIBE after GOAWAY"; SendSubscribeError(message.request_id, RequestErrorCode::kUnauthorized, - "SUBSCRIBE after GOAWAY", message.track_alias); + "SUBSCRIBE after GOAWAY"); return; } if (session_->subscribed_track_names_.contains(message.full_track_name)) { @@ -1058,7 +1045,7 @@ << " rejected by the application: " << track_publisher.status(); SendSubscribeError(message.request_id, RequestErrorCode::kTrackDoesNotExist, - track_publisher.status().message(), message.track_alias); + track_publisher.status().message()); return; } @@ -1111,6 +1098,13 @@ } SubscribeRemoteTrack* subscribe = static_cast<SubscribeRemoteTrack*>(track); subscribe->OnObjectOrOk(); + auto [it, success] = + session_->subscribe_by_alias_.try_emplace(message.track_alias, subscribe); + if (!success) { + session_->Error(MoqtError::kDuplicateTrackAlias, ""); + return; + } + subscribe->set_track_alias(message.track_alias); // TODO(martinduke): Handle expires field. if (message.largest_location.has_value()) { subscribe->TruncateStart(message.largest_location->next()); @@ -1151,16 +1145,10 @@ // an error due to a duplicate track name. The other entries for this // subscribe will be deleted after calling Subscribe(). session_->subscribe_by_name_.erase(subscribe->full_track_name()); - if (message.error_code == RequestErrorCode::kRetryTrackAlias) { - // Automatically resubscribe with new alias. - MoqtSubscribe& subscribe_message = subscribe->GetSubscribe(); - session_->Subscribe(subscribe_message, subscribe->visitor(), - message.track_alias); - } else if (subscribe->visitor() != nullptr) { + if (subscribe->visitor() != nullptr) { subscribe->visitor()->OnReply(subscribe->full_track_name(), std::nullopt, message.reason_phrase); } - session_->subscribe_by_alias_.erase(subscribe->track_alias()); session_->upstream_by_id_.erase(subscribe->request_id()); } @@ -1849,8 +1837,8 @@ if (it == session_->subscribe_by_alias_.end()) { QUIC_DLOG(INFO) << ENDPOINT << "Received object for a track with no SUBSCRIBE"; - // This is a not a session error because there might be an UNSUBSCRIBE in - // flight. + // This is a not a session error because there might be an UNSUBSCRIBE or + // SUBSCRIBE_OK (containing the track alias) in flight. stream_->SendStopSending(kResetCodeCanceled); return; } @@ -1898,7 +1886,6 @@ : session_(session), track_publisher_(track_publisher), request_id_(subscribe.request_id), - track_alias_(subscribe.track_alias), filter_type_(subscribe.filter_type), forward_(subscribe.forward), window_(SubscribeMessageToWindow(subscribe)), @@ -1993,19 +1980,26 @@ } MoqtSubscribeOk subscribe_ok; subscribe_ok.request_id = request_id_; + subscribe_ok.track_alias = session_->next_local_track_alias_++; subscribe_ok.group_order = track_publisher_->GetDeliveryOrder(); subscribe_ok.largest_location = largest_location; + track_alias_.emplace(subscribe_ok.track_alias); // TODO(martinduke): Support sending DELIVERY_TIMEOUT parameter as the // publisher. stream->SendOrBufferMessage( session_->framer_.SerializeSubscribeOk(subscribe_ok)); + if (!PublisherHasData(*track_publisher_)) { + return; + } + // TODO(martinduke): If we buffer objects that arrived previously, the arrival + // of the track alias disambiguates what subscription they belong to. Send + // them. } void MoqtSession::PublishedSubscription::OnSubscribeRejected( MoqtSubscribeErrorReason reason, std::optional<uint64_t> track_alias) { session_->GetControlStream()->SendSubscribeError( - request_id_, reason.error_code, reason.reason_phrase, - track_alias.value_or(track_alias_)); + request_id_, reason.error_code, reason.reason_phrase); session_->published_subscriptions_.erase(request_id_); // No class access below this line! } @@ -2331,6 +2325,9 @@ void MoqtSession::OutgoingDataStream::SendObjects( PublishedSubscription& subscription) { + if (!subscription.track_alias().has_value()) { + return; + } while (stream_->CanWrite()) { std::optional<PublishedObject> object = subscription.publisher().GetCachedObject(index_.group, index_.subgroup, @@ -2362,7 +2359,7 @@ return; } if (!session_->WriteObjectToStream( - stream_, subscription.track_alias(), object->metadata, + stream_, *subscription.track_alias(), object->metadata, std::move(object->payload), stream_type_, !stream_header_written_, object->fin_after_this)) { // WriteObjectToStream() closes the connection on error, meaning that @@ -2483,9 +2480,11 @@ << "Got notification about an object that is not in the cache"; return; } - + if (!track_alias_.has_value()) { + return; + } MoqtObject header; - header.track_alias = track_alias(); + header.track_alias = *track_alias_; header.group_id = object->metadata.location.group; header.object_id = object->metadata.location.object; header.publisher_priority = object->metadata.publisher_priority;
diff --git a/quiche/quic/moqt/moqt_session.h b/quiche/quic/moqt/moqt_session.h index 25c5ae2..79b4613 100644 --- a/quiche/quic/moqt/moqt_session.h +++ b/quiche/quic/moqt/moqt_session.h
@@ -291,8 +291,7 @@ void SendOrBufferMessage(quiche::QuicheBuffer message, bool fin = false); void SendSubscribeError(uint64_t request_id, RequestErrorCode error_code, - absl::string_view reason_phrase, - uint64_t track_alias); + absl::string_view reason_phrase); private: friend class test::MoqtSessionPeer; @@ -363,7 +362,7 @@ uint64_t request_id() const { return request_id_; } MoqtTrackPublisher& publisher() { return *track_publisher_; } - uint64_t track_alias() const { return track_alias_; } + std::optional<uint64_t> track_alias() const { return track_alias_; } std::optional<Location> largest_sent() const { return largest_sent_; } MoqtPriority subscriber_priority() const { return subscriber_priority_; } std::optional<MoqtDeliveryOrder> subscriber_delivery_order() const { @@ -447,6 +446,7 @@ uint64_t streams_opened() const { return streams_opened_; } private: + friend class test::MoqtSessionPeer; SendStreamMap& stream_map(); quic::Perspective perspective() const { return session_->parameters_.perspective; @@ -462,7 +462,7 @@ MoqtSession* session_; std::shared_ptr<MoqtTrackPublisher> track_publisher_; uint64_t request_id_; - uint64_t track_alias_; + std::optional<const uint64_t> track_alias_; MoqtFilterType filter_type_; bool forward_; // If window_ is nullopt, any arriving objects are ignored. This could be @@ -711,10 +711,9 @@ // is present. void SendControlMessage(quiche::QuicheBuffer message); - // Returns false if the SUBSCRIBE isn't sent. |provided_track_alias| has a - // value only if this call is due to a SUBSCRIBE_ERROR. - bool Subscribe(MoqtSubscribe& message, SubscribeRemoteTrack::Visitor* visitor, - std::optional<uint64_t> provided_track_alias = std::nullopt); + // Returns false if the SUBSCRIBE isn't sent. + bool Subscribe(MoqtSubscribe& message, + SubscribeRemoteTrack::Visitor* visitor); // Opens a new data stream, or queues it if the session is flow control // blocked. @@ -793,8 +792,7 @@ absl::flat_hash_map<uint64_t, SubscribeRemoteTrack*> subscribe_by_alias_; // All SUBSCRIBEs, indexed by track name. absl::flat_hash_map<FullTrackName, SubscribeRemoteTrack*> subscribe_by_name_; - // The next track alias to guess on a SUBSCRIBE. - uint64_t next_remote_track_alias_ = 0; + // The next subscribe ID that the local endpoint can send. uint64_t next_request_id_ = 0; // The local endpoint can send subscribe IDs less than this value.
diff --git a/quiche/quic/moqt/moqt_session_test.cc b/quiche/quic/moqt/moqt_session_test.cc index 55e47dc..9862e11 100644 --- a/quiche/quic/moqt/moqt_session_test.cc +++ b/quiche/quic/moqt/moqt_session_test.cc
@@ -65,7 +65,6 @@ MoqtSubscribe DefaultSubscribe(uint64_t request_id) { MoqtSubscribe subscribe = { request_id, - /*track_alias=*/2, kDefaultTrackName(), /*subscriber_priority=*/0x80, /*group_order=*/std::nullopt, @@ -156,7 +155,7 @@ // publish objects. MoqtObjectListener* ReceiveSubscribeSynchronousOk( MockTrackPublisher* publisher, MoqtSubscribe& subscribe, - MoqtControlParserVisitor* control_parser) { + MoqtControlParserVisitor* control_parser, uint64_t track_alias = 0) { MoqtObjectListener* listener_ptr = nullptr; EXPECT_CALL(*publisher, AddObjectListener) .WillOnce([&](MoqtObjectListener* listener) { @@ -170,6 +169,7 @@ } MoqtSubscribeOk expected_ok = { /*request_id=*/subscribe.request_id, + track_alias, /*expires=*/quic::QuicTimeDelta::FromMilliseconds(0), /*group_order=*/MoqtDeliveryOrder::kAscending, (*track_status == MoqtTrackStatusCode::kInProgress) @@ -507,8 +507,7 @@ mock_stream_, Writev(ControlMessageOfType(MoqtMessageType::kSubscribeError), _)); listener->OnSubscribeRejected( - MoqtSubscribeErrorReason(RequestErrorCode::kInternalError, "Test error"), - request.track_alias); + MoqtSubscribeErrorReason(RequestErrorCode::kInternalError, "Test error")); EXPECT_EQ(MoqtSessionPeer::GetSubscription(&session_, kDefaultPeerRequestId), nullptr); } @@ -607,7 +606,8 @@ // Subscribe again, succeeds. request.request_id = 3; request.start = Location(12, 0); - ReceiveSubscribeSynchronousOk(track, request, stream_input.get()); + ReceiveSubscribeSynchronousOk(track, request, stream_input.get(), + /*track_alias=*/1); } TEST_F(MoqtSessionTest, RequestIdTooHigh) { @@ -638,7 +638,6 @@ stream_input->OnSubscribeMessage(request); // Second request is a protocol violation. - ++request.track_alias; request.full_track_name = FullTrackName({"dead", "beef"}); EXPECT_CALL(mock_session_, CloseSession(static_cast<uint64_t>(MoqtError::kInvalidRequestId), @@ -701,6 +700,7 @@ MoqtSubscribeOk ok = { /*request_id=*/0, + /*track_alias=*/2, /*expires=*/quic::QuicTimeDelta::FromMilliseconds(0), }; EXPECT_CALL(remote_track_visitor, OnReply(_, _, _)) @@ -720,7 +720,6 @@ EXPECT_CALL(mock_session_, GetStreamById(_)).WillOnce(Return(&mock_stream_)); MoqtSubscribe subscribe = { /*request_id=*/0, - /*track_alias=*/0, FullTrackName("foo", "bar"), kDefaultSubscriberPriority, /*group_order=*/std::nullopt, @@ -738,6 +737,7 @@ MoqtSubscribeOk ok = { /*request_id=*/0, + /*track_alias=*/2, /*expires=*/quic::QuicTimeDelta::FromMilliseconds(0), }; EXPECT_CALL(remote_track_visitor, OnReply(_, _, _)) @@ -763,6 +763,7 @@ VersionSpecificParameters()); MoqtSubscribeOk ok = { /*request_id=*/0, + /*track_alias=*/2, /*expires=*/quic::QuicTimeDelta::FromMilliseconds(0), }; EXPECT_CALL(remote_track_visitor, OnReply); @@ -773,7 +774,7 @@ EXPECT_TRUE(session_.SubscribeUpdate( FullTrackName("foo", "bar"), Location(2, 1), 9, std::nullopt, std::nullopt, VersionSpecificParameters())); - SubscribeRemoteTrack* track = MoqtSessionPeer::remote_track(&session_, 0); + SubscribeRemoteTrack* track = MoqtSessionPeer::remote_track(&session_, 2); EXPECT_FALSE(track->InWindow(Location(2, 0))); EXPECT_TRUE(track->InWindow(Location(2, 1))); EXPECT_TRUE(track->InWindow(Location(9, UINT64_MAX))); @@ -793,6 +794,7 @@ VersionSpecificParameters()); MoqtSubscribeOk ok = { /*request_id=*/0, + /*track_alias=*/2, /*expires=*/quic::QuicTimeDelta::FromMilliseconds(0), }; EXPECT_CALL(remote_track_visitor, OnReply); @@ -880,7 +882,6 @@ /*request_id=*/0, /*error_code=*/RequestErrorCode::kInvalidRange, /*reason_phrase=*/"deadbeef", - /*track_alias=*/2, }; EXPECT_CALL(remote_track_visitor, OnReply(_, _, _)) .WillOnce([&](const FullTrackName& ftn, @@ -897,7 +898,7 @@ MoqtSessionPeer::CreateControlStream(&session_, &mock_stream_); MockSubscribeRemoteTrackVisitor remote_track_visitor; MoqtSessionPeer::CreateRemoteTrack(&session_, DefaultSubscribe(), - &remote_track_visitor); + /*track_alias=*/2, &remote_track_visitor); EXPECT_CALL(mock_stream_, Writev(ControlMessageOfType(MoqtMessageType::kUnsubscribe), _)); EXPECT_NE(MoqtSessionPeer::remote_track(&session_, 2), nullptr); @@ -1054,7 +1055,8 @@ MockSubscribeRemoteTrackVisitor visitor_; FullTrackName ftn("foo", "bar"); std::string payload = "deadbeef"; - MoqtSessionPeer::CreateRemoteTrack(&session_, DefaultSubscribe(), &visitor_); + MoqtSessionPeer::CreateRemoteTrack(&session_, DefaultSubscribe(), + /*track_alias=*/2, &visitor_); MoqtObject object = { /*track_alias=*/2, /*group_sequence=*/0, @@ -1079,7 +1081,8 @@ MockSubscribeRemoteTrackVisitor visitor_; FullTrackName ftn("foo", "bar"); std::string payload = "deadbeef"; - MoqtSessionPeer::CreateRemoteTrack(&session_, DefaultSubscribe(), &visitor_); + MoqtSessionPeer::CreateRemoteTrack(&session_, DefaultSubscribe(), + /*track_alias=*/2, &visitor_); MoqtObject object = { /*track_alias=*/2, /*group_sequence=*/0, @@ -1110,7 +1113,8 @@ MockSubscribeRemoteTrackVisitor visitor_; FullTrackName ftn("foo", "bar"); std::string payload = "deadbeef"; - MoqtSessionPeer::CreateRemoteTrack(&session, DefaultSubscribe(), &visitor_); + MoqtSessionPeer::CreateRemoteTrack(&session, DefaultSubscribe(), + /*track_alias=*/2, &visitor_); MoqtObject object = { /*track_alias=*/2, /*group_sequence=*/0, @@ -1137,7 +1141,7 @@ FullTrackName ftn("foo", "bar"); std::string payload = "deadbeef"; MoqtSessionPeer::CreateRemoteTrack(&session_, DefaultLocalSubscribe(), - &visitor_); + std::nullopt, &visitor_); MoqtObject object = { /*track_alias=*/2, /*group_sequence=*/0, @@ -1151,22 +1155,13 @@ std::unique_ptr<MoqtDataParserVisitor> object_stream = MoqtSessionPeer::CreateIncomingDataStream(&session_, &mock_stream_, kDefaultSubgroupStreamType); - - EXPECT_CALL(visitor_, OnObjectFragment) - .WillOnce([&](const FullTrackName& full_track_name, - const PublishedObjectMetadata& metadata, - absl::string_view payload, bool end_of_message) { - EXPECT_EQ(full_track_name, ftn); - EXPECT_EQ(metadata.location.group, object.group_id); - EXPECT_EQ(metadata.location.object, object.object_id); - }); - EXPECT_CALL(mock_stream_, GetStreamId()) - .WillRepeatedly(Return(kIncomingUniStreamId)); + EXPECT_CALL(mock_stream_, SendStopSending); object_stream->OnObjectMessage(object, payload, true); // SUBSCRIBE_OK arrives MoqtSubscribeOk ok = { kDefaultLocalRequestId, + /*track_alias=*/2, /*expires=*/quic::QuicTimeDelta::FromMilliseconds(0), /*group_order=*/MoqtDeliveryOrder::kAscending, /*largest_location=*/std::nullopt, @@ -1178,97 +1173,33 @@ control_stream->OnSubscribeOkMessage(ok); } -TEST_F(MoqtSessionTest, ObjectBeforeSubscribeError) { +TEST_F(MoqtSessionTest, SubscribeOkWithBadTrackAlias) { MockSubscribeRemoteTrackVisitor visitor; - FullTrackName ftn("foo", "bar"); - std::string payload = "deadbeef"; + // Create open subscription. MoqtSessionPeer::CreateRemoteTrack(&session_, DefaultLocalSubscribe(), + /*track_alias=*/2, &visitor); + MoqtSubscribe subscribe2 = DefaultLocalSubscribe(); + subscribe2.request_id += 2; + subscribe2.full_track_name = FullTrackName("foo2", "bar2"); + MoqtSessionPeer::CreateRemoteTrack(&session_, subscribe2, std::nullopt, &visitor); - MoqtObject object = { + + // SUBSCRIBE_OK arrives + MoqtSubscribeOk subscribe_ok = { + subscribe2.request_id, /*track_alias=*/2, - /*group_sequence=*/0, - /*object_sequence=*/0, - /*publisher_priority=*/0, - /*extension_headers=*/"", - /*object_status=*/MoqtObjectStatus::kNormal, - /*subgroup_id=*/0, - /*payload_length=*/8, - }; - std::unique_ptr<MoqtDataParserVisitor> object_stream = - MoqtSessionPeer::CreateIncomingDataStream(&session_, &mock_stream_, - kDefaultSubgroupStreamType); - - EXPECT_CALL(visitor, OnObjectFragment) - .WillOnce([&](const FullTrackName& full_track_name, - const PublishedObjectMetadata& metadata, - absl::string_view payload, bool end_of_message) { - EXPECT_EQ(full_track_name, ftn); - EXPECT_EQ(metadata.location.group, object.group_id); - EXPECT_EQ(metadata.location.object, object.object_id); - }); - EXPECT_CALL(mock_stream_, GetStreamId()) - .WillRepeatedly(Return(kIncomingUniStreamId)); - object_stream->OnObjectMessage(object, payload, true); - - // SUBSCRIBE_ERROR arrives - MoqtSubscribeError subscribe_error = { - kDefaultLocalRequestId, - /*error_code=*/RequestErrorCode::kRetryTrackAlias, - /*reason_phrase=*/"foo", - /*track_alias =*/3, + /*expires=*/quic::QuicTimeDelta::FromMilliseconds(0), + /*group_order=*/MoqtDeliveryOrder::kAscending, + /*largest_location=*/std::nullopt, + VersionSpecificParameters(), }; webtransport::test::MockStream mock_control_stream; std::unique_ptr<MoqtControlParserVisitor> control_stream = MoqtSessionPeer::CreateControlStream(&session_, &mock_control_stream); EXPECT_CALL( mock_session_, - CloseSession(static_cast<uint64_t>(MoqtError::kProtocolViolation), - "Received SUBSCRIBE_ERROR after SUBSCRIBE_OK or objects")) - .Times(1); - control_stream->OnSubscribeErrorMessage(subscribe_error); -} - -TEST_F(MoqtSessionTest, SubscribeErrorWithTrackAlias) { - MockSubscribeRemoteTrackVisitor visitor; - MoqtSessionPeer::CreateRemoteTrack(&session_, DefaultLocalSubscribe(), - &visitor); - - // SUBSCRIBE_ERROR arrives - MoqtSubscribeError subscribe_error = { - kDefaultLocalRequestId, - /*error_code=*/RequestErrorCode::kRetryTrackAlias, - /*reason_phrase=*/"foo", - /*track_alias =*/3, - }; - webtransport::test::MockStream mock_control_stream; - std::unique_ptr<MoqtControlParserVisitor> control_stream = - MoqtSessionPeer::CreateControlStream(&session_, &mock_control_stream); - EXPECT_CALL(mock_control_stream, - Writev(ControlMessageOfType(MoqtMessageType::kSubscribe), _)) - .Times(1); - control_stream->OnSubscribeErrorMessage(subscribe_error); -} - -TEST_F(MoqtSessionTest, SubscribeErrorWithBadTrackAlias) { - MockSubscribeRemoteTrackVisitor visitor; - MoqtSessionPeer::CreateRemoteTrack(&session_, DefaultLocalSubscribe(), - &visitor); - - // SUBSCRIBE_ERROR arrives - MoqtSubscribeError subscribe_error = { - kDefaultLocalRequestId, - /*error_code=*/RequestErrorCode::kRetryTrackAlias, - /*reason_phrase=*/"foo", - /*track_alias =*/2, - }; - webtransport::test::MockStream mock_control_stream; - std::unique_ptr<MoqtControlParserVisitor> control_stream = - MoqtSessionPeer::CreateControlStream(&session_, &mock_control_stream); - EXPECT_CALL(mock_session_, - CloseSession(static_cast<uint64_t>(MoqtError::kProtocolViolation), - "Provided track alias already in use")) - .Times(1); - control_stream->OnSubscribeErrorMessage(subscribe_error); + CloseSession(static_cast<uint64_t>(MoqtError::kDuplicateTrackAlias), "")); + control_stream->OnSubscribeOkMessage(subscribe_ok); } TEST_F(MoqtSessionTest, CreateOutgoingDataStreamAndSend) { @@ -2013,7 +1944,8 @@ MockSubscribeRemoteTrackVisitor visitor_; FullTrackName ftn("foo", "bar"); std::string payload = "deadbeef"; - MoqtSessionPeer::CreateRemoteTrack(&session_, DefaultSubscribe(), &visitor_); + MoqtSessionPeer::CreateRemoteTrack(&session_, DefaultSubscribe(), + /*track_alias=*/2, &visitor_); MoqtObject object = { /*track_alias=*/2, /*group_sequence=*/0, @@ -2044,7 +1976,8 @@ TEST_F(MoqtSessionTest, DataStreamTypeMismatch) { MockSubscribeRemoteTrackVisitor visitor_; std::string payload = "deadbeef"; - MoqtSessionPeer::CreateRemoteTrack(&session_, DefaultSubscribe(), &visitor_); + MoqtSessionPeer::CreateRemoteTrack(&session_, DefaultSubscribe(), + /*track_alias=*/2, &visitor_); MoqtObject object = { /*track_alias=*/2, /*group_sequence=*/0, @@ -2078,7 +2011,8 @@ std::string payload = "deadbeef"; MoqtSubscribe subscribe = DefaultSubscribe(); subscribe.start = Location(1, 0); - MoqtSessionPeer::CreateRemoteTrack(&session_, subscribe, &visitor_); + MoqtSessionPeer::CreateRemoteTrack(&session_, subscribe, /*track_alias=*/2, + &visitor_); MoqtObject object = { /*track_alias=*/2, /*group_sequence=*/0, @@ -2101,7 +2035,8 @@ std::string payload = "deadbeef"; MoqtSubscribe subscribe = DefaultSubscribe(); subscribe.start = Location(1, 0); - MoqtSessionPeer::CreateRemoteTrack(&session_, subscribe, &visitor_); + MoqtSessionPeer::CreateRemoteTrack(&session_, subscribe, /*track_alias=*/2, + &visitor_); char datagram[] = {0x01, 0x02, 0x00, 0x00, 0x80, 0x00, 0x08, 0x64, 0x65, 0x61, 0x64, 0x62, 0x65, 0x65, 0x66}; EXPECT_CALL(visitor_, OnObjectFragment).Times(0); @@ -2655,7 +2590,6 @@ .WillRepeatedly(Return(&mock_stream_)); MoqtSubscribe expected_subscribe = { /*request_id=*/0, - /*track_alias=*/0, /*full_track_name=*/FullTrackName("foo", "bar"), /*subscriber_priority=*/0x80, /*group_order=*/MoqtDeliveryOrder::kAscending, @@ -2696,7 +2630,7 @@ EXPECT_CALL(remote_track_visitor, OnReply).Times(1); stream_input->OnSubscribeOkMessage( - MoqtSubscribeOk(0, quic::QuicTimeDelta::FromMilliseconds(0), + MoqtSubscribeOk(0, 2, quic::QuicTimeDelta::FromMilliseconds(0), MoqtDeliveryOrder::kAscending, Location(2, 0), VersionSpecificParameters())); stream_input->OnFetchOkMessage(MoqtFetchOk(2, MoqtDeliveryOrder::kAscending, @@ -3452,6 +3386,7 @@ VersionSpecificParameters())); MoqtSubscribeOk ok = { /*request_id=*/0, + /*track_alias=*/0, /*expires=*/quic::QuicTimeDelta::FromMilliseconds(10000), /*group_order=*/MoqtDeliveryOrder::kAscending, /*largest_location=*/std::nullopt, @@ -3510,6 +3445,7 @@ VersionSpecificParameters())); MoqtSubscribeOk ok = { /*request_id=*/0, + /*track_alias=*/0, /*expires=*/quic::QuicTimeDelta::FromMilliseconds(10000), /*group_order=*/MoqtDeliveryOrder::kAscending, /*largest_location=*/std::nullopt, @@ -3565,6 +3501,7 @@ VersionSpecificParameters())); MoqtSubscribeOk ok = { /*request_id=*/0, + /*track_alias=*/0, /*expires=*/quic::QuicTimeDelta::FromMilliseconds(10000), /*group_order=*/MoqtDeliveryOrder::kAscending, /*largest_location=*/std::nullopt, @@ -3616,7 +3553,7 @@ TEST_F(MoqtSessionTest, SubgroupStreamOutOfOrder) { MockSubscribeRemoteTrackVisitor remote_track_visitor; MoqtSessionPeer::CreateRemoteTrack(&session_, DefaultSubscribe(), - &remote_track_visitor); + /*track_alias=*/2, &remote_track_visitor); webtransport::test::MockStream control_stream; std::unique_ptr<MoqtControlParserVisitor> stream_input = MoqtSessionPeer::CreateControlStream(&session_, &control_stream); @@ -3647,7 +3584,7 @@ TEST_F(MoqtSessionTest, SubgroupStreamObjectAfterGroupEnd) { MockSubscribeRemoteTrackVisitor remote_track_visitor; MoqtSessionPeer::CreateRemoteTrack(&session_, DefaultSubscribe(), - &remote_track_visitor); + /*track_alias=*/2, &remote_track_visitor); webtransport::test::MockStream control_stream; std::unique_ptr<MoqtControlParserVisitor> stream_input = MoqtSessionPeer::CreateControlStream(&session_, &control_stream); @@ -3678,7 +3615,7 @@ TEST_F(MoqtSessionTest, SubgroupStreamObjectAfterTrackEnd) { MockSubscribeRemoteTrackVisitor remote_track_visitor; MoqtSessionPeer::CreateRemoteTrack(&session_, DefaultSubscribe(), - &remote_track_visitor); + /*track_alias=*/2, &remote_track_visitor); webtransport::test::MockStream control_stream; std::unique_ptr<MoqtControlParserVisitor> stream_input = MoqtSessionPeer::CreateControlStream(&session_, &control_stream);
diff --git a/quiche/quic/moqt/moqt_track.h b/quiche/quic/moqt/moqt_track.h index be3dece..d2bfc18 100644 --- a/quiche/quic/moqt/moqt_track.h +++ b/quiche/quic/moqt/moqt_track.h
@@ -124,11 +124,9 @@ SubscribeWindow(subscribe.start.value_or(Location()), subscribe.end_group), subscribe.subscriber_priority), - track_alias_(subscribe.track_alias), forward_(subscribe.forward), visitor_(visitor), - delivery_timeout_(subscribe.parameters.delivery_timeout), - subscribe_(std::make_unique<MoqtSubscribe>(subscribe)) {} + delivery_timeout_(subscribe.parameters.delivery_timeout) {} ~SubscribeRemoteTrack() override { if (subscribe_done_alarm_ != nullptr) { subscribe_done_alarm_->PermanentCancel(); @@ -136,16 +134,14 @@ } void OnObjectOrOk() override { - subscribe_.reset(); // No SUBSCRIBE_ERROR, no need to store this anymore. RemoteTrack::OnObjectOrOk(); } - uint64_t track_alias() const { return track_alias_; } - Visitor* visitor() { return visitor_; } - MoqtSubscribe& GetSubscribe() { - return *subscribe_; - // This class will soon be destroyed, so there's no need to null the - // unique_ptr; + std::optional<uint64_t> track_alias() const { return track_alias_; } + void set_track_alias(uint64_t track_alias) { + track_alias_.emplace(track_alias); } + Visitor* visitor() { return visitor_; } + // Returns false if the forwarding preference is changing on the track. bool OnObject(bool is_datagram) { OnObjectOrOk(); @@ -191,7 +187,7 @@ void FetchObjects(); std::unique_ptr<MoqtFetchTask> fetch_task_; - const uint64_t track_alias_; + std::optional<const uint64_t> track_alias_; bool forward_; Visitor* visitor_; std::optional<bool> is_datagram_; @@ -205,10 +201,6 @@ quic::QuicTimeDelta delivery_timeout_ = quic::QuicTimeDelta::Infinite(); std::unique_ptr<quic::QuicAlarm> subscribe_done_alarm_ = nullptr; const quic::QuicClock* clock_ = nullptr; - - // For convenience, store the subscribe message if it has to be re-sent with - // a new track alias. - std::unique_ptr<MoqtSubscribe> subscribe_; }; // MoqtSession calls this when a FETCH_OK or FETCH_ERROR is received. The
diff --git a/quiche/quic/moqt/moqt_track_test.cc b/quiche/quic/moqt/moqt_track_test.cc index 9f3956c..d8fa4a2 100644 --- a/quiche/quic/moqt/moqt_track_test.cc +++ b/quiche/quic/moqt/moqt_track_test.cc
@@ -46,7 +46,6 @@ MockSubscribeRemoteTrackVisitor visitor_; MoqtSubscribe subscribe_ = { /*subscribe_id=*/1, - /*track_alias=*/2, /*full_track_name=*/FullTrackName("foo", "bar"), /*subscriber_priority=*/128, /*group_order=*/std::nullopt, @@ -62,9 +61,11 @@ TEST_F(SubscribeRemoteTrackTest, Queries) { EXPECT_EQ(track_.full_track_name(), FullTrackName("foo", "bar")); EXPECT_EQ(track_.request_id(), 1); - EXPECT_EQ(track_.track_alias(), 2); + EXPECT_FALSE(track_.track_alias().has_value()); EXPECT_EQ(track_.visitor(), &visitor_); EXPECT_FALSE(track_.is_fetch()); + track_.set_track_alias(1); + EXPECT_EQ(track_.track_alias(), 1); } TEST_F(SubscribeRemoteTrackTest, UpdateDataStreamType) { @@ -75,7 +76,6 @@ TEST_F(SubscribeRemoteTrackTest, AllowError) { EXPECT_TRUE(track_.ErrorIsAllowed()); - EXPECT_EQ(track_.GetSubscribe().request_id, subscribe_.request_id); track_.OnObjectOrOk(); EXPECT_FALSE(track_.ErrorIsAllowed()); }
diff --git a/quiche/quic/moqt/test_tools/moqt_session_peer.h b/quiche/quic/moqt/test_tools/moqt_session_peer.h index f411c82..4064ed3 100644 --- a/quiche/quic/moqt/test_tools/moqt_session_peer.h +++ b/quiche/quic/moqt/test_tools/moqt_session_peer.h
@@ -86,10 +86,13 @@ static void CreateRemoteTrack(MoqtSession* session, const MoqtSubscribe& subscribe, + const std::optional<uint64_t> track_alias, SubscribeRemoteTrack::Visitor* visitor) { auto track = std::make_unique<SubscribeRemoteTrack>(subscribe, visitor); - session->subscribe_by_alias_.try_emplace(subscribe.track_alias, - track.get()); + if (track_alias.has_value()) { + track->set_track_alias(*track_alias); + session->subscribe_by_alias_.try_emplace(*track_alias, track.get()); + } session->subscribe_by_name_.try_emplace(subscribe.full_track_name, track.get()); session->upstream_by_id_.try_emplace(subscribe.request_id, @@ -102,7 +105,6 @@ uint64_t start_object) { MoqtSubscribe subscribe; subscribe.full_track_name = publisher->GetTrackName(); - subscribe.track_alias = track_alias; subscribe.request_id = subscribe_id; subscribe.forward = true; subscribe.filter_type = MoqtFilterType::kAbsoluteStart; @@ -112,6 +114,8 @@ subscribe_id, std::make_unique<MoqtSession::PublishedSubscription>( session, std::move(publisher), subscribe, /*monitoring_interface=*/nullptr)); + session->published_subscriptions_[subscribe_id]->track_alias_.emplace( + track_alias); return session->published_subscriptions_[subscribe_id].get(); }
diff --git a/quiche/quic/moqt/test_tools/moqt_test_message.h b/quiche/quic/moqt/test_tools/moqt_test_message.h index cf5fdfd..8c3bd33 100644 --- a/quiche/quic/moqt/test_tools/moqt_test_message.h +++ b/quiche/quic/moqt/test_tools/moqt_test_message.h
@@ -570,10 +570,6 @@ QUIC_LOG(INFO) << "SUBSCRIBE subscribe ID mismatch"; return false; } - if (cast.track_alias != subscribe_.track_alias) { - QUIC_LOG(INFO) << "SUBSCRIBE track alias mismatch"; - return false; - } if (cast.full_track_name != subscribe_.full_track_name) { QUIC_LOG(INFO) << "SUBSCRIBE track name mismatch"; return false; @@ -610,7 +606,7 @@ } void ExpandVarints() override { - ExpandVarintsImpl("vvvv---v-------vvvvv--vv-----"); + ExpandVarintsImpl("vvv---v-------vvvvv--vv-----"); } MessageStructuredData structured_data() const override { @@ -618,25 +614,43 @@ } private: - uint8_t raw_packet_[32] = { - 0x03, 0x00, 0x1d, 0x01, 0x02, // id and alias - 0x01, 0x03, 0x66, 0x6f, 0x6f, // track_namespace = "foo" - 0x04, 0x61, 0x62, 0x63, 0x64, // track_name = "abcd" - 0x20, // subscriber priority = 0x20 - 0x02, // group order = descending - 0x01, // forward = true - 0x03, // Filter type: Absolute Start - 0x04, // start_group = 4 - 0x01, // start_object = 1 + uint8_t raw_packet_[31] = { + 0x03, + 0x00, + 0x1c, + 0x01, // request_id = 1 + 0x01, + 0x03, + 0x66, + 0x6f, + 0x6f, // track_namespace = "foo" + 0x04, + 0x61, + 0x62, + 0x63, + 0x64, // track_name = "abcd" + 0x20, // subscriber priority = 0x20 + 0x02, // group order = descending + 0x01, // forward = true + 0x03, // Filter type: Absolute Start + 0x04, // start_group = 4 + 0x01, // start_object = 1 // No EndGroup or EndObject - 0x02, // 2 parameters - 0x02, 0x67, 0x10, // delivery_timeout = 10000 ms - 0x03, 0x05, 0x03, 0x00, 0x62, 0x61, 0x72, // authorization_tag = "bar" + 0x02, // 2 parameters + 0x02, + 0x67, + 0x10, // delivery_timeout = 10000 ms + 0x03, + 0x05, + 0x03, + 0x00, + 0x62, + 0x61, + 0x72, // authorization_tag = "bar" }; MoqtSubscribe subscribe_ = { - /*subscribe_id=*/1, - /*track_alias=*/2, + /*request_id=*/1, FullTrackName("foo", "abcd"), /*subscriber_priority=*/0x20, /*group_order=*/MoqtDeliveryOrder::kDescending, @@ -661,6 +675,10 @@ QUIC_LOG(INFO) << "SUBSCRIBE OK subscribe ID mismatch"; return false; } + if (cast.track_alias != subscribe_ok_.track_alias) { + QUIC_LOG(INFO) << "SUBSCRIBE OK track alias mismatch"; + return false; + } if (cast.expires != subscribe_ok_.expires) { QUIC_LOG(INFO) << "SUBSCRIBE OK expiration mismatch"; return false; @@ -680,34 +698,35 @@ return true; } - void ExpandVarints() override { ExpandVarintsImpl("vv--vvvv--v--"); } + void ExpandVarints() override { ExpandVarintsImpl("vvv--vvvv--v--"); } MessageStructuredData structured_data() const override { return TestMessageBase::MessageStructuredData(subscribe_ok_); } void SetInvalidContentExists() { - raw_packet_[6] = 0x02; + raw_packet_[7] = 0x02; SetWireImage(raw_packet_, sizeof(raw_packet_)); } void SetInvalidDeliveryOrder() { - raw_packet_[5] = 0x10; + raw_packet_[6] = 0x10; SetWireImage(raw_packet_, sizeof(raw_packet_)); } private: - uint8_t raw_packet_[16] = { - 0x04, 0x00, 0x0d, 0x01, 0x03, // request_id = 1, expires = 3 - 0x02, 0x01, // group_order = 2, content exists - 0x0c, 0x14, // largest_location = (12, 20) - 0x02, // 2 parameters - 0x02, 0x67, 0x10, // delivery_timeout = 10000 - 0x04, 0x67, 0x10, // max_cache_duration = 10000 + uint8_t raw_packet_[17] = { + 0x04, 0x00, 0x0e, 0x01, 0x02, 0x03, // request_id, alias, expires + 0x02, 0x01, // group_order = 2, content exists + 0x0c, 0x14, // largest_location = (12, 20) + 0x02, // 2 parameters + 0x02, 0x67, 0x10, // delivery_timeout = 10000 + 0x04, 0x67, 0x10, // max_cache_duration = 10000 }; MoqtSubscribeOk subscribe_ok_ = { /*request_id=*/1, + /*track_alias=*/2, /*expires=*/quic::QuicTimeDelta::FromMilliseconds(3), /*group_order=*/MoqtDeliveryOrder::kDescending, /*largest_location=*/Location(12, 20), @@ -736,33 +755,27 @@ QUIC_LOG(INFO) << "SUBSCRIBE ERROR reason phrase mismatch"; return false; } - if (cast.track_alias != subscribe_error_.track_alias) { - QUIC_LOG(INFO) << "SUBSCRIBE ERROR track alias mismatch"; - return false; - } return true; } - void ExpandVarints() override { ExpandVarintsImpl("vvv---v"); } + void ExpandVarints() override { ExpandVarintsImpl("vvv---"); } MessageStructuredData structured_data() const override { return TestMessageBase::MessageStructuredData(subscribe_error_); } private: - uint8_t raw_packet_[10] = { - 0x05, 0x00, 0x07, + uint8_t raw_packet_[9] = { + 0x05, 0x00, 0x06, 0x02, // request_id = 2 0x05, // error_code = 5 0x03, 0x62, 0x61, 0x72, // reason_phrase = "bar" - 0x04, // track_alias = 4 }; MoqtSubscribeError subscribe_error_ = { /*request_id=*/2, /*error_code=*/RequestErrorCode::kInvalidRange, /*reason_phrase=*/"bar", - /*track_alias=*/4, }; };