Skip to content

Add handling of unknown control message, avoid blocking #4

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mondain merged 4 commits into from
Apr 14, 2026
Merged

Add handling of unknown control message, avoid blocking #4

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
6a8fe17
Add handling of unknown control message, to avoid FETCH blocking mess…
pchave-akamai Apr 13, 2026
4ae49d9
Update src/transport/moqt_session.cpp
mondain Apr 13, 2026
6a8c0eb
made explict list of unhandled messages to avoid mishandling truly un…
pchave-akamai Apr 13, 2026
ac5a9b1
Update src/transport/moqt_session.cpp
mondain Apr 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
45 changes: 21 additions & 24 deletions src/transport/moqt_control_messages.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -297,6 +297,7 @@ bool next_control_message(std::span<const std::uint8_t> bytes, std::size_t& mess
}

switch (type) {
// uint16 payload length: type varint + uint16 length + payload
case kClientSetupType:
case kServerSetupType:
case kSubscribeUpdateType:
Expand All @@ -307,7 +308,19 @@ bool next_control_message(std::span<const std::uint8_t> bytes, std::size_t& mess
case kPublishNamespaceOkType:
case kPublishNamespaceErrorType:
case kPublishNamespaceDoneType:
case kPublishDoneType: {
case kPublishDoneType:
case kMaxRequestIdType:
case kPublishType:
case kPublishOkType:
case kPublishErrorType:
case 0x0a: // UNSUBSCRIBE
case 0x10: // GOAWAY
case 0x14: // SUBSCRIBE_DONE (draft-14)
case 0x16: // FETCH
case 0x17: // FETCH_OK
case 0x18: // FETCH_CANCEL
case 0x1a: // REQUESTS_BLOCKED (draft-16)
Comment thread
mondain marked this conversation as resolved.
{
if (offset + 2 > bytes.size()) {
return false;
}
Expand All @@ -316,36 +329,20 @@ bool next_control_message(std::span<const std::uint8_t> bytes, std::size_t& mess
message_size = offset + 2 + payload_length;
return bytes.size() >= message_size;
}
case kSubscribeNamespaceType:
case kSubscribeNamespaceOkType:
case kPublishType:
case kPublishOkType:
case kPublishErrorType: {
if (type == kPublishType || type == kPublishOkType || type == kPublishErrorType) {
if (offset + 2 > bytes.size()) {
return false;
}
const std::size_t payload_length =
(static_cast<std::size_t>(bytes[offset]) << 8) | static_cast<std::size_t>(bytes[offset + 1]);
message_size = offset + 2 + payload_length;
return bytes.size() >= message_size;
}
// varint payload length: type varint + varint length + payload
// SUBSCRIBE_NAMESPACE family — same framing as 0x11 and 0x12.
case kSubscribeNamespaceType: // 0x11
case kSubscribeNamespaceOkType: // 0x12
case 0x13: // SUBSCRIBE_NAMESPACE_ERROR (draft-14)
case 0x1b: // UNSUBSCRIBE_NAMESPACE (draft-14)
{
std::uint64_t payload_length = 0;
if (!decode_varint_impl(bytes, offset, payload_length)) {
return false;
}
message_size = offset + static_cast<std::size_t>(payload_length);
return bytes.size() >= message_size;
}
case kMaxRequestIdType: {
if (offset + 2 > bytes.size()) {
return false;
}
const std::size_t payload_length =
(static_cast<std::size_t>(bytes[offset]) << 8) | static_cast<std::size_t>(bytes[offset + 1]);
message_size = offset + 2 + payload_length;
return bytes.size() >= message_size;
}
default:
return false;
}
Expand Down
37 changes: 34 additions & 3 deletions src/transport/moqt_session.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,6 +87,24 @@ const char* control_message_type_name(std::uint64_t message_type) {
return "CLIENT_SETUP";
case 0x21:
return "SERVER_SETUP";
case 0x0a:
return "UNSUBSCRIBE";
case 0x10:
return "GOAWAY";
case 0x13:
return "SUBSCRIBE_NAMESPACE_ERROR";
case 0x14:
return "SUBSCRIBE_DONE";
case 0x16:
return "FETCH";
case 0x17:
return "FETCH_OK";
case 0x18:
return "FETCH_CANCEL";
case 0x1a:
return "REQUESTS_BLOCKED";
case 0x1b:
return "UNSUBSCRIBE_NAMESPACE";
default:
return "UNKNOWN";
}
Expand Down Expand Up @@ -715,7 +733,22 @@ TransportStatus serve_subscriptions(PublisherTransport& transport,
}
trace_control_message(message_bytes, draft);

if (message_type == 0x12 || message_type == 0x07 || message_type == 0x1e) {
// Discard parsed control messages we don't act on (for example,
// acknowledged responses, FETCH, GOAWAY, or UNSUBSCRIBE) so they
// do not block the control-stream buffer. This only applies to
// messages that next_control_message() can frame successfully;
// log them for visibility only when tracing is explicitly enabled.
const bool is_handled_type =
message_type == 0x02 || // SUBSCRIBE_UPDATE
message_type == 0x11 || // SUBSCRIBE_NAMESPACE
message_type == 0x03; // SUBSCRIBE
if (!is_handled_type) {
if (trace_enabled()) {
std::cerr << "[moqt-session] skipping unhandled control message type=0x"
<< std::hex << message_type << std::dec
<< " (" << control_message_type_name(message_type) << ")"
<< " size=" << message_size << '\n';
}
Comment thread
mondain marked this conversation as resolved.
buffer.erase(buffer.begin(), buffer.begin() + message_size);
continue;
}
Expand Down Expand Up @@ -847,8 +880,6 @@ TransportStatus serve_subscriptions(PublisherTransport& transport,
buffer.erase(buffer.begin(), buffer.begin() + message_size);
continue;
}

return TransportStatus::failure("received unsupported control request");
}

if (!fin && !pending_subscription_order.empty()) {
Expand Down