Merge pull request #21 from TilsonJoji/main

Keep idle publisher sessions alive, and make track preannounce opt-in

Author: mondain

include/openmoq/publisher/cli_options.h

@@ -62,6 +62,8 @@ struct CliOptions {
     bool stream_per_object = false;
     bool paced = false;
     bool loop = false;
+    // See PublisherConfig::preannounce_tracks.
+    bool preannounce_tracks = false;
     bool dump_plan = false;
     bool print_msf_urls = false;
     std::chrono::seconds subscriber_timeout = std::chrono::seconds(30);

include/openmoq/publisher/publisher_api.h

@@ -46,6 +46,14 @@ struct PublisherConfig {
     std::string track_namespace = "media";
     bool forward = false;
     bool publish_catalog = false;
+    // Send a PUBLISH for each track immediately after PUBLISH_NAMESPACE, before any
+    // subscriber exists. Relays differ here: some accept tracks that way and never
+    // forward a SUBSCRIBE upstream, so a publisher waiting for one would stall,
+    // while others resolve the track namespace only once a subscriber appears and
+    // are disturbed by an early PUBLISH. Off by default because the path that
+    // sends it does not process the PUBLISH_OK it would receive, so the request
+    // would be made and its answer ignored.
+    bool preannounce_tracks = false;
     bool include_sap = false;
     bool include_msf_timeline = false;
     bool split_cmaf_chunks = true;

include/openmoq/publisher/transport/moqt_session.h

@@ -115,6 +115,11 @@ class MoqtSession {
         catalog_republish_interval_ = interval;
     }
 
+    // See PublisherConfig::preannounce_tracks.
+    void set_preannounce_tracks(bool enabled) {
+        preannounce_tracks_ = enabled;
+    }
+
 private:
     void reset_publish_stats();
     void record_published_object(const std::string& track_name, std::uint64_t group_id, std::size_t payload_bytes);
@@ -137,6 +142,7 @@ class MoqtSession {
     std::string track_namespace_;
     bool auto_forward_ = false;
     bool publish_catalog_ = false;
+    bool preannounce_tracks_ = false;
     bool paced_ = false;
     bool loop_ = false;
     std::chrono::seconds subscriber_timeout_ = std::chrono::seconds(30);

src/cli_options.cpp

@@ -375,6 +375,8 @@ CliOptions parse_cli_options(int argc, char** argv) {
             options.stream_per_object = true;
         } else if (argument == "--timeout") {
             options.subscriber_timeout = parse_timeout(require_value("--timeout"));
+        } else if (argument == "--preannounce-tracks") {
+            options.preannounce_tracks = true;
         } else if (argument == "--paced") {
             options.paced = true;
         } else if (argument == "--loop") {
@@ -495,7 +497,7 @@ std::string build_usage(const char* argv0) {
            " --input <mp4|-> [--live-source auto|stdin|srt|dash] [--srt-config <path>]"
            " [--dash-listen host:port] [--dash-path <prefix>] [--dash-queue-depth <count>]"
            " [--transport raw|webtransport] [--draft 14|16|17|18] [--namespace <value>] [--forward 0|1] [--timeout <seconds>]"
-           " [--publish-catalog] [--sap] [--msf-timeline] [--coalesce-cmaf-chunks] [--stream-per-object] [--paced] [--loop] [--dump-plan] [--print-msf-urls] [--emit-dir <dir>]"
+           " [--publish-catalog] [--sap] [--msf-timeline] [--coalesce-cmaf-chunks] [--stream-per-object] [--paced] [--loop] [--preannounce-tracks] [--dump-plan] [--print-msf-urls] [--emit-dir <dir>]"
            " [--vod] [--catalog-republish-interval <seconds>] [--drm-config <path>]"
            " [--endpoint host:port|moqt://host:port/path|https://host:port/path] [--url moqt://host/path#msf:ns--track] [--alpn value] [--sni value]"
            " [--cert file] [--key file] [--ca file] [--insecure]";

src/main.cpp

@@ -51,6 +51,7 @@ int main(int argc, char** argv) {
             .track_namespace = options.track_namespace,
             .forward = options.forward,
             .publish_catalog = options.publish_catalog,
+            .preannounce_tracks = options.preannounce_tracks,
             .include_sap = options.include_sap,
             .include_msf_timeline = options.include_msf_timeline,
             .split_cmaf_chunks = options.split_cmaf_chunks,

src/publisher_api.cpp

@@ -191,6 +191,7 @@ transport::TransportStatus Publisher::publish(const PreparedPublish& prepared,
         config_.subscriber_timeout,
         config_.authorization);
     active->session->set_catalog_republish_interval(config_.catalog_republish_interval);
+    active->session->set_preannounce_tracks(config_.preannounce_tracks);
 
     const transport::EndpointConfig resolved_endpoint = resolve_endpoint(endpoint, endpoint_alpn_overridden);
     set_active_session(active, resolved_endpoint, false);
@@ -358,6 +359,7 @@ transport::TransportStatus Publisher::publish_live(const LiveIngestConfig& inges
         config_.subscriber_timeout,
         config_.authorization);
     active->session->set_catalog_republish_interval(config_.catalog_republish_interval);
+    active->session->set_preannounce_tracks(config_.preannounce_tracks);
 
     const transport::EndpointConfig resolved_endpoint = resolve_endpoint(endpoint, endpoint_alpn_overridden);
     set_active_session(active, resolved_endpoint, true);
@@ -492,6 +494,7 @@ transport::TransportStatus Publisher::publish_live_objects(const LiveObjectSourc
         config_.subscriber_timeout,
         config_.authorization);
     active->session->set_catalog_republish_interval(config_.catalog_republish_interval);
+    active->session->set_preannounce_tracks(config_.preannounce_tracks);
 
     const transport::EndpointConfig resolved_endpoint = resolve_endpoint(endpoint, endpoint_alpn_overridden);
     set_active_session(active, resolved_endpoint, true);

src/transport/moqt_session.cpp

@@ -4792,7 +4792,19 @@ TransportStatus MoqtSession::publish_live_objects(const openmoq::publisher::Live
         ++next_alias;
     }
 
-    if (!uses_request_streams(draft_version)) {
+    // Opt-in preannounce: send a PUBLISH per track before any subscriber exists.
+    //
+    // Relays split into two camps. Some accept tracks this way, answer PUBLISH_OK
+    // and never forward a SUBSCRIBE upstream, so a publisher that waits for one
+    // would stall without this. Others resolve the track namespace only once a
+    // subscriber appears; an early PUBLISH there is at best ignored and at worst
+    // disturbs the namespace registration, after which SUBSCRIBEs are rejected.
+    //
+    // Off by default, because this path never dispatches PUBLISH_OK: the Forward
+    // State in the reply cannot be read, so the request would be made and its
+    // answer discarded. Callers whose relay needs it set
+    // PublisherConfig::preannounce_tracks (CLI: --preannounce-tracks).
+    if (!uses_request_streams(draft_version) && preannounce_tracks_) {
         std::uint64_t request_id = 2;
         for (const auto& [track_name, alias] : alias_by_track) {
             TrackMessage track_message{

src/transport/picoquic_client.cpp

@@ -543,6 +543,21 @@ TransportStatus PicoquicClient::connect() {
 
     picoquic_set_callback(impl_->cnx, client_callback, impl_.get());
 
+    // Keep the connection alive while the application has nothing to send.
+    //
+    // A publisher in await-subscribe mode parks in a blocking control-stream read
+    // until a SUBSCRIBE arrives, transmitting nothing meanwhile. With no keepalive
+    // the peer's idle timeout expires first and tears the connection down beneath
+    // that read, which surfaces to the caller as "transport close requested" after
+    // roughly 30 seconds. That makes a long-running publisher unusable whenever a
+    // subscriber is absent or between sessions.
+    //
+    // The interval must be comfortably below the negotiated idle timeout. Five
+    // seconds is far enough inside any commonly negotiated value to be safe, and
+    // the traffic is one PING frame, so the cost is negligible.
+    constexpr std::uint64_t kKeepAliveIntervalUs = 5'000'000;
+    picoquic_enable_keep_alive(impl_->cnx, kKeepAliveIntervalUs);
+
     if (picoquic_start_client_cnx(impl_->cnx) != 0) {
         picoquic_free(impl_->quic);
         impl_->cnx = nullptr;

tests/moqt_session_test.cpp

@@ -3119,6 +3119,8 @@ int main() {
 
         MoqtSession object_live_session(
             object_live_transport, std::string(kTestTrackNamespace), false, false, false, std::chrono::seconds(1));
+        // Preannounce is opt-in; this case exists to exercise it.
+        object_live_session.set_preannounce_tracks(true);
         status = object_live_session.connect(endpoint, tls);
         ok &= expect(status.ok, "expected arbitrary live-object session connect to succeed");
         status = object_live_session.publish_live_objects(source, DraftVersion::kDraft14);
@@ -3134,6 +3136,52 @@ int main() {
                      "expected arbitrary live-object publish to finish with PUBLISH_NAMESPACE_DONE");
     }
 
+    {
+        // Default (preannounce off): no PUBLISH is emitted before a subscriber
+        // exists. The same flow as above with the opt-in left at its default, so a
+        // regression in the gate shows up as a count change rather than silently.
+        MockTransport default_no_preannounce_transport;
+        default_no_preannounce_transport.reads[0].push_back(encode_server_setup_message({
+            .draft = DraftVersion::kDraft14,
+            .max_request_id = 8,
+        }));
+        default_no_preannounce_transport.reads[0].push_back(
+            encode_publish_namespace_ok_message(DraftVersion::kDraft14, 0));
+        default_no_preannounce_transport.reads[0].push_back({});
+
+        std::vector<LiveObject> objects = {
+            LiveObject{
+                .track_name = "events",
+                .group_id = 7,
+                .subgroup_id = 0,
+                .object_id = 3,
+                .payload = {'O', 'K'},
+            },
+        };
+        std::size_t object_index = 0;
+        LiveObjectSource source{
+            .tracks = {LiveTrack{.track_name = "events"}},
+            .next_object = [&objects, &object_index]() -> std::optional<LiveObject> {
+                if (object_index >= objects.size()) {
+                    return std::nullopt;
+                }
+                return objects[object_index++];
+            },
+        };
+
+        MoqtSession default_no_preannounce_session(default_no_preannounce_transport,
+                                                   std::string(kTestTrackNamespace),
+                                                   false,
+                                                   false,
+                                                   false,
+                                                   std::chrono::seconds(1));
+        status = default_no_preannounce_session.connect(endpoint, tls);
+        ok &= expect(status.ok, "expected default live-object session connect to succeed");
+        status = default_no_preannounce_session.publish_live_objects(source, DraftVersion::kDraft14);
+        ok &= expect(control_message_count(default_no_preannounce_transport, 0x1d) == 0,
+                     "expected no PUBLISH preannounce when preannounce_tracks is left at its default");
+    }
+
     {
         // Regression: draft-18 assigns a track alias in PUBLISH, so an
         // auto-forward live-object publisher must establish every track on a
@@ -3840,6 +3888,8 @@ int main() {
             true,
             false,
             std::chrono::seconds(1));
+        // Preannounce is opt-in; this case asserts the PUBLISH count it produces.
+        dash_live_session.set_preannounce_tracks(true);
         status = dash_live_session.connect(endpoint, tls);
         ok &= expect(status.ok, "expected FFmpeg-style DASH session connect to succeed");
         status = dash_live_session.publish_live_objects(source, DraftVersion::kDraft16);