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

collect_control_acknowledgements() waits up to 2s for
PUBLISH_NAMESPACE_OK/REQUEST_OK on the control stream; when that read
timed out, publish_live() (draft-14/16 path) propagated the failure
and returned, aborting the whole publish before ever announcing tracks
or entering the await-subscriptions loop. Against a relay that acks
slowly or not at all, this made the stdin/RTMP live-publish path give
up every ~2-4s and get torn down by the caller, even though the
transport itself was perfectly healthy.

A genuine transport failure (the PUBLISH_NAMESPACE write itself
failing, or read_stream reporting a real close/reset) is still fatal.
Only the specific "timed out waiting for stream data" read timeout --
which both transport backends only return when the connection is
otherwise alive -- is now treated as non-fatal: publish_live() logs it
and proceeds to publish tracks and await subscriptions, matching the
live-publish contract that a namespace ack is not a precondition for
staying up.

collect_control_acknowledgements() also now preserves whatever control
bytes it had already parsed-but-deferred or buffered-but-unparsed
before bailing out on that timeout, instead of discarding them. That
way a namespace ack that does show up shortly after publish_live()
moves on isn't lost -- it sits in pending_control_bytes_ and gets
drained by the main control-message loop, which already skips message
types (like PUBLISH_NAMESPACE_OK/REQUEST_OK) it doesn't act on rather
than treating them as a protocol violation.

Adds two regression tests covering: (1) a namespace ack that never
arrives, and (2) one that arrives late during the await-subscribe
loop.

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/