Add subscriber wait timeout

mondain · mondain · commit 87268e8f472b · 2026-03-16T14:17:22.000-07:00
diff --git a/CMakeLists.txt b/CMakeLists.txt
@@ -78,6 +78,12 @@ if(OPENMOQ_BUILD_TESTS)
     target_link_libraries(openmoq-publisher-packaging-tests PRIVATE openmoq_publisher_lib)
     add_test(NAME openmoq-publisher-packaging-tests COMMAND openmoq-publisher-packaging-tests)
 
+    add_executable(openmoq-publisher-cli-tests
+        tests/cli_options_test.cpp
+    )
+    target_link_libraries(openmoq-publisher-cli-tests PRIVATE openmoq_publisher_lib)
+    add_test(NAME openmoq-publisher-cli-tests COMMAND openmoq-publisher-cli-tests)
+
     add_executable(openmoq-publisher-transport-tests
         tests/moqt_session_test.cpp
     )
diff --git a/README.md b/README.md
@@ -187,6 +187,7 @@ OPENMOQ_PICOQUIC_TRACE=1 ./build/openmoq-publisher \
   --endpoint moqt://interop-relay.cloudflare.mediaoverquic.com:443/moq \
   --namespace interop \
   --forward 0 \
+  --timeout 10 \
   --paced \
   --insecure
 ```
@@ -197,6 +198,7 @@ Current status as of March 13, 2026:
 - `CLIENT_SETUP` succeeds and the client prints the negotiated connection ID to stdout after setup
 - `PUBLISH_NAMESPACE` is accepted with `PUBLISH_NAMESPACE_OK`
 - with `--forward 0`, the current client waits for inbound `SUBSCRIBE`; relays may consume `SUBSCRIBE_NAMESPACE` themselves and only forward `SUBSCRIBE` to the publisher
+- `--timeout <seconds>` controls how long the publisher waits for inbound `SUBSCRIBE` requests before failing the publish attempt
 - the Cloudflare endpoints accepted setup and namespace announce in testing, but did not issue subscriptions, so the publish attempt timed out waiting for control-stream data
 - with `--forward 1`, `moq-relay.red5.net:8443` now progresses through `PUBLISH_OK` for the catalog and media tracks, after which the client begins sending object streams
 - `fb.mvfst.net:9448` now accepts the draft-14 publish flow end-to-end after switching `PUBLISH`, `PUBLISH_OK`, and `PUBLISH_ERROR` control messages to `u16` outer lengths; the current draft-16 flow is still rejected with MOQT application error `3` (`PROTOCOL_VIOLATION`) immediately after setup
@@ -246,6 +248,7 @@ Transport-oriented CLI flags are also present now:
   --endpoint localhost:4433 \
   --namespace media \
   --forward 0 \
+  --timeout 3 \
   --paced \
   --insecure
 ```
@@ -264,6 +267,7 @@ Current status:
 - the local picoquic loopback handshake works, including object publication over QUIC streams
 - `--namespace` lets you choose the advertised track namespace during transport tests
 - `--forward 0|1` selects whether the publisher waits for `SUBSCRIBE` (`0`) or immediately sends `PUBLISH` requests and forwards objects after namespace announce (`1`)
+- `--timeout <seconds>` sets the subscriber wait timeout used when the publisher is waiting for `SUBSCRIBE`
 - ALPN is selected from the requested draft unless `--alpn` explicitly overrides it
 - `--paced` delays media-object sends to match fragment media timestamps instead of sending the whole file as fast as possible; it only has an effect once object transmission begins
 - after setup completes, the CLI prints `connection_id=<hex>` to stdout
diff --git a/include/openmoq/publisher/cli_options.h b/include/openmoq/publisher/cli_options.h
@@ -1,6 +1,7 @@
 #pragma once
 
 #include <filesystem>
+#include <chrono>
 #include <optional>
 #include <string>
 
@@ -20,6 +21,7 @@ struct CliOptions {
     bool forward = false;
     bool paced = false;
     bool dump_plan = false;
+    std::chrono::seconds subscriber_timeout = std::chrono::seconds(3);
 };
 
 CliOptions parse_cli_options(int argc, char** argv);
diff --git a/include/openmoq/publisher/transport/moqt_session.h b/include/openmoq/publisher/transport/moqt_session.h
@@ -4,6 +4,7 @@
 #include "openmoq/publisher/transport/publisher_transport.h"
 
 #include <optional>
+#include <chrono>
 #include <span>
 #include <string>
 #include <string_view>
@@ -16,7 +17,8 @@ class MoqtSession {
     explicit MoqtSession(PublisherTransport& transport,
                          std::string track_namespace = "media",
                          bool auto_forward = false,
-                         bool paced = false);
+                         bool paced = false,
+                         std::chrono::seconds subscriber_timeout = std::chrono::seconds(3));
 
     TransportStatus connect(const EndpointConfig& endpoint, const TlsConfig& tls);
     TransportStatus publish(const openmoq::publisher::PublishPlan& plan);
@@ -31,6 +33,7 @@ class MoqtSession {
     std::string track_namespace_;
     bool auto_forward_ = false;
     bool paced_ = false;
+    std::chrono::seconds subscriber_timeout_ = std::chrono::seconds(3);
     std::optional<EndpointConfig> endpoint_;
     std::uint64_t control_stream_id_ = 0;
     std::uint64_t peer_max_request_id_ = 0;
diff --git a/src/cli_options.cpp b/src/cli_options.cpp
@@ -59,6 +59,14 @@ bool parse_forward_flag(std::string_view value) {
     throw std::runtime_error("unsupported --forward value: expected 0 or 1");
 }
 
+std::chrono::seconds parse_timeout(std::string_view value) {
+    const int timeout = std::stoi(std::string(value));
+    if (timeout < 0) {
+        throw std::runtime_error("subscriber timeout must be zero or greater");
+    }
+    return std::chrono::seconds(timeout);
+}
+
 }  // namespace
 
 CliOptions parse_cli_options(int argc, char** argv) {
@@ -99,6 +107,8 @@ CliOptions parse_cli_options(int argc, char** argv) {
             options.track_namespace = std::string(require_value("--namespace"));
         } else if (argument == "--forward") {
             options.forward = parse_forward_flag(require_value("--forward"));
+        } else if (argument == "--timeout") {
+            options.subscriber_timeout = parse_timeout(require_value("--timeout"));
         } else if (argument == "--paced") {
             options.paced = true;
         } else if (argument == "--emit-dir") {
@@ -128,7 +138,8 @@ CliOptions parse_cli_options(int argc, char** argv) {
 
 std::string build_usage(const char* argv0) {
     return std::string("Usage: ") + argv0 +
-           " --input <mp4> [--draft 14|16] [--namespace <value>] [--forward 0|1] [--paced] [--dump-plan] [--emit-dir <dir>]"
+           " --input <mp4> [--draft 14|16] [--namespace <value>] [--forward 0|1] [--timeout <seconds>]"
+           " [--paced] [--dump-plan] [--emit-dir <dir>]"
            " [--endpoint host:port|moqt://host:port/path] [--alpn value]"
            " [--cert file] [--key file] [--ca file] [--insecure]";
 }
diff --git a/src/main.cpp b/src/main.cpp
@@ -34,7 +34,8 @@ int main(int argc, char** argv) {
                 endpoint.alpn = default_alpn(options.draft_version);
             }
             PicoquicClient transport;
-            MoqtSession session(transport, options.track_namespace, options.forward, options.paced);
+            MoqtSession session(
+                transport, options.track_namespace, options.forward, options.paced, options.subscriber_timeout);
 
             TransportStatus status = session.connect(endpoint, options.tls);
             if (!status.ok) {
diff --git a/src/transport/moqt_session.cpp b/src/transport/moqt_session.cpp
@@ -448,6 +448,7 @@ TransportStatus serve_subscriptions(PublisherTransport& transport,
                                     openmoq::publisher::DraftVersion draft,
                                     std::string_view track_namespace,
                                     bool paced,
+                                    std::chrono::milliseconds subscriber_timeout,
                                     std::vector<std::uint8_t>& pending_control_bytes,
                                     bool send_namespace_done = true) {
     std::vector<std::uint8_t> buffer = std::move(pending_control_bytes);
@@ -621,7 +622,7 @@ TransportStatus serve_subscriptions(PublisherTransport& transport,
         }
 
         std::vector<std::uint8_t> chunk;
-        const TransportStatus read_status = transport.read_stream(control_stream_id, chunk, fin, std::chrono::seconds(3));
+        const TransportStatus read_status = transport.read_stream(control_stream_id, chunk, fin, subscriber_timeout);
         if (!read_status.ok) {
             if (served_any_subscription &&
                 (read_status.message == "timed out waiting for stream data" ||
@@ -656,6 +657,7 @@ TransportStatus forward_published_tracks(PublisherTransport& transport,
                                          std::uint64_t peer_max_request_id,
                                          std::string_view track_namespace,
                                          bool paced,
+                                         std::chrono::milliseconds subscriber_timeout,
                                          std::vector<std::uint8_t>& pending_control_bytes) {
     std::map<std::string, std::uint64_t> request_id_by_track;
     std::map<std::string, PublishedTrack> tracks_by_name;
@@ -779,6 +781,7 @@ TransportStatus forward_published_tracks(PublisherTransport& transport,
                                      plan.draft.version,
                                      track_namespace,
                                      paced,
+                                     subscriber_timeout,
                                      pending_control_bytes,
                                      false);
         if (!status.ok) {
@@ -797,8 +800,16 @@ TransportStatus forward_published_tracks(PublisherTransport& transport,
 
 }  // namespace
 
-MoqtSession::MoqtSession(PublisherTransport& transport, std::string track_namespace, bool auto_forward, bool paced)
-    : transport_(transport), track_namespace_(std::move(track_namespace)), auto_forward_(auto_forward), paced_(paced) {}
+MoqtSession::MoqtSession(PublisherTransport& transport,
+                         std::string track_namespace,
+                         bool auto_forward,
+                         bool paced,
+                         std::chrono::seconds subscriber_timeout)
+    : transport_(transport),
+      track_namespace_(std::move(track_namespace)),
+      auto_forward_(auto_forward),
+      paced_(paced),
+      subscriber_timeout_(subscriber_timeout) {}
 
 TransportStatus MoqtSession::connect(const EndpointConfig& endpoint, const TlsConfig& tls) {
     endpoint_ = endpoint;
@@ -856,7 +867,15 @@ TransportStatus MoqtSession::publish(const openmoq::publisher::PublishPlan& plan
 
     if (auto_forward_) {
         return forward_published_tracks(
-            transport_, control_stream_id_, plan, tracks, peer_max_request_id_, track_namespace_, paced_, pending_control_bytes_);
+            transport_,
+            control_stream_id_,
+            plan,
+            tracks,
+            peer_max_request_id_,
+            track_namespace_,
+            paced_,
+            subscriber_timeout_,
+            pending_control_bytes_);
     }
 
     std::cerr << "[moqt-session] awaiting SUBSCRIBE for tracks:";
@@ -872,6 +891,7 @@ TransportStatus MoqtSession::publish(const openmoq::publisher::PublishPlan& plan
                                plan.draft.version,
                                track_namespace_,
                                paced_,
+                               subscriber_timeout_,
                                pending_control_bytes_);
 }
 
diff --git a/tests/cli_options_test.cpp b/tests/cli_options_test.cpp
@@ -0,0 +1,60 @@
+#include "openmoq/publisher/cli_options.h"
+
+#include <chrono>
+#include <iostream>
+#include <stdexcept>
+#include <string>
+#include <vector>
+
+namespace {
+
+using openmoq::publisher::CliOptions;
+using openmoq::publisher::parse_cli_options;
+
+bool expect(bool condition, const std::string& message) {
+    if (!condition) {
+        std::cerr << "FAIL: " << message << '\n';
+        return false;
+    }
+    return true;
+}
+
+CliOptions parse(std::vector<std::string> args) {
+    std::vector<char*> argv;
+    argv.reserve(args.size());
+    for (auto& arg : args) {
+        argv.push_back(arg.data());
+    }
+    return parse_cli_options(static_cast<int>(argv.size()), argv.data());
+}
+
+}  // namespace
+
+int main() {
+    bool ok = true;
+
+    {
+        const CliOptions options = parse({"openmoq-publisher", "--input", "sample.mp4"});
+        ok &= expect(options.subscriber_timeout == std::chrono::seconds(3),
+                     "expected default subscriber timeout to remain 3 seconds");
+    }
+
+    {
+        const CliOptions options =
+            parse({"openmoq-publisher", "--input", "sample.mp4", "--timeout", "9", "--forward", "0"});
+        ok &= expect(options.subscriber_timeout == std::chrono::seconds(9),
+                     "expected --timeout to override subscriber timeout");
+    }
+
+    {
+        bool threw = false;
+        try {
+            static_cast<void>(parse({"openmoq-publisher", "--input", "sample.mp4", "--timeout", "-1"}));
+        } catch (const std::runtime_error& error) {
+            threw = std::string(error.what()) == "subscriber timeout must be zero or greater";
+        }
+        ok &= expect(threw, "expected negative --timeout to be rejected");
+    }
+
+    return ok ? 0 : 1;
+}
diff --git a/tests/moqt_session_test.cpp b/tests/moqt_session_test.cpp
@@ -5,6 +5,7 @@
 
 #include <chrono>
 #include <cstdint>
+#include <functional>
 #include <iomanip>
 #include <iostream>
 #include <map>
@@ -88,7 +89,11 @@ struct MockTransport final : PublisherTransport {
                                 std::vector<std::uint8_t>& bytes,
                                 bool& fin,
                                 std::chrono::milliseconds timeout) override {
-        static_cast<void>(timeout);
+        read_timeouts.push_back(timeout);
+        ++read_count;
+        if (on_read) {
+            on_read(*this, stream_id);
+        }
         const auto it = reads.find(stream_id);
         if (it == reads.end()) {
             return TransportStatus::failure("no queued read for stream");
@@ -124,8 +129,11 @@ struct MockTransport final : PublisherTransport {
     std::uint64_t next_bidi_ = 0;
     std::uint64_t next_uni_ = 2;
     std::uint64_t last_close_code = 0;
+    std::size_t read_count = 0;
     std::vector<WriteEvent> writes;
+    std::vector<std::chrono::milliseconds> read_timeouts;
     std::map<std::uint64_t, std::vector<std::vector<std::uint8_t>>> reads;
+    std::function<void(const MockTransport&, std::uint64_t)> on_read;
 };
 
 void append_be16(std::vector<std::uint8_t>& out, std::uint16_t value) {
@@ -554,6 +562,64 @@ int main() {
         ok &= expect(transport.writes[8].bytes == std::vector<std::uint8_t>({0x09, 0x00, 0x09, 0x01, 0x07, 0x69, 0x6e,
                                                                              0x74, 0x65, 0x72, 0x6f, 0x70}),
                      "expected draft-14 PUBLISH_NAMESPACE_DONE to contain the configured track namespace");
+        ok &= expect(!transport.read_timeouts.empty() && transport.read_timeouts.back() == std::chrono::seconds(3),
+                     "expected default subscriber wait timeout to be 3 seconds");
+    }
+
+    {
+        MockTransport transport;
+        transport.reads[0].push_back(encode_server_setup_message({
+            .draft = DraftVersion::kDraft14,
+            .max_request_id = 8,
+        }));
+        transport.reads[0].push_back(encode_publish_namespace_ok_message(DraftVersion::kDraft14, 0));
+        transport.reads[0].push_back(encode_subscribe_message(2, kTestTrackNamespace, "catalog", 0));
+        transport.reads[0].push_back(encode_subscribe_message(4, kTestTrackNamespace, "vide_1", 0));
+
+        bool saw_media_before_media_subscribe = false;
+        transport.on_read = [&](const MockTransport& current, std::uint64_t stream_id) {
+            if (stream_id != 0 || current.read_count != 4) {
+                return;
+            }
+
+            for (const auto& write : current.writes) {
+                if (write.stream_id != 6) {
+                    continue;
+                }
+                std::uint64_t stream_type = 0;
+                std::uint64_t track_alias = 0;
+                std::uint64_t group_id = 0;
+                std::uint64_t subgroup_id = 0;
+                std::uint64_t publisher_priority = 0;
+                std::uint64_t object_id_delta = 0;
+                std::uint64_t payload_length = 0;
+                std::vector<std::uint8_t> payload;
+                if (decode_object_stream_fields(write.bytes,
+                                                stream_type,
+                                                track_alias,
+                                                group_id,
+                                                subgroup_id,
+                                                publisher_priority,
+                                                object_id_delta,
+                                                payload_length,
+                                                payload) &&
+                    track_alias == 1) {
+                    saw_media_before_media_subscribe = true;
+                }
+            }
+        };
+
+        MoqtSession session(transport, std::string(kTestTrackNamespace), false);
+
+        auto status = session.connect(endpoint, tls);
+        ok &= expect(status.ok, "expected delayed-subscriber session connect to succeed");
+
+        const PublishPlan materialized =
+            materialize_publish_plan(make_span_backed_plan(DraftVersion::kDraft14), source_bytes);
+        status = session.publish(materialized);
+        ok &= expect(status.ok, "expected publish to succeed with delayed media subscriber");
+        ok &= expect(!saw_media_before_media_subscribe,
+                     "expected forward=0 to avoid sending media before the media subscriber arrives");
     }
 
     {
@@ -830,5 +896,27 @@ int main() {
     ok &= expect(status.ok, "expected close to succeed");
     ok &= expect(close_transport.last_close_code == 7, "expected close code to propagate");
 
+    {
+        MockTransport timeout_transport;
+        timeout_transport.reads[0].push_back(encode_server_setup_message({
+            .draft = DraftVersion::kDraft14,
+            .max_request_id = 8,
+        }));
+        queue_subscribe_requests(timeout_transport,
+                                 DraftVersion::kDraft14,
+                                 kTestTrackNamespace,
+                                 {{2, "catalog"}, {4, "vide_1"}});
+        MoqtSession timeout_session(
+            timeout_transport, std::string(kTestTrackNamespace), false, false, std::chrono::seconds(11));
+        status = timeout_session.connect(endpoint, tls);
+        ok &= expect(status.ok, "expected custom-timeout session connect to succeed");
+        status = timeout_session.publish(
+            materialize_publish_plan(make_span_backed_plan(DraftVersion::kDraft14), source_bytes));
+        ok &= expect(status.ok, "expected publish to succeed with custom subscriber timeout");
+        ok &= expect(!timeout_transport.read_timeouts.empty() &&
+                         timeout_transport.read_timeouts.back() == std::chrono::seconds(11),
+                     "expected custom subscriber timeout to reach transport reads");
+    }
+
     return ok ? 0 : 1;
 }

Original file line number	Diff line number	Diff line change
`@@ -34,7 +34,8 @@ int main(int argc, char** argv) {`
`34`	`34`	`endpoint.alpn = default_alpn(options.draft_version);`
`35`	`35`	`}`
`36`	`36`	`PicoquicClient transport;`
`37`		`- MoqtSession session(transport, options.track_namespace, options.forward, options.paced);`
	`37`	`+ MoqtSession session(`
	`38`	`+ transport, options.track_namespace, options.forward, options.paced, options.subscriber_timeout);`
`38`	`39`
`39`	`40`	`TransportStatus status = session.connect(endpoint, options.tls);`
`40`	`41`	`if (!status.ok) {`