fix: restore live DASH playback

Author: mondain

docs/quickstart.md

@@ -156,6 +156,13 @@ ffmpeg -re \
   http://127.0.0.1:8080/ingest/
 ```
 
+For FFmpeg DASH input, the ingest adapter expands `tfhd` sample defaults into
+the outgoing `trun` entries before publishing CMAF objects. Video keyframes
+start shared media groups, and audio objects join the active video group;
+media received before the first video keyframe is discarded. This keeps the
+representations independently addressable while preserving aligned render
+groups for downstream playback.
+
 Useful DASH ingest flags:
 
 - `--dash-listen <host:port>` chooses the local HTTP listener address

docs/testing.md

@@ -25,7 +25,8 @@ This covers:
 
 Publish-plan numbering notes:
 
-- `group_id` is allocated per track, not across all tracks, so interleaved audio and video fragments can both use `0, 1, 2, ...`
+- file-based publish plans allocate `group_id` per track, so interleaved audio and video fragments can both use `0, 1, 2, ...`
+- live DASH uses each video keyframe to start a shared media group; audio joins that active group and each track maintains its own `object_id` sequence within it
 - by default, `object_id` advances within a group when CMAF content is split into multiple MOQT objects for lower latency
 - `--coalesce-cmaf-chunks` forces `object_id = 0` for the current one-object-per-group fallback
 - MSF media timeline tracks are disabled by default; add `--msf-timeline` when you want a `timeline` metadata track and object
@@ -115,4 +116,9 @@ cmake --build build --target openmoq-publisher-live-dash-tests openmoq-publisher
 ./build/openmoq-publisher-transport-tests
 ```
 
-The DASH test covers separate init and media requests on `/ingest/video0`, `/ingest/video1`, and `/ingest/video2`. The transport test verifies that a draft-16 live publisher advertises those tracks, waits for `SUBSCRIBE`, acknowledges it, and sends only the catalog plus the subscribed representation.
+The DASH test covers separate init and media requests, shared audio/video
+keyframe groups, and normalization of FFmpeg `tfhd` sample defaults into
+explicit `trun` sample fields. The transport test verifies that draft-16 keeps
+objects from the same group on one subgroup stream, closes that stream at a
+group transition, waits for `SUBSCRIBE`, and preserves the catalog while
+waiting for media interest.

docs/webtransport-compliance.md

@@ -200,4 +200,12 @@ For `--forward 0`, once `PUBLISH_NAMESPACE_OK` has been received:
 
 This behavior is intentional. It keeps interoperability probes from being reported as publish failures when the relay accepted the namespace but no subscriber appeared during the configured wait window.
 
-Automated regressions cover the reported live DASH shape with catalog plus `/ingest/video0`, `/ingest/video1`, and `/ingest/video2` representations. The draft-16 session test verifies that source objects are not consumed before `SUBSCRIBE`, that `SUBSCRIBE_OK` is returned, and that only the catalog plus subscribed representation are sent. Draft-18 session coverage separately verifies that a `SUBSCRIBE` split across multiple request-stream reads is reassembled and answered on that request stream.
+Automated regressions cover the reported live DASH shape with catalog plus
+multiple representation paths. They verify shared audio/video keyframe groups,
+explicit `trun` sample metadata derived from FFmpeg `tfhd` defaults, subgroup
+stream reuse within a group, and stream closure at a group transition. The
+draft-16 session test also verifies that source objects are not consumed before
+`SUBSCRIBE`, that `SUBSCRIBE_OK` is returned, and that the catalog remains
+available while waiting for media interest. Draft-18 session coverage
+separately verifies that a `SUBSCRIBE` split across multiple request-stream
+reads is reassembled and answered on that request stream.

include/openmoq/publisher/live_dash_ingest.h

@@ -8,6 +8,7 @@
 #include <memory>
 #include <mutex>
 #include <optional>
+#include <set>
 #include <span>
 #include <string>
 #include <vector>
@@ -107,6 +108,11 @@ class LiveDashIngestSession {
     // aborting the whole session on an unknown track.
     bool tracks_frozen_ = false;
     std::size_t catalog_group_id_ = 0;
+    std::size_t shared_media_group_id_ = 0;
+    std::uint64_t shared_group_start_time_us_ = 0;
+    std::map<std::string, std::size_t> shared_object_id_by_track_;
+    std::set<std::string> video_tracks_started_in_group_;
+    bool shared_media_group_started_ = false;
     bool catalog_dirty_ = false;
     bool closed_ = false;
     mutable std::mutex mutex_;

src/cmaf_segmenter.cpp

@@ -212,6 +212,124 @@ std::vector<std::uint8_t> concat_boxes(const std::vector<std::vector<std::uint8_
     return out;
 }
 
+struct LiveTrunSample {
+    std::uint32_t duration = 0;
+    std::uint32_t size = 0;
+    std::uint32_t flags = 0;
+    std::uint32_t composition_offset = 0;
+};
+
+std::vector<std::uint8_t> materialize_live_trun_defaults(
+    std::span<const std::uint8_t> moof_bytes,
+    std::uint32_t default_sample_duration,
+    std::uint32_t default_sample_size,
+    std::uint32_t default_sample_flags) {
+    const auto boxes = parse_mp4_boxes(moof_bytes);
+    if (boxes.empty() || boxes.front().type != "moof") {
+        throw std::runtime_error("materialize_live_trun_defaults: expected moof box");
+    }
+    const Mp4Box& moof = boxes.front();
+    const Mp4Box* traf = find_child_box(moof, "traf");
+    const Mp4Box* trun = traf == nullptr ? nullptr : find_child_box(*traf, "trun");
+    if (traf == nullptr || trun == nullptr || trun->payload.size < 8) {
+        throw std::runtime_error("materialize_live_trun_defaults: missing traf/trun");
+    }
+
+    const std::uint8_t version = moof_bytes[trun->payload.offset];
+    const std::uint32_t original_flags = read_full_box_flags(*trun, moof_bytes);
+    std::size_t cursor = trun->payload.offset + 4;
+    const std::uint32_t sample_count = read_be32(moof_bytes, cursor);
+    cursor += 4;
+    if ((original_flags & 0x000001U) != 0) {
+        if (cursor + 4 > moof_bytes.size()) {
+            throw std::runtime_error("materialize_live_trun_defaults: truncated data offset");
+        }
+        cursor += 4;
+    }
+
+    std::uint32_t first_sample_flags = default_sample_flags;
+    if ((original_flags & 0x000004U) != 0) {
+        if (cursor + 4 > moof_bytes.size()) {
+            throw std::runtime_error("materialize_live_trun_defaults: truncated first-sample flags");
+        }
+        first_sample_flags = read_be32(moof_bytes, cursor);
+        cursor += 4;
+    }
+
+    std::vector<LiveTrunSample> samples;
+    samples.reserve(sample_count);
+    for (std::uint32_t index = 0; index < sample_count; ++index) {
+        LiveTrunSample sample{
+            .duration = default_sample_duration,
+            .size = default_sample_size,
+            .flags = index == 0 ? first_sample_flags : default_sample_flags,
+        };
+        const auto read_optional = [&](std::uint32_t flag, std::uint32_t& value) {
+            if ((original_flags & flag) == 0) {
+                return;
+            }
+            if (cursor + 4 > moof_bytes.size()) {
+                throw std::runtime_error("materialize_live_trun_defaults: truncated sample fields");
+            }
+            value = read_be32(moof_bytes, cursor);
+            cursor += 4;
+        };
+        read_optional(0x000100U, sample.duration);
+        read_optional(0x000200U, sample.size);
+        read_optional(0x000400U, sample.flags);
+        read_optional(0x000800U, sample.composition_offset);
+        if (sample.size == 0) {
+            throw std::runtime_error("materialize_live_trun_defaults: missing sample size");
+        }
+        samples.push_back(sample);
+    }
+
+    const std::uint32_t normalized_flags = (original_flags & ~0x000004U) | 0x000701U;
+    const auto build_trun = [&](std::uint32_t data_offset) {
+        std::vector<std::uint8_t> payload;
+        append_be32(payload, sample_count);
+        append_be32(payload, data_offset);
+        for (const auto& sample : samples) {
+            append_be32(payload, sample.duration);
+            append_be32(payload, sample.size);
+            append_be32(payload, sample.flags);
+            if ((normalized_flags & 0x000800U) != 0) {
+                append_be32(payload, sample.composition_offset);
+            }
+        }
+        return make_full_box("trun", version, normalized_flags, payload);
+    };
+
+    const auto rebuild_moof = [&](const std::vector<std::uint8_t>& normalized_trun) {
+        std::vector<std::vector<std::uint8_t>> traf_children;
+        traf_children.reserve(traf->children.size());
+        for (const auto& child : traf->children) {
+            if (child.span.offset == trun->span.offset) {
+                traf_children.push_back(normalized_trun);
+            } else {
+                const auto raw = slice_bytes(moof_bytes, child.span);
+                traf_children.emplace_back(raw.begin(), raw.end());
+            }
+        }
+        const auto normalized_traf = make_box("traf", concat_boxes(traf_children));
+
+        std::vector<std::vector<std::uint8_t>> moof_children;
+        moof_children.reserve(moof.children.size());
+        for (const auto& child : moof.children) {
+            if (child.span.offset == traf->span.offset) {
+                moof_children.push_back(normalized_traf);
+            } else {
+                const auto raw = slice_bytes(moof_bytes, child.span);
+                moof_children.emplace_back(raw.begin(), raw.end());
+            }
+        }
+        return make_box("moof", concat_boxes(moof_children));
+    };
+
+    const auto placeholder_moof = rebuild_moof(build_trun(0));
+    return rebuild_moof(build_trun(static_cast<std::uint32_t>(placeholder_moof.size() + 8)));
+}
+
 const Mp4Box* require_child(const Mp4Box& box, std::string_view type) {
     const Mp4Box* child = find_child_box(box, type);
     if (child == nullptr) {
@@ -1048,6 +1166,7 @@ MediaFragment build_live_fragment(std::span<const std::uint8_t> moof_bytes,
     }
 
     std::uint32_t default_sample_duration = 0;
+    std::uint32_t default_sample_size = 0;
     std::uint32_t default_sample_flags = 0x02000000U;
     if (const Mp4Box* tfhd = find_child_box(*traf, "tfhd")) {
         const std::uint32_t flags = read_full_box_flags(*tfhd, moof_bytes);
@@ -1059,6 +1178,7 @@ MediaFragment build_live_fragment(std::span<const std::uint8_t> moof_bytes,
             cursor += 4;
         }
         if ((flags & 0x000010U) != 0 && cursor + 4 <= moof_bytes.size()) {
+            default_sample_size = read_be32(moof_bytes, cursor);
             cursor += 4;
         }
         if ((flags & 0x000020U) != 0 && cursor + 4 <= moof_bytes.size()) {
@@ -1131,10 +1251,16 @@ MediaFragment build_live_fragment(std::span<const std::uint8_t> moof_bytes,
     // A video keyframe: video track with sync first sample
     const bool is_video_keyframe = is_video && first_sample_is_sync;
 
+    // FFmpeg's DASH muxer relies on tfhd defaults and emits minimal trun
+    // entries. Materialize those values for CMAF consumers that read sample
+    // metadata directly from trun.
+    const std::vector<std::uint8_t> normalized_moof = materialize_live_trun_defaults(
+        moof_bytes, default_sample_duration, default_sample_size, default_sample_flags);
+
     // Combine moof+mdat into a single owned payload (CMSF compliance).
     std::vector<std::uint8_t> payload;
-    payload.reserve(moof_bytes.size() + mdat_bytes.size());
-    payload.insert(payload.end(), moof_bytes.begin(), moof_bytes.end());
+    payload.reserve(normalized_moof.size() + mdat_bytes.size());
+    payload.insert(payload.end(), normalized_moof.begin(), normalized_moof.end());
     payload.insert(payload.end(), mdat_bytes.begin(), mdat_bytes.end());
 
     return MediaFragment{

src/live_dash_ingest.cpp

@@ -453,7 +453,40 @@ void LiveDashIngestSession::process_box_locked(PathState& path_state,
     if (!track_published_locked(fragment.track_name)) {
         return;
     }
-    fragment.group_id = path_state.next_group_by_track[fragment.track_name]++;
+    const auto track_it = std::find_if(path_state.tracks.begin(), path_state.tracks.end(),
+                                       [&](const TrackDescription& track) {
+                                           return track.track_name == fragment.track_name;
+                                       });
+    const bool is_video = track_it != path_state.tracks.end() && track_it->handler_type == "vide";
+    const bool has_video = std::any_of(tracks_.begin(), tracks_.end(),
+                                       [](const RegisteredTrack& track) {
+                                           return track.description.handler_type == "vide";
+                                       });
+    if (has_video) {
+        if (is_video && fragment.is_video_keyframe) {
+            if (!shared_media_group_started_) {
+                shared_media_group_started_ = true;
+                shared_group_start_time_us_ = fragment.start_time_us;
+            } else if (fragment.start_time_us > shared_group_start_time_us_) {
+                ++shared_media_group_id_;
+                shared_group_start_time_us_ = fragment.start_time_us;
+                shared_object_id_by_track_.clear();
+                video_tracks_started_in_group_.clear();
+            } else if (fragment.start_time_us < shared_group_start_time_us_) {
+                return;
+            }
+            video_tracks_started_in_group_.insert(fragment.track_name);
+        }
+        if (!shared_media_group_started_ ||
+            (is_video && !video_tracks_started_in_group_.contains(fragment.track_name))) {
+            return;
+        }
+        fragment.group_id = shared_media_group_id_;
+        fragment.object_id = shared_object_id_by_track_[fragment.track_name]++;
+    } else {
+        fragment.group_id = path_state.next_group_by_track[fragment.track_name]++;
+        fragment.object_id = 0;
+    }
     enqueue_locked(LiveObject{
         .track_name = fragment.track_name,
         .group_id = fragment.group_id,
@@ -462,6 +495,7 @@ void LiveDashIngestSession::process_box_locked(PathState& path_state,
         .media_time_us = fragment.start_time_us,
         .media_duration_us = fragment.duration_us,
         .payload = std::move(fragment.payload.owned_bytes),
+        .final_in_subgroup = false,
     });
 }
 

src/transport/moqt_session.cpp

@@ -4596,6 +4596,7 @@ TransportStatus MoqtSession::publish_live_objects(const openmoq::publisher::Live
     bool control_fin = false;
     std::optional<std::chrono::steady_clock::time_point> object_pacing_start;
     std::optional<std::uint64_t> object_first_media_time_us;
+    std::map<std::string, std::uint64_t> last_group_id_by_track;
     bool live_object_catalog_sent = !alias_by_track.contains("catalog");
     const bool has_media_tracks = std::any_of(alias_by_track.begin(), alias_by_track.end(),
                                               [](const auto& entry) { return entry.first != "catalog"; });
@@ -4738,7 +4739,16 @@ TransportStatus MoqtSession::publish_live_objects(const openmoq::publisher::Live
             }
             pace_until(*object_pacing_start, *object_first_media_time_us, object, true);
         }
-        status = sender_by_track[next->track_name].serve(
+        auto& sender = sender_by_track[next->track_name];
+        const auto group_it = last_group_id_by_track.find(next->track_name);
+        if (group_it != last_group_id_by_track.end() &&
+            group_it->second != static_cast<std::uint64_t>(next->group_id)) {
+            status = sender.finish_group(transport_);
+            if (!status.ok) {
+                return status;
+            }
+        }
+        status = sender.serve(
             transport_,
             draft_version,
             alias_it->second,
@@ -4758,6 +4768,7 @@ TransportStatus MoqtSession::publish_live_objects(const openmoq::publisher::Live
         record_published_object(next->track_name,
                                 static_cast<std::uint64_t>(next->group_id),
                                 next->payload.size());
+        last_group_id_by_track[next->track_name] = static_cast<std::uint64_t>(next->group_id);
         if (stop_requested_.load(std::memory_order_acquire)) {
             break;
         }