dash: fix crash-safety, catalog initData, late-track, and shutdown races

Resolve the confirmed findings from the PR #15 review of the CTE LL-DASH
ingest publisher:

- Guard the ingest worker thread so a malformed fragment (e.g. a moof
  referencing an unknown track) is dropped instead of escaping the thread
  and calling std::terminate.
- Embed each track's base64 CMAF init segment in the catalog via a reusable
  track_init_data_base64 helper so subscribers can initialize decoders.
- Freeze the announced track set when source() is taken; paths whose init
  segments arrive afterwards are ignored rather than enqueued against an
  unknown track (which aborted the whole publish session).
- Add a bounded poll plus an is_finished predicate on LiveObjectSource so
  the publisher services control messages during media gaps and ends only
  on close, instead of blocking indefinitely.
- Give each catalog emission a monotonic group id so a re-published catalog
  no longer collides with an already-delivered object id.
- Erase a client fd from the active set before closing it, and defer closing
  the listen fd until after the accept thread joins, to close two fd-reuse
  races on shutdown.
- Reject trailing garbage in --dash-listen/--dash-queue-depth with clear
  messages and require --dash-path/--dash-queue-depth to accompany
  --live-source dash.

Add regression tests for crash-safety, catalog initData, late-path
rejection, and the CLI validation.

Author: mondain

include/openmoq/publisher/cmsf_packager.h

@@ -60,4 +60,12 @@ LiveCatalog build_live_catalog(const std::vector<TrackDescription>& tracks,
                                std::span<const std::uint8_t> init_segment,
                                bool is_live = true);
 
+// Base64-encoded, track-specific CMAF init segment (ftyp + single-track moov)
+// for the track at track_index within init_segment. This is exactly the value
+// build_live_catalog embeds in each track's "initData" field, exposed so live
+// sources that assemble catalogs from per-path init segments can reuse it.
+std::string track_init_data_base64(std::span<const std::uint8_t> init_segment,
+                                   const TrackDescription& track,
+                                   std::size_t track_index);
+
 }  // namespace openmoq::publisher

include/openmoq/publisher/live_dash_ingest.h

@@ -69,6 +69,8 @@ class LiveDashIngestSession {
     LiveObjectSource source();
     std::optional<LiveObject> try_next_object();
 
+    bool finished() const;
+
 private:
     struct PathState {
         StreamingMp4Reader reader;
@@ -79,18 +81,32 @@ class LiveDashIngestSession {
         bool initialized = false;
     };
 
+    // A track announced to subscribers, paired with the base64 CMAF init
+    // segment carried in the catalog so subscribers can initialize decoders.
+    struct RegisteredTrack {
+        TrackDescription description;
+        std::string init_data_base64;
+    };
+
     std::optional<LiveObject> next_object_blocking();
     void process_box_locked(PathState& path_state,
                             std::string_view path,
                             const StreamingBoxResult& box);
     void enqueue_locked(LiveObject object);
+    bool track_published_locked(std::string_view track_name) const;
     std::vector<LiveTrack> snapshot_tracks_locked() const;
-    LiveObject build_catalog_locked() const;
+    LiveObject build_catalog_locked();
 
     std::size_t queue_depth_ = 0;
     std::map<std::string, PathState> paths_;
-    std::vector<TrackDescription> tracks_;
+    std::vector<RegisteredTrack> tracks_;
     std::deque<LiveObject> queue_;
+    // Once source() hands the track snapshot to the publisher, the announced
+    // track set is frozen: paths that appear later cannot be added (the
+    // publisher's alias table is fixed), so their media is dropped rather than
+    // aborting the whole session on an unknown track.
+    bool tracks_frozen_ = false;
+    std::size_t catalog_group_id_ = 0;
     bool catalog_dirty_ = false;
     bool closed_ = false;
     mutable std::mutex mutex_;

include/openmoq/publisher/live_object.h

@@ -27,6 +27,12 @@ struct LiveObject {
 struct LiveObjectSource {
     std::vector<LiveTrack> tracks;
     std::function<std::optional<LiveObject>()> next_object;
+    // Optional liveness predicate. When set, a nullopt from next_object() is
+    // treated as a transient gap (the publisher keeps polling and servicing
+    // control messages) as long as this returns false; it means end-of-stream
+    // only once this returns true. When unset, a nullopt means end-of-stream,
+    // preserving the behavior of finite sources.
+    std::function<bool()> is_finished;
 };
 
 }  // namespace openmoq::publisher

src/cli_options.cpp

@@ -123,21 +123,41 @@ LiveSourceKind parse_live_source(std::string_view value) {
     throw std::runtime_error("unsupported --live-source value: expected auto, stdin, srt, or dash");
 }
 
+int parse_strict_int(std::string_view value, std::string_view option_name) {
+    // Reject empty, non-numeric, or trailing-garbage input with a clear,
+    // option-specific message rather than a bare std::stoi("stoi") exception,
+    // and so "80abc" is not silently accepted as 80.
+    if (value.empty()) {
+        throw std::runtime_error(std::string(option_name) + " requires a numeric value");
+    }
+    std::size_t consumed = 0;
+    int parsed = 0;
+    try {
+        parsed = std::stoi(std::string(value), &consumed);
+    } catch (const std::exception&) {
+        throw std::runtime_error(std::string(option_name) + " must be a valid integer");
+    }
+    if (consumed != value.size()) {
+        throw std::runtime_error(std::string(option_name) + " must be a valid integer");
+    }
+    return parsed;
+}
+
 std::pair<std::string, std::uint16_t> parse_host_port(std::string_view value, std::string_view option_name) {
     const std::size_t colon = value.rfind(':');
     if (colon == std::string_view::npos || colon == 0 || colon + 1 >= value.size()) {
         throw std::runtime_error(std::string(option_name) + " must be in host:port form");
     }
     const std::string host(value.substr(0, colon));
-    const int port = std::stoi(std::string(value.substr(colon + 1)));
+    const int port = parse_strict_int(value.substr(colon + 1), std::string(option_name) + " port");
     if (port <= 0 || port > 65535) {
         throw std::runtime_error(std::string(option_name) + " port must be between 1 and 65535");
     }
     return {host, static_cast<std::uint16_t>(port)};
 }
 
 std::size_t parse_queue_depth(std::string_view value) {
-    const int depth = std::stoi(std::string(value));
+    const int depth = parse_strict_int(value, "--dash-queue-depth");
     if (depth <= 0) {
         throw std::runtime_error("--dash-queue-depth must be greater than zero");
     }
@@ -148,6 +168,8 @@ std::size_t parse_queue_depth(std::string_view value) {
 
 CliOptions parse_cli_options(int argc, char** argv) {
     CliOptions options;
+    bool dash_path_set = false;
+    bool dash_queue_depth_set = false;
 
     for (int index = 1; index < argc; ++index) {
         const std::string_view argument = argv[index];
@@ -174,8 +196,10 @@ CliOptions parse_cli_options(int argc, char** argv) {
             options.dash_listen_port = port;
         } else if (argument == "--dash-path") {
             options.dash_path_prefix = std::string(require_value("--dash-path"));
+            dash_path_set = true;
         } else if (argument == "--dash-queue-depth") {
             options.dash_queue_depth = parse_queue_depth(require_value("--dash-queue-depth"));
+            dash_queue_depth_set = true;
         } else if (argument == "--transport") {
             options.transport = parse_transport_kind(require_value("--transport"));
         } else if (argument == "--endpoint") {
@@ -261,6 +285,12 @@ CliOptions parse_cli_options(int argc, char** argv) {
     if (!live_source_uses_dash && options.dash_listen.has_value()) {
         throw std::runtime_error("--dash-listen requires --live-source dash");
     }
+    if (!live_source_uses_dash && dash_path_set) {
+        throw std::runtime_error("--dash-path requires --live-source dash");
+    }
+    if (!live_source_uses_dash && dash_queue_depth_set) {
+        throw std::runtime_error("--dash-queue-depth requires --live-source dash");
+    }
     if (!options.dash_path_prefix.empty() && options.dash_path_prefix.front() != '/') {
         throw std::runtime_error("--dash-path must start with /");
     }

src/cmsf_packager.cpp

@@ -795,4 +795,10 @@ LiveCatalog build_live_catalog(const std::vector<TrackDescription>& tracks,
     return result;
 }
 
+std::string track_init_data_base64(std::span<const std::uint8_t> init_segment,
+                                   const TrackDescription& track,
+                                   std::size_t track_index) {
+    return base64_encode(build_track_specific_init_segment(init_segment, track, track_index));
+}
+
 }  // namespace openmoq::publisher