Merge pull request #20 from mondain/fix/publish-live-namespace-ack-nonfatal

publish_live: don't tear down the session on a missing or late namespace ack

Author: mondain

src/transport/moqt_session.cpp

@@ -839,6 +839,14 @@ TransportStatus collect_control_acknowledgements(PublisherTransport& transport,
         std::vector<std::uint8_t> chunk;
         const TransportStatus status = transport.read_stream(control_stream_id, chunk, fin, std::chrono::seconds(2));
         if (!status.ok) {
+            // Preserve whatever bytes we already parsed-but-deferred (or received
+            // but hadn't parsed yet) before bailing out. A caller that treats this
+            // failure as non-fatal -- e.g. publish_live() tolerating a missing
+            // namespace acknowledgement -- still needs those bytes available to
+            // its own control-message loop, so a late-arriving ack is not
+            // silently dropped here.
+            deferred_messages.insert(deferred_messages.end(), buffer.begin(), buffer.end());
+            pending_control_bytes = std::move(deferred_messages);
             return status;
         }
         if (trace_enabled()) {
@@ -3766,15 +3774,38 @@ TransportStatus MoqtSession::publish_live(std::istream& input,
         if (status.ok) {
             namespace_stream_open_ = true;
         }
+        if (!status.ok) {
+            return status;
+        }
     } else {
         status = write_frame(control_stream_id_, encode_namespace_message(namespace_message), false);
-        if (status.ok) {
-            status = collect_control_acknowledgements(
-                transport_, control_stream_id_, draft_version, 1, 0, pending_control_bytes_);
+        if (!status.ok) {
+            // The write itself failed -- the transport is broken. Always fatal.
+            return status;
+        }
+        status = collect_control_acknowledgements(
+            transport_, control_stream_id_, draft_version, 1, 0, pending_control_bytes_);
+        if (!status.ok) {
+            if (status.message != "timed out waiting for stream data") {
+                // A real transport failure (connection closed/reset, malformed
+                // response, peer rejection, ...) -- still fatal.
+                return status;
+            }
+            // Only the per-read wait (collect_control_acknowledgements's 2s
+            // read_stream timeout) elapsed with no PUBLISH_NAMESPACE_OK/
+            // REQUEST_OK; the connection itself is healthy. Per the live-publish
+            // contract, a missing/late namespace acknowledgement must not tear
+            // the session down: moqxr stays ready, publishes its tracks, and
+            // waits for subscriptions until the RTMP source itself ends. Any
+            // ack that does arrive later is still sitting in
+            // pending_control_bytes_ (collect_control_acknowledgements
+            // preserves it even on this early return) and will be parsed
+            // harmlessly -- and silently, since it isn't one of the message
+            // types process_control_messages() acts on -- once the
+            // await-subscriptions loop below starts draining control bytes.
+            std::cerr << "[moqt-session] live: no namespace acknowledgement within timeout; "
+                         "continuing without waiting further (relay may ack late or not at all)\n";
         }
-    }
-    if (!status.ok) {
-        return status;
     }
 
     // Build track alias map (self-assigned, matching legacy serve_subscriptions behavior)

tests/moqt_session_test.cpp

@@ -2615,6 +2615,95 @@ int main() {
                      "expected live publish to finish with PUBLISH_NAMESPACE_DONE");
     }
 
+    {
+        // Regression for the stack owner's required behavior: once publish_live()
+        // has sent PUBLISH_NAMESPACE, a relay that never acknowledges it must NOT
+        // cause the session to tear down. moqxr should still publish its tracks
+        // and enter the await-subscriptions loop, exiting only via the normal
+        // subscriber-timeout idle path (or when the RTMP/stdin source ends), not
+        // because the ack never arrived. Before the fix, collect_control_
+        // acknowledgements()'s read_stream timeout ("timed out waiting for
+        // stream data") was propagated straight out of publish_live() as fatal.
+        MockTransport no_ack_transport;
+        no_ack_transport.reads[0].push_back(encode_server_setup_message({
+            .draft = DraftVersion::kDraft16,
+            .max_request_id = 8,
+        }));
+        // Deliberately no PUBLISH_NAMESPACE_OK/REQUEST_OK queued for stream 0:
+        // once the setup message is consumed, every further read on the control
+        // stream reports the same failure a real transport reports for a
+        // genuine per-read timeout with a healthy connection.
+        no_ack_transport.missing_read_error = "timed out waiting for stream data";
+
+        MoqtSession no_ack_session(
+            no_ack_transport, std::string(kTestTrackNamespace), false, false, false, std::chrono::seconds(1));
+        status = no_ack_session.connect(endpoint, tls);
+        ok &= expect(status.ok, "expected no-ack session connect to succeed");
+
+        const auto no_ack_bytes = make_live_init_mp4();
+        std::string no_ack_input_bytes(no_ack_bytes.begin(), no_ack_bytes.end());
+        std::istringstream no_ack_input(no_ack_input_bytes);
+        status = no_ack_session.publish_live(no_ack_input, DraftVersion::kDraft16, false);
+        ok &= expect(status.ok,
+                     "expected publish_live to tolerate a missing namespace acknowledgement "
+                     "rather than tearing the session down");
+        ok &= expect(control_message_count(no_ack_transport, 0x1d) == 1,
+                     "expected publish_live to still preannounce media tracks after a missing ack");
+        ok &= expect(!no_ack_transport.writes.empty() &&
+                         message_type(no_ack_transport.writes.back().bytes) == 0x09,
+                     "expected publish_live to still reach a clean PUBLISH_NAMESPACE_DONE exit "
+                     "via the normal idle await-subscribe timeout, not an error return");
+    }
+
+    {
+        // Companion regression: an acknowledgement that arrives late (after
+        // collect_control_acknowledgements() has already given up and
+        // publish_live() has moved on) must be consumed harmlessly by the main
+        // control-message loop rather than being lost or mis-parsed as a
+        // protocol violation. PUBLISH_NAMESPACE_OK/REQUEST_OK (type 0x07) is not
+        // one of the message types process_control_messages() acts on, so it
+        // should simply be drained off pending_control_bytes_ once it shows up.
+        MockTransport late_ack_transport;
+        late_ack_transport.reads[0].push_back(encode_server_setup_message({
+            .draft = DraftVersion::kDraft16,
+            .max_request_id = 8,
+        }));
+        // No ack queued yet -- the first collect_control_acknowledgements() read
+        // times out (same mechanism as the no-ack case above). on_read injects
+        // the ack just before the *second* read of stream 0, modeling a relay
+        // that answers slightly later than the initial short wait -- this is
+        // read by the main await-subscribe loop, not by
+        // collect_control_acknowledgements() (which has already returned).
+        std::size_t stream0_reads = 0;
+        late_ack_transport.on_read = [&stream0_reads](MockTransport& transport, std::uint64_t stream_id) {
+            if (stream_id != 0) {
+                return;
+            }
+            ++stream0_reads;
+            if (stream0_reads == 2) {
+                transport.reads[0].push_back(encode_publish_namespace_ok_message(DraftVersion::kDraft16, 0));
+            }
+        };
+
+        MoqtSession late_ack_session(
+            late_ack_transport, std::string(kTestTrackNamespace), false, false, false, std::chrono::seconds(1));
+        status = late_ack_session.connect(endpoint, tls);
+        ok &= expect(status.ok, "expected late-ack session connect to succeed");
+
+        const auto late_ack_bytes = make_live_init_mp4();
+        std::string late_ack_input_bytes(late_ack_bytes.begin(), late_ack_bytes.end());
+        std::istringstream late_ack_input(late_ack_input_bytes);
+        status = late_ack_session.publish_live(late_ack_input, DraftVersion::kDraft16, false);
+        ok &= expect(status.ok,
+                     "expected a late-arriving namespace acknowledgement to be consumed without "
+                     "surfacing as a protocol violation");
+        ok &= expect(stream0_reads >= 2,
+                     "expected the main loop to perform a second read that picks up the late ack");
+        ok &= expect(!late_ack_transport.writes.empty() &&
+                         message_type(late_ack_transport.writes.back().bytes) == 0x09,
+                     "expected the late-ack session to still reach a clean PUBLISH_NAMESPACE_DONE exit");
+    }
+
     {
         // Phase 5 regression: CMSF 4.1.2 makes an absent contentProtectionRefIDs
         // mean "not protected", so a CENC-protected track (encv/sinf/schm/schi/