Skip to content

Commit 7bebe08

Browse files
mondainmondain
committed
Fix MoQT pass 3 compliance gaps
1 parent 96baf5a commit 7bebe08

4 files changed

Lines changed: 203 additions & 12 deletions

File tree

‎src/transport/moqt_control_messages.cpp‎

Lines changed: 12 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@ constexpr std::uint64_t kPublishNamespaceOkType = 0x07;
2323
constexpr std::uint64_t kRequestOkType = 0x07;
2424
constexpr std::uint64_t kPublishNamespaceErrorType = 0x08;
2525
constexpr std::uint64_t kPublishNamespaceDoneType = 0x09;
26+
constexpr std::uint64_t kNamespaceDoneType = 0x0e;
2627
constexpr std::uint64_t kPublishDoneType = 0x0b;
2728
constexpr std::uint64_t kSubscribeNamespaceType = 0x11;
2829
constexpr std::uint64_t kSubscribeNamespaceOkType = 0x12;
@@ -464,6 +465,7 @@ bool next_control_message(std::span<const std::uint8_t> bytes, DraftVersion draf
464465
case kPublishNamespaceOkType:
465466
case kPublishNamespaceErrorType:
466467
case kPublishNamespaceDoneType:
468+
case kNamespaceDoneType:
467469
case kPublishDoneType:
468470
case kMaxRequestIdType:
469471
case kSubscribeNamespaceTypeDraft18:
@@ -815,7 +817,7 @@ bool decode_request_ok(std::span<const std::uint8_t> bytes, DraftVersion draft,
815817
std::uint64_t previous_parameter_type = 0;
816818
for (std::uint64_t parameter_index = 0; parameter_index < parameter_count; ++parameter_index) {
817819
std::uint64_t parameter_type = 0;
818-
if (!decode_parameter_type(bytes, offset, draft, previous_parameter_type, draft == DraftVersion::kDraft16, parameter_type)) {
820+
if (!decode_parameter_type(bytes, offset, draft, previous_parameter_type, true, parameter_type)) {
819821
return false;
820822
}
821823
if ((parameter_type & 0x1ULL) == 0) {
@@ -940,7 +942,7 @@ bool decode_subscribe_namespace_message(std::span<const std::uint8_t> bytes,
940942
std::uint64_t previous_parameter_type = 0;
941943
for (std::uint64_t index = 0; index < parameters; ++index) {
942944
std::uint64_t parameter_type = 0;
943-
if (!decode_parameter_type(bytes, offset, draft, previous_parameter_type, draft == DraftVersion::kDraft16, parameter_type)) {
945+
if (!decode_parameter_type(bytes, offset, draft, previous_parameter_type, true, parameter_type)) {
944946
return false;
945947
}
946948
if ((parameter_type & 0x1ULL) == 0) {
@@ -1053,7 +1055,7 @@ bool decode_subscribe_message(std::span<const std::uint8_t> bytes, DraftVersion
10531055
message.subscriber_priority = bytes[offset++];
10541056
message.group_order = bytes[offset++];
10551057
message.forward = bytes[offset++];
1056-
if (message.group_order > 2 || message.forward > 1) {
1058+
if (message.group_order == 0 || message.group_order > 2 || message.forward > 1) {
10571059
return false;
10581060
}
10591061

@@ -1404,11 +1406,13 @@ std::vector<std::uint8_t> encode_publish_namespace_done_message(const NamespaceM
14041406
} else if (!uses_moq_vi64(message.draft)) {
14051407
append_moqint(payload, message.draft, message.request_id);
14061408
} else {
1407-
return {};
1409+
append_track_namespace(payload, message.draft, message.track_namespace);
14081410
}
14091411

14101412
std::vector<std::uint8_t> message_bytes;
1411-
append_moqint(message_bytes, message.draft, kPublishNamespaceDoneType);
1413+
append_moqint(message_bytes,
1414+
message.draft,
1415+
uses_moq_vi64(message.draft) ? kNamespaceDoneType : kPublishNamespaceDoneType);
14121416
append_uint16(message_bytes, static_cast<std::uint16_t>(payload.size()));
14131417
message_bytes.insert(message_bytes.end(), payload.begin(), payload.end());
14141418
return message_bytes;
@@ -1500,6 +1504,9 @@ bool decode_publish_ok(std::span<const std::uint8_t> bytes, DraftVersion draft,
15001504
message.forward = bytes[offset++];
15011505
message.subscriber_priority = bytes[offset++];
15021506
message.group_order = bytes[offset++];
1507+
if (message.group_order == 0 || message.group_order > 2 || message.forward > 1) {
1508+
return false;
1509+
}
15031510
if (!decode_moqint_impl(bytes, offset, draft, message.filter_type)) {
15041511
return false;
15051512
}

‎src/transport/moqt_session.cpp‎

Lines changed: 56 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -914,7 +914,14 @@ TransportStatus send_request_stream_and_wait(PublisherTransport& transport,
914914
const bool is_goaway = response_type == 0x10;
915915

916916
if (is_goaway) {
917-
return protocol_violation(transport, "request stream received GOAWAY");
917+
if (draft != openmoq::publisher::DraftVersion::kDraft18) {
918+
return protocol_violation(transport, "request stream received GOAWAY");
919+
}
920+
const TransportStatus reset_status = transport.reset_stream(request_stream_id, 0x0);
921+
if (!reset_status.ok) {
922+
return reset_status;
923+
}
924+
return TransportStatus::failure("request stream received GOAWAY migration");
918925
}
919926

920927
bool response_type_allowed = false;
@@ -1180,6 +1187,23 @@ bool decode_legacy_subscribe_update_message(std::span<const std::uint8_t> bytes,
11801187
return false;
11811188
}
11821189

1190+
bool decode_unsubscribe_message(std::span<const std::uint8_t> bytes,
1191+
openmoq::publisher::DraftVersion draft,
1192+
std::uint64_t& request_id) {
1193+
std::size_t offset = 0;
1194+
std::uint64_t message_type = 0;
1195+
if (!decode_moqint(bytes, offset, draft, message_type) || message_type != 0x0a || offset + 2 > bytes.size()) {
1196+
return false;
1197+
}
1198+
const std::size_t payload_length =
1199+
(static_cast<std::size_t>(bytes[offset]) << 8) | static_cast<std::size_t>(bytes[offset + 1]);
1200+
offset += 2;
1201+
const std::size_t payload_end = offset + payload_length;
1202+
return payload_end == bytes.size() &&
1203+
decode_moqint(bytes, offset, draft, request_id) &&
1204+
offset == payload_end;
1205+
}
1206+
11831207
bool find_next_matching_object_index(const openmoq::publisher::PublishPlan& plan,
11841208
const SubscribeMessage& subscribe,
11851209
std::size_t start_index,
@@ -1691,6 +1715,37 @@ TransportStatus serve_subscriptions(PublisherTransport& transport,
16911715
message_type == 0x16 || message_type == 0x1d || message_type == 0x51)) {
16921716
return protocol_violation(transport, "draft-18 request message received on control stream");
16931717
}
1718+
if (message_type == 0x0a) {
1719+
if (uses_request_streams(draft)) {
1720+
return protocol_violation(transport, "UNSUBSCRIBE received on draft-18 control stream");
1721+
}
1722+
std::uint64_t unsubscribe_request_id = 0;
1723+
if (!decode_unsubscribe_message(message_bytes, draft, unsubscribe_request_id)) {
1724+
return protocol_violation(transport, "received invalid UNSUBSCRIBE");
1725+
}
1726+
pending_subscriptions.erase(unsubscribe_request_id);
1727+
auto active_it = active_subscriptions.find(unsubscribe_request_id);
1728+
if (active_it != active_subscriptions.end()) {
1729+
const TransportStatus finish_status = active_it->second.sender->finish_group(transport);
1730+
if (!finish_status.ok) {
1731+
return finish_status;
1732+
}
1733+
const TransportStatus finalize_status =
1734+
finalize_subscription(transport,
1735+
draft,
1736+
control_stream_id,
1737+
unsubscribe_request_id,
1738+
active_it->second.sender->stream_count(),
1739+
completed_request_ids);
1740+
if (!finalize_status.ok) {
1741+
return finalize_status;
1742+
}
1743+
active_subscriptions.erase(active_it);
1744+
served_any_subscription = true;
1745+
}
1746+
buffer.erase(buffer.begin(), buffer.begin() + message_size);
1747+
continue;
1748+
}
16941749
const bool is_handled_type =
16951750
message_type == 0x02 || // SUBSCRIBE_UPDATE
16961751
(!uses_request_streams(draft) &&

‎tests/moqt_control_messages_test.cpp‎

Lines changed: 67 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -630,8 +630,13 @@ bool test_publisher_control_message_encoders_for_all_drafts() {
630630

631631
ok &= expect_uint16_frame(encode_publish_done_message(draft, 44, 2), 0x0b, frame, label + " publish done");
632632
if (uses_vi64(draft)) {
633-
ok &= expect(encode_publish_namespace_done_message(namespace_message).empty(),
634-
label + " publish namespace done is not emitted");
633+
const auto namespace_done = encode_publish_namespace_done_message(namespace_message);
634+
ok &= expect_uint16_frame(namespace_done, 0x0e, frame, label + " namespace done");
635+
offset = frame.payload_offset;
636+
ok &= expect(read_track_namespace(namespace_done, offset, frame.payload_end, draft, tuple) &&
637+
tuple == std::vector<std::string>({"live", "alpha"}),
638+
label + " namespace done suffix placement");
639+
ok &= expect(offset == frame.payload_end, label + " namespace done boundary");
635640
} else {
636641
ok &= expect_uint16_frame(encode_publish_namespace_done_message(namespace_message), 0x09, frame,
637642
label + " publish namespace done");
@@ -814,6 +819,30 @@ bool test_control_message_framing_and_parameter_regressions() {
814819
ok &= expect(!decode_subscribe_message(duplicate_parameter_subscribe, DraftVersion::kDraft16, subscribe),
815820
"draft-16 rejects duplicate delta-encoded parameter type");
816821

822+
std::vector<std::uint8_t> draft14_bad_group_order_subscribe =
823+
build_subscribe_message(DraftVersion::kDraft14);
824+
for (std::size_t index = 0; index + 3 < draft14_bad_group_order_subscribe.size(); ++index) {
825+
if (draft14_bad_group_order_subscribe[index] == 128 &&
826+
draft14_bad_group_order_subscribe[index + 1] == 1 &&
827+
draft14_bad_group_order_subscribe[index + 2] == 1 &&
828+
draft14_bad_group_order_subscribe[index + 3] == 3) {
829+
draft14_bad_group_order_subscribe[index + 1] = 0;
830+
break;
831+
}
832+
}
833+
ok &= expect(!decode_subscribe_message(
834+
draft14_bad_group_order_subscribe, DraftVersion::kDraft14, subscribe),
835+
"draft-14 SUBSCRIBE rejects group order 0");
836+
837+
std::vector<std::uint8_t> draft14_bad_group_order_publish_ok =
838+
build_publish_ok_message(DraftVersion::kDraft14);
839+
if (draft14_bad_group_order_publish_ok.size() > 5) {
840+
draft14_bad_group_order_publish_ok[5] = 0;
841+
}
842+
PublishOk publish_ok;
843+
ok &= expect(!decode_publish_ok(draft14_bad_group_order_publish_ok, DraftVersion::kDraft14, publish_ok),
844+
"draft-14 PUBLISH_OK rejects group order 0");
845+
817846
std::vector<std::uint8_t> request_ok_payload;
818847
append_varint(request_ok_payload, 44);
819848
append_varint(request_ok_payload, 1);
@@ -826,6 +855,42 @@ bool test_control_message_framing_and_parameter_regressions() {
826855
"draft-16 REQUEST_OK decodes delta-encoded parameters");
827856
ok &= expect(namespace_ok.request_id == 44, "draft-16 REQUEST_OK request id with parameters");
828857

858+
std::vector<std::uint8_t> draft18_request_ok_payload;
859+
append_moqint(draft18_request_ok_payload, DraftVersion::kDraft18, 2);
860+
append_moqint(draft18_request_ok_payload, DraftVersion::kDraft18, 0x21);
861+
append_moqint(draft18_request_ok_payload, DraftVersion::kDraft18, 0);
862+
append_moqint(draft18_request_ok_payload, DraftVersion::kDraft18, 1);
863+
append_moqint(draft18_request_ok_payload, DraftVersion::kDraft18, 7);
864+
std::vector<std::uint8_t> draft18_request_ok;
865+
append_moqint(draft18_request_ok, DraftVersion::kDraft18, 0x07);
866+
draft18_request_ok.push_back(static_cast<std::uint8_t>((draft18_request_ok_payload.size() >> 8) & 0xff));
867+
draft18_request_ok.push_back(static_cast<std::uint8_t>(draft18_request_ok_payload.size() & 0xff));
868+
draft18_request_ok.insert(
869+
draft18_request_ok.end(), draft18_request_ok_payload.begin(), draft18_request_ok_payload.end());
870+
ok &= expect(decode_request_ok(draft18_request_ok, DraftVersion::kDraft18, namespace_ok),
871+
"draft-18 REQUEST_OK decodes multi-parameter delta KVPs");
872+
873+
std::vector<std::uint8_t> draft18_subscribe_namespace_payload;
874+
append_moqint(draft18_subscribe_namespace_payload, DraftVersion::kDraft18, 2);
875+
append_track_namespace(draft18_subscribe_namespace_payload, DraftVersion::kDraft18, {"live"});
876+
append_moqint(draft18_subscribe_namespace_payload, DraftVersion::kDraft18, 2);
877+
append_moqint(draft18_subscribe_namespace_payload, DraftVersion::kDraft18, 0x21);
878+
append_moqint(draft18_subscribe_namespace_payload, DraftVersion::kDraft18, 0);
879+
append_moqint(draft18_subscribe_namespace_payload, DraftVersion::kDraft18, 1);
880+
append_moqint(draft18_subscribe_namespace_payload, DraftVersion::kDraft18, 7);
881+
std::vector<std::uint8_t> draft18_subscribe_namespace;
882+
append_moqint(draft18_subscribe_namespace, DraftVersion::kDraft18, 0x50);
883+
draft18_subscribe_namespace.push_back(
884+
static_cast<std::uint8_t>((draft18_subscribe_namespace_payload.size() >> 8) & 0xff));
885+
draft18_subscribe_namespace.push_back(static_cast<std::uint8_t>(draft18_subscribe_namespace_payload.size() & 0xff));
886+
draft18_subscribe_namespace.insert(draft18_subscribe_namespace.end(),
887+
draft18_subscribe_namespace_payload.begin(),
888+
draft18_subscribe_namespace_payload.end());
889+
SubscribeNamespaceMessage subscribe_namespace;
890+
ok &= expect(decode_subscribe_namespace_message(
891+
draft18_subscribe_namespace, DraftVersion::kDraft18, subscribe_namespace),
892+
"draft-18 SUBSCRIBE_NAMESPACE decodes multi-parameter delta KVPs");
893+
829894
const std::vector<std::uint8_t> draft18_namespace_ok =
830895
encode_subscribe_namespace_ok_message(DraftVersion::kDraft18, 44);
831896
Uint16Frame frame;

‎tests/moqt_session_test.cpp‎

Lines changed: 68 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -243,7 +243,7 @@ struct MockTransport final : PublisherTransport {
243243
std::vector<std::chrono::milliseconds> read_timeouts;
244244
std::map<std::uint64_t, std::vector<std::vector<std::uint8_t>>> reads;
245245
std::set<std::uint64_t> accepted_streams;
246-
std::function<void(const MockTransport&, std::uint64_t)> on_read;
246+
std::function<void(MockTransport&, std::uint64_t)> on_read;
247247
};
248248

249249
std::vector<std::size_t> object_write_indices(const MockTransport& transport) {
@@ -484,6 +484,14 @@ std::vector<std::uint8_t> encode_subscribe_update_message(std::uint64_t request_
484484
return message;
485485
}
486486

487+
std::vector<std::uint8_t> encode_unsubscribe_message(DraftVersion draft, std::uint64_t request_id) {
488+
std::vector<std::uint8_t> payload = encode_moqint(draft, request_id);
489+
std::vector<std::uint8_t> message = encode_moqint(draft, 0x0a);
490+
append_be16(message, static_cast<std::uint16_t>(payload.size()));
491+
message.insert(message.end(), payload.begin(), payload.end());
492+
return message;
493+
}
494+
487495
std::vector<std::uint8_t> encode_legacy_subscribe_update_message(std::uint64_t track_alias) {
488496
std::vector<std::uint8_t> payload = encode_varint(track_alias);
489497
const std::vector<std::uint8_t> start_group = encode_varint(0);
@@ -963,6 +971,57 @@ int main() {
963971
"expected default subscriber wait timeout to be 30 seconds");
964972
}
965973

974+
{
975+
MockTransport transport;
976+
transport.reads[0].push_back(encode_server_setup_message({
977+
.draft = DraftVersion::kDraft14,
978+
.max_request_id = 8,
979+
}));
980+
transport.reads[0].push_back(encode_publish_namespace_ok_message(DraftVersion::kDraft14, 0));
981+
transport.reads[0].push_back(encode_subscribe_message(1, kTestTrackNamespace, "vide_1", 1));
982+
983+
bool injected_unsubscribe = false;
984+
transport.on_read = [&](MockTransport& current, std::uint64_t stream_id) {
985+
if (stream_id != 0 || injected_unsubscribe) {
986+
return;
987+
}
988+
std::size_t object_payload_writes = 0;
989+
for (const auto& write : current.writes) {
990+
if (write.stream_id != 0 && !write.bytes.empty()) {
991+
++object_payload_writes;
992+
}
993+
}
994+
if (object_payload_writes == 1) {
995+
current.reads[0].push_back(encode_unsubscribe_message(DraftVersion::kDraft14, 1));
996+
injected_unsubscribe = true;
997+
}
998+
};
999+
1000+
MoqtSession session(transport, std::string(kTestTrackNamespace), false);
1001+
auto status = session.connect(endpoint, tls);
1002+
ok &= expect(status.ok, "expected unsubscribe session connect to succeed");
1003+
1004+
status = session.publish(make_multi_object_subgroup_plan());
1005+
ok &= expect(status.ok, "expected UNSUBSCRIBE flow to finish cleanly");
1006+
ok &= expect(injected_unsubscribe, "expected test to inject UNSUBSCRIBE after first object");
1007+
std::size_t object_payload_writes = 0;
1008+
bool saw_stream_fin = false;
1009+
for (const auto& write : transport.writes) {
1010+
if (write.stream_id == 0) {
1011+
continue;
1012+
}
1013+
if (write.bytes.empty() && write.fin) {
1014+
saw_stream_fin = true;
1015+
} else if (!write.bytes.empty()) {
1016+
++object_payload_writes;
1017+
}
1018+
}
1019+
ok &= expect(object_payload_writes == 1, "expected UNSUBSCRIBE to stop further object writes");
1020+
ok &= expect(saw_stream_fin, "expected UNSUBSCRIBE to FIN the open data stream");
1021+
ok &= expect(control_message_count(transport, 0x0b) == 1,
1022+
"expected UNSUBSCRIBE to emit PUBLISH_DONE for active subscription");
1023+
}
1024+
9661025
{
9671026
MockTransport transport;
9681027
transport.reads[0].push_back(encode_server_setup_message({
@@ -2121,9 +2180,14 @@ int main() {
21212180
const PublishPlan draft18_materialized =
21222181
materialize_publish_plan(make_span_backed_plan(DraftVersion::kDraft18), source_bytes);
21232182
status = draft18_goaway_session.publish(draft18_materialized);
2124-
ok &= expect(!status.ok, "expected draft-18 publish to fail on request-stream GOAWAY");
2125-
ok &= expect(status.message == "request stream received GOAWAY",
2126-
"expected explicit GOAWAY request-stream failure message");
2183+
ok &= expect(!status.ok, "expected draft-18 publish to surface request-stream GOAWAY migration");
2184+
ok &= expect(status.message == "request stream received GOAWAY migration",
2185+
"expected explicit GOAWAY request-stream migration message");
2186+
ok &= expect(draft18_goaway_transport.state() == ConnectionState::kConnected,
2187+
"expected request-stream GOAWAY not to close draft-18 session");
2188+
ok &= expect(!draft18_goaway_transport.reset_calls.empty() &&
2189+
draft18_goaway_transport.reset_calls.back().first == 0,
2190+
"expected request-stream GOAWAY to reset the old request stream");
21272191
}
21282192

21292193
{

0 commit comments

Comments
 (0)