Add handling of unknown control message, avoid blocking

src/transport/moqt_control_messages.cpp

@@ -337,7 +337,12 @@ bool next_control_message(std::span<const std::uint8_t> bytes, std::size_t& mess
             message_size = offset + static_cast<std::size_t>(payload_length);
             return bytes.size() >= message_size;
         }
-        case kMaxRequestIdType: {
+        case kMaxRequestIdType:
+        default: {
+            // All known length-prefixed messages (including unknown future types) use
+            // a uint16 payload length immediately after the type varint.  Attempt to
+            // consume the message this way so that unrecognised messages (e.g. FETCH,
+            // GOAWAY, UNSUBSCRIBE) do not block the buffer.
             if (offset + 2 > bytes.size()) {
                 return false;
             }
@@ -346,8 +351,6 @@ 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;
         }
-        default:
-            return false;
     }
 }
 

src/transport/moqt_session.cpp

@@ -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";
     }
@@ -715,7 +733,18 @@ TransportStatus serve_subscriptions(PublisherTransport& transport,
             }
             trace_control_message(message_bytes, draft);
 
-            if (message_type == 0x12 || message_type == 0x07 || message_type == 0x1e) {
+            // Discard messages we don't act on (acknowledged responses,
+            // FETCH, GOAWAY, UNSUBSCRIBE, and any future message types) so they
+            // never block the control-stream buffer. Log them for visibility.
+            const bool is_handled_type =
+                message_type == 0x02 ||  // SUBSCRIBE_UPDATE
+                message_type == 0x11 ||  // SUBSCRIBE_NAMESPACE
+                message_type == 0x03;    // SUBSCRIBE
+            if (!is_handled_type) {
+                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';
                 buffer.erase(buffer.begin(), buffer.begin() + message_size);
                 continue;
             }
@@ -847,8 +876,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()) {