From 8c05ab92853f84b4feb29906a7e11abba8e976c9 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 27 Oct 2026 23:36:17 -0700 Subject: [PATCH 01/14] chore(quest): claim quest/m1/simulcast-rung-disable.md Quest-Claim: e0203655a73462a1 From c9620ec733a83d7c7b3c86818810b1d7fa25e300 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 27 Oct 2026 23:54:29 -0700 Subject: [PATCH 02/14] feat(obs): disable simulcast rungs the grant cannot fund OBS follows each video encoder's grant through moq-ffi's new MoqRateControl, the same moq_mux::rate::Control the native publishers use, replacing the C++ MoQRate mirror. A rung above a lower encoder sits one track priority below it (MoqVideoInit.lower), drops its packets while disabled (catalog enabled=false via MoqMediaTrackProducer::set_enabled), and resumes at the next keyframe. Co-Authored-By: Claude Opus 5.5 --- cpp/obs/src/moq-output.cpp | 83 ++++++++++++++++++-- cpp/obs/src/moq-output.h | 20 +++-- cpp/obs/src/moq-rate.h | 62 --------------- cpp/obs/test/moq-output-test.cpp | 122 +++++++++++++++++++++++++---- cpp/obs/test/moq-rate-test.cpp | 87 -------------------- rs/moq-ffi/src/bandwidth.rs | 111 ++++++++++++++++++++++++++ rs/moq-ffi/src/media.rs | 65 ++++++++++++--- rs/moq-ffi/src/test.rs | 46 +++++++++++ rs/moq-mux/src/codec/av1/import.rs | 7 ++ rs/moq-mux/src/codec/vp8/import.rs | 7 ++ rs/moq-mux/src/codec/vp9/import.rs | 7 ++ rs/moq-mux/src/import/init.rs | 6 ++ rs/moq-mux/src/import/track.rs | 88 ++++++++++++++++++--- sh/obs/unit.sh | 2 +- 14 files changed, 513 insertions(+), 200 deletions(-) delete mode 100644 cpp/obs/src/moq-rate.h delete mode 100644 cpp/obs/test/moq-rate-test.cpp diff --git a/cpp/obs/src/moq-output.cpp b/cpp/obs/src/moq-output.cpp index 752ea8c3775..519bfd242f1 100644 --- a/cpp/obs/src/moq-output.cpp +++ b/cpp/obs/src/moq-output.cpp @@ -387,9 +387,9 @@ void MoQOutput::Reset() for (auto &[encoder, track] : video_tracks) { if (track.producer) track.producer->finish(); - if (track.rate && track.rate->Target() != track.rate->Max()) { + if (track.rate && track.applied_bps != track.bitrate_bps) { if (OBSEncoder strong = OBSGetStrongRef(track.encoder)) - restore.emplace_back(std::move(strong), track.rate->Max()); + restore.emplace_back(std::move(strong), track.bitrate_bps); } } video_tracks.clear(); @@ -576,8 +576,8 @@ bool MoQOutput::TryGetVideoRate(VideoRate *out) if (!track.rate || !track.reservation) continue; following = true; - rate.target_bps += track.rate->Target(); - rate.configured_bps += track.rate->Max(); + rate.target_bps += track.enabled ? track.applied_bps : 0; + rate.configured_bps += track.bitrate_bps; } } if (following && out) @@ -677,8 +677,21 @@ void MoQOutput::VideoData(struct encoder_packet *packet) // Read before the frame is written, so a retune lands on the encoder's next // frame. obs_encoder_update only flags an active encoder for its own thread. if (track.rate && track.reservation) { - if (auto next = track.rate->Update(track.reservation->rate(), MoQRate::Clock::now())) - SetBitrate(encoder, *next); + if (auto next = track.rate->update(track.reservation)) + Retarget(encoder, track, *next); + } + + // OBS can't stop one encoder of a running output, or force it to keyframe, so a + // disabled rung keeps encoding and its packets are dropped here. It resumes on + // the encoder's next natural keyframe once the grant funds it again. + if (!track.sending) { + if (!track.enabled || !packet->keyframe) + return; + if (auto result = track.producer->set_enabled(true); !result) { + LOG_ERROR("Failed to enable video track: %s", result.error().to_string().c_str()); + return; + } + track.sending = true; } auto timestamp = PacketTimestamp(packet); @@ -695,6 +708,28 @@ void MoQOutput::VideoData(struct encoder_packet *packet) total_bytes_sent += packet->size; } +// NOTE: Caller must hold media_mutex. +void MoQOutput::Retarget(obs_encoder_t *encoder, Track &track, const moq::RateTarget &target) +{ + if (const auto *enabled = std::get_if(&target.get_variant())) { + SetBitrate(encoder, enabled->bitrate_bps); + track.applied_bps = enabled->bitrate_bps; + track.enabled = true; + return; + } + + track.enabled = false; + if (!track.sending) + return; + track.sending = false; + // The lower renditions keep their share; viewers switch to one instead of + // waiting on frames that stopped coming. + if (auto result = track.producer->discontinuity(); !result) + LOG_ERROR("Failed to mark a disabled video track: %s", result.error().to_string().c_str()); + if (auto result = track.producer->set_enabled(false); !result) + LOG_ERROR("Failed to disable video track: %s", result.error().to_string().c_str()); +} + // NOTE: Caller must hold media_mutex. void MoQOutput::Reserve(Track &track) { @@ -766,6 +801,33 @@ void MoQOutput::VideoInit(obs_encoder_t *encoder) hint.optimize_for_latency = true; config.hint = hint; + // Every video encoder on the output encodes the same program, so the encoder with + // the next lower bitrate is this rendition's fallback. That track is published + // first, so this one can sit one priority below it and the allocator funds it first. + obs_encoder_t *lower = nullptr; + int lower_kbps = 0; + for (uint32_t idx = 0; idx < MAX_OUTPUT_VIDEO_ENCODERS; idx++) { + obs_encoder_t *other = obs_output_get_video_encoder2(output, idx); + if (!other || other == encoder) + continue; + OBSDataAutoRelease other_settings = obs_encoder_get_settings(other); + const int kbps = other_settings ? (int)obs_data_get_int(other_settings, "bitrate") : 0; + if (kbps > lower_kbps && kbps < video_bitrate_kbps) { + lower = other; + lower_kbps = kbps; + } + } + std::optional lower_bps; + if (lower) { + if (video_tracks.find(lower) == video_tracks.end()) + VideoInit(lower); + auto below = video_tracks.find(lower); + if (below != video_tracks.end() && below->second.producer) { + config.lower = below->second.producer; + lower_bps = (uint64_t)lower_kbps * 1000ULL; + } + } + auto track = moq::MediaTrackProducer::video(broadcast, moq::MediaTarget::kNamed{}, config); if (!track) { LOG_ERROR("Failed to initialize video track: %s", track.error().to_string().c_str()); @@ -781,7 +843,8 @@ void MoQOutput::VideoInit(obs_encoder_t *encoder) // bitrate drives; a quality target (CQP, CRF) only reserves. const bool dynamic = obs_encoder_get_caps(encoder) & OBS_ENCODER_CAP_DYN_BITRATE; if (dynamic && BitrateDriven(rate_control)) { - entry.rate.emplace(entry.bitrate_bps); + entry.rate = moq::RateControl::init(moq::RatePolicy{entry.bitrate_bps, lower_bps}); + entry.applied_bps = entry.bitrate_bps; entry.encoder = OBSGetWeakRef(encoder); } else { LOG_INFO("Video bitrate stays at %d kbps: the encoder cannot follow the bandwidth grant (%s)", @@ -790,7 +853,11 @@ void MoQOutput::VideoInit(obs_encoder_t *encoder) } video_tracks[encoder] = std::move(entry); - LOG_INFO("Video track initialized (%ux%u, %d kbps)", video_width, video_height, video_bitrate_kbps); + if (lower_bps) + LOG_INFO("Video track initialized (%ux%u, %d kbps, above a %d kbps rendition)", video_width, video_height, + video_bitrate_kbps, lower_kbps); + else + LOG_INFO("Video track initialized (%ux%u, %d kbps)", video_width, video_height, video_bitrate_kbps); } // NOTE: Caller must hold media_mutex. diff --git a/cpp/obs/src/moq-output.h b/cpp/obs/src/moq-output.h index 00ed4ea892c..50220f265cb 100644 --- a/cpp/obs/src/moq-output.h +++ b/cpp/obs/src/moq-output.h @@ -15,7 +15,6 @@ #include #include #include "logger.h" -#include "moq-rate.h" #include "moq-worker.h" class MoQOutput { @@ -68,8 +67,9 @@ class MoQOutput { void CopyLastFailure(int *code, std::string *reason); // The bitrate the video encoders follow the bandwidth grant at, summed over - // every encoder that follows, beside the configured rate that caps it. - // False when no video encoder follows the grant. + // every encoder that follows, beside the configured rate that caps it. A rung + // disabled for lack of bandwidth counts as zero. False when no video encoder + // follows the grant. struct VideoRate { uint64_t target_bps = 0; uint64_t configured_bps = 0; @@ -129,10 +129,17 @@ class MoQOutput { uint64_t bitrate_bps = 0; // Claimed on the first frame after the session connects. std::shared_ptr reservation; - // Video only, when the encoder can retune: the grant it follows, and a + // Video only, when the encoder can retune: the grant it follows, which + // disables it while the grant cannot fund it above a lower rendition, and a // weak ref so Reset() can hand the configured bitrate back. - std::optional rate; + std::shared_ptr rate; OBSWeakEncoder encoder; + // The bitrate the encoder was last told, which Reset() restores when it moved. + uint64_t applied_bps = 0; + // Whether the last target enabled this rendition. + bool enabled = true; + // False from a disable until the keyframe that resumes it, dropping packets. + bool sending = true; }; void VideoInit(obs_encoder_t *encoder); @@ -141,6 +148,9 @@ class MoQOutput { void AudioData(struct encoder_packet *packet); // NOTE: Caller must hold media_mutex. void Reserve(Track &track); + // Apply a rate target to `encoder`'s track: retune it, or stop sending it. + // NOTE: Caller must hold media_mutex. + void Retarget(obs_encoder_t *encoder, Track &track, const moq::RateTarget &target); obs_output_t *output; diff --git a/cpp/obs/src/moq-rate.h b/cpp/obs/src/moq-rate.h deleted file mode 100644 index 4b30ed207fe..00000000000 --- a/cpp/obs/src/moq-rate.h +++ /dev/null @@ -1,62 +0,0 @@ -// SPDX-License-Identifier: GPL-2.0-or-later -#pragma once - -#include -#include -#include -#include - -// Turns a reservation's grant into the bitrate an OBS encoder should run at, -// with the same shape as moq_mux::rate::Control: drops apply at once, raises -// climb at most 25% per second, and moves under 5% are ignored so a jittering -// grant doesn't retune the encoder on every frame. Keep the two in step. -class MoQRate { -public: - using Clock = std::chrono::steady_clock; - - // Starts at `max`, the configured bitrate, which the grant can only lower. - // The floor is a tenth of it: below that the video isn't worth sending. - explicit MoQRate(uint64_t max) : max(max), min(max / 10), target(max) {} - - uint64_t Target() const { return target; } - uint64_t Max() const { return max; } - - // Feed a grant, returning the new target when it moved enough to apply. - // A missing grant (no estimate) holds the target: losing the estimate is not - // evidence the uplink got better. - std::optional Update(std::optional grant, Clock::time_point now) - { - if (!grant) - return std::nullopt; - - const uint64_t desired = std::clamp(*grant, min, max); - uint64_t next = desired; - if (desired > target && applied) { - const double elapsed = std::chrono::duration(now - *applied).count(); - const auto grown = static_cast(static_cast(target) * - (1.0 + RAMP * std::max(elapsed, 0.0))); - next = std::clamp(std::min(grown, desired), min, max); - } - - // Hysteresis leaves `applied` alone, so suppressed raises keep accruing - // ramp allowance. The final step onto `max` is exempt, or a recovery - // would stall a few percent under the configured rate for good. - const bool settling = next > target && next == max; - const uint64_t diff = next > target ? next - target : target - next; - if (!settling && static_cast(diff) < static_cast(target) * HYSTERESIS) - return std::nullopt; - - target = next; - applied = now; - return next; - } - -private: - static constexpr double HYSTERESIS = 0.05; - static constexpr double RAMP = 0.25; - - uint64_t max; - uint64_t min; - uint64_t target; - std::optional applied; -}; diff --git a/cpp/obs/test/moq-output-test.cpp b/cpp/obs/test/moq-output-test.cpp index 002164e8895..a0579408c03 100644 --- a/cpp/obs/test/moq-output-test.cpp +++ b/cpp/obs/test/moq-output-test.cpp @@ -77,6 +77,9 @@ const std::vector g_opus_head = TestOpusHead(); obs_encoder_t *const VIDEO_ENCODER = reinterpret_cast(0x2); obs_encoder_t *const AUDIO_ENCODER = reinterpret_cast(0x3); +// A second, 2 Mbps video encoder below VIDEO_ENCODER's 8 Mbps, while g_ladder is set. +obs_encoder_t *const LOW_ENCODER = reinterpret_cast(0x5); +std::atomic g_ladder{false}; } // namespace extern "C" { @@ -114,7 +117,9 @@ void obs_data_release(obs_data_t *) {} obs_encoder_t *obs_output_get_video_encoder2(const obs_output_t *, size_t idx) { - return idx == 0 ? VIDEO_ENCODER : nullptr; + if (idx == 0) + return VIDEO_ENCODER; + return idx == 1 && g_ladder ? LOW_ENCODER : nullptr; } bool obs_output_begin_data_capture(obs_output_t *, uint32_t) @@ -163,10 +168,11 @@ const char *obs_encoder_get_codec(const obs_encoder_t *encoder) } // VideoInit reads coded size and CBR from the encoder. Any new libobs call in -// src/moq-output.cpp needs a stub here or `just obs ci` fails at link. -obs_data_t *obs_encoder_get_settings(const obs_encoder_t *) +// src/moq-output.cpp needs a stub here or `just obs ci` fails at link. The +// settings are the encoder itself, so each reports its own bitrate. +obs_data_t *obs_encoder_get_settings(const obs_encoder_t *encoder) { - return reinterpret_cast(0x4); + return reinterpret_cast(const_cast(encoder)); } uint32_t obs_encoder_get_width(const obs_encoder_t *) @@ -184,9 +190,9 @@ const char *obs_data_get_string(obs_data_t *, const char *) return g_rate_control.c_str(); } -long long obs_data_get_int(obs_data_t *, const char *) +long long obs_data_get_int(obs_data_t *settings, const char *) { - return 8000; + return settings == reinterpret_cast(LOW_ENCODER) ? 2000 : 8000; } // The settings a retune builds carry only the bitrate, so the stub is that bitrate. @@ -298,6 +304,7 @@ void reset(std::string url) g_settings_error.clear(); g_rate_control = "CBR"; g_encoder_caps = OBS_ENCODER_CAP_DYN_BITRATE; + g_ladder = false; { std::lock_guard lock(g_updates_mutex); g_updates.clear(); @@ -344,8 +351,11 @@ std::shared_ptr relayBroadcast(TestRelay &relay) return TestOk(available.get(), "available"); } -// The broadcast's catalog, once it has video and audio. -std::optional relayCatalog(const std::shared_ptr &broadcast) +// The broadcast's first catalog `ready` accepts, by default one with video and audio. +std::optional relayCatalog(const std::shared_ptr &broadcast, + std::function ready = [](const moq::Catalog &catalog) { + return !catalog.video.empty() && !catalog.audio.empty(); + }) { if (!broadcast) return std::nullopt; @@ -357,7 +367,7 @@ std::optional relayCatalog(const std::shared_ptrvideo.empty() && !catalog->audio.empty()) + if (ready(*catalog)) return catalog; } } @@ -433,14 +443,31 @@ struct MoQOutputStatsTest { } return true; } - // Drive every following video track to its floor, as a collapsed uplink would. - static void Collapse(MoQOutput &output) + // Apply `target` to `encoder`'s track, as a grant the track's rate control returned. + static void Retarget(MoQOutput &output, obs_encoder_t *encoder, const moq::RateTarget &target) { std::lock_guard lock(output.media_mutex); - for (auto &[encoder, track] : output.video_tracks) { - if (track.rate) - track.rate->Update(1, MoQRate::Clock::now()); + auto it = output.video_tracks.find(encoder); + if (it != output.video_tracks.end()) + output.Retarget(encoder, it->second, target); + } + // The name and publisher priority of `encoder`'s track, or nothing without one. + static std::optional> Published(MoQOutput &output, obs_encoder_t *encoder) + { + std::shared_ptr broadcast; + std::shared_ptr producer; + { + std::lock_guard lock(output.media_mutex); + auto it = output.video_tracks.find(encoder); + if (it == output.video_tracks.end() || !it->second.producer) + return std::nullopt; + broadcast = output.broadcast; + producer = it->second.producer; } + auto name = TestOk(producer->demand(), "demand")->name(); + auto consumer = TestOk(broadcast->consume(), "consume"); + auto track = TestOk(consumer->subscribe_track(name, std::nullopt).get(), "subscribe_track"); + return std::make_pair(name, track->info().priority); } }; @@ -619,7 +646,8 @@ int main() auto keyframe = TestH264Keyframe(); auto video = videoPacket(keyframe, 0); o.Data(&video); - MoQOutputStatsTest::Collapse(o); + // The floor a collapsed uplink drives it to. + MoQOutputStatsTest::Retarget(o, VIDEO_ENCODER, moq::RateTarget::kEnabled{800'000}); MoQOutput::VideoRate rate; CHECK(o.TryGetVideoRate(&rate)); CHECK(rate.target_bps == 800'000); @@ -630,6 +658,70 @@ int main() } printf("stop restores the configured bitrate: ok\n"); + // Two video encoders form a ladder: the higher rung sits one priority below the + // lower, and a disabled rung leaves the catalog disabled, dropping its packets + // until a keyframe after the grant funds it again. No encoder follows here, so + // only the targets the test applies move it. + { + TestRelay relay; + reset(relay.Url()); + g_ladder = true; + g_encoder_caps = 0; + MoQOutput o(nullptr, OUTPUT); + CHECK(o.Start()); + CHECK(WaitFor([&] { return o.IsLiveSession(); })); + + auto keyframe = TestH264Keyframe(); + std::vector delta = {0x00, 0x00, 0x00, 0x01, 0x41, 0x9a, 0x00, 0x33}; + auto video = videoPacket(keyframe, 0); + o.Data(&video); + size_t sent = keyframe.size(); + CHECK(o.GetTotalBytes() == sent); + + const auto high = MoQOutputStatsTest::Published(o, VIDEO_ENCODER); + const auto low = MoQOutputStatsTest::Published(o, LOW_ENCODER); + CHECK(high.has_value() && low.has_value()); + if (high && low) { + CHECK(high->second == low->second - 1); + auto broadcast = relayBroadcast(relay); + + MoQOutputStatsTest::Retarget(o, VIDEO_ENCODER, moq::RateTarget::kDisabled{}); + auto disabled = relayCatalog(broadcast, [&](const moq::Catalog &catalog) { + auto it = catalog.video.find(high->first); + return it != catalog.video.end() && !it->second.enabled; + }); + CHECK(disabled.has_value()); + if (disabled) { + auto lower = disabled->video.find(low->first); + CHECK(lower != disabled->video.end() && lower->second.enabled); + } + + auto dropped = videoPacket(delta, 1); + dropped.keyframe = false; + o.Data(&dropped); + auto skipped = videoPacket(keyframe, 2); + o.Data(&skipped); + CHECK(o.GetTotalBytes() == sent); + + // Funded again, it still waits for a keyframe to resume on. + MoQOutputStatsTest::Retarget(o, VIDEO_ENCODER, moq::RateTarget::kEnabled{4'000'000}); + auto early = videoPacket(delta, 3); + early.keyframe = false; + o.Data(&early); + CHECK(o.GetTotalBytes() == sent); + auto resumed = videoPacket(keyframe, 4); + o.Data(&resumed); + sent += keyframe.size(); + CHECK(o.GetTotalBytes() == sent); + CHECK(relayCatalog(broadcast, [&](const moq::Catalog &catalog) { + auto it = catalog.video.find(high->first); + return it != catalog.video.end() && it->second.enabled; + }).has_value()); + } + o.Stop(); + } + printf("a ladder rung disables, drops packets, and resumes on a keyframe: ok\n"); + // Stop drains the session, even when the output is destroyed straight after: a // large frame written just before it still reaches the relay, and the track then // ends cleanly instead of being cut off with the session. diff --git a/cpp/obs/test/moq-rate-test.cpp b/cpp/obs/test/moq-rate-test.cpp deleted file mode 100644 index c14155f18be..00000000000 --- a/cpp/obs/test/moq-rate-test.cpp +++ /dev/null @@ -1,87 +0,0 @@ -// SPDX-License-Identifier: GPL-2.0-or-later -// -// The cases moq_mux::rate::Control is tested with, so the two policies agree. -#include "moq-rate.h" - -#include -#include - -using namespace std::chrono_literals; - -namespace { -int g_failures = 0; - -#define CHECK(cond) \ - do { \ - if (!(cond)) { \ - fprintf(stderr, "FAIL %s:%d: %s\n", __FILE__, __LINE__, #cond); \ - g_failures++; \ - } \ - } while (0) - -const MoQRate::Clock::time_point START{}; -} // namespace - -int main() -{ - { - // Starts at the configured rate and drops at once. - MoQRate rate(4'000'000); - CHECK(rate.Target() == 4'000'000); - CHECK(rate.Update(2'000'000, START) == 2'000'000); - } - { - // A missing grant holds, however long it lasts. - MoQRate rate(4'000'000); - rate.Update(2'000'000, START); - CHECK(!rate.Update(std::nullopt, START + 10s)); - CHECK(rate.Target() == 2'000'000); - } - { - // Never above the configured rate, never below a tenth of it. - MoQRate rate(4'000'000); - CHECK(!rate.Update(100'000'000, START)); - CHECK(rate.Update(1, START) == 400'000); - } - { - // One second of ramp allows 1M -> 1.25M, not the whole grant. - MoQRate rate(4'000'000); - rate.Update(1'000'000, START); - CHECK(rate.Update(4'000'000, START + 1s) == 1'250'000); - } - { - // Each 100ms tick alone is under the 5% deadband; suppression must keep - // the ramp anchor so the raise lands on the second tick. - MoQRate rate(4'000'000); - rate.Update(1'000'000, START); - CHECK(!rate.Update(4'000'000, START + 100ms)); - CHECK(rate.Update(4'000'000, START + 200ms) == 1'050'000); - } - { - // A recovery lands exactly on the configured rate, then stops retuning. - MoQRate rate(4'000'000); - rate.Update(1'000'000, START); - std::optional last; - for (int tick = 1; tick <= 200; tick++) { - if (auto next = rate.Update(4'000'000, START + tick * 100ms)) - last = next; - } - CHECK(last == 4'000'000); - CHECK(!rate.Update(4'000'000, START + 60s)); - } - { - // Small moves either way stay inside the deadband. - MoQRate rate(4'000'000); - rate.Update(2'000'000, START); - CHECK(!rate.Update(2'010'000, START + 100ms)); - CHECK(!rate.Update(1'960'000, START + 1s)); - CHECK(rate.Update(1'600'000, START + 2s) == 1'600'000); - } - - if (g_failures) { - fprintf(stderr, "%d rate check(s) failed\n", g_failures); - return EXIT_FAILURE; - } - std::puts("rate follows the grant like moq_mux::rate::Control: ok"); - return EXIT_SUCCESS; -} diff --git a/rs/moq-ffi/src/bandwidth.rs b/rs/moq-ffi/src/bandwidth.rs index bc0d22e1f61..1a0803ff16b 100644 --- a/rs/moq-ffi/src/bandwidth.rs +++ b/rs/moq-ffi/src/bandwidth.rs @@ -88,6 +88,84 @@ impl MoqReservation { } } +/// How a [`MoqRateControl`] maps a reservation's grant onto the bitrate a sender produces. +#[derive(Clone, Debug, uniffi::Record)] +pub struct MoqRatePolicy { + /// The most the sender produces, normally its configured bitrate. The grant only lowers it. + pub max_bps: u64, + /// The next lower rendition's `max_bps`, when this one is encoded beside lower renditions + /// of the same source. A grant below `(max_bps + 2 * lower_bps) / 3` then disables this + /// one, since the lower rendition serves viewers better than a starved copy of this one. + /// Absent for the lowest or only rendition, which never disables. + #[uniffi(default = None)] + pub lower_bps: Option, +} + +/// What a sender following a [`MoqRateControl`] should be doing. +#[derive(Clone, Copy, Debug, PartialEq, Eq, uniffi::Enum)] +pub enum MoqRateTarget { + /// Produce at this bitrate, and advertise the rendition as enabled. + Enabled { + /// The bitrate to produce at, in bits per second. + bitrate_bps: u64, + }, + /// Stop sending, and advertise the rendition as disabled until a later target enables it. + Disabled, +} + +impl From for MoqRateTarget { + fn from(target: moq_mux::rate::Target) -> Self { + match target { + moq_mux::rate::Target::Enabled(rate) => Self::Enabled { + bitrate_bps: rate.as_bps(), + }, + moq_mux::rate::Target::Disabled => Self::Disabled, + } + } +} + +/// Turns a reservation's grant into a sender's target, the way moq-ffi's own encoders follow it. +/// +/// Drops apply at once, raises climb at most 25% per second, and moves under 5% are ignored, +/// so a jittering grant doesn't retune the sender on every frame. For a sender that encodes +/// on its own, such as a pre-encoded track whose encoder the caller retunes: call +/// [`update`](Self::update) on every frame and apply what it returns. +#[derive(uniffi::Object)] +pub struct MoqRateControl { + inner: std::sync::Mutex, +} + +#[uniffi::export] +impl MoqRateControl { + /// Start at the policy's `max_bps`, until a grant says otherwise. + #[uniffi::constructor] + pub fn new(policy: MoqRatePolicy) -> Arc { + let mut native = moq_mux::rate::Policy::new(moq_net::bandwidth::Rate::from_bps(policy.max_bps)); + native.lower = policy.lower_bps.map(moq_net::bandwidth::Rate::from_bps); + Arc::new(Self { + inner: std::sync::Mutex::new(moq_mux::rate::Control::new(native)), + }) + } + + /// What the sender should currently be doing. + pub fn target(&self) -> MoqRateTarget { + self.inner.lock().unwrap().target().into() + } + + /// Feed `reservation`'s current grant, returning the new target when it moved enough to + /// apply, or `None` to keep doing what the last target said. + /// + /// A rendition only disables while a session probes for the reservation, since nothing + /// else could raise the grant to resume it. A missing grant (no estimate) holds the target. + pub fn update(&self, reservation: &MoqReservation) -> Option { + let _guard = crate::ffi::enter(); + let grant = reservation.inner.consume(); + let mut control = self.inner.lock().unwrap(); + control.set_probed(grant.probed()); + control.update(grant.peek(), moq_clock::now()).map(Into::into) + } +} + #[cfg(test)] mod tests { use super::*; @@ -203,6 +281,39 @@ mod tests { assert_eq!(reserved.rate(), Some(1_000_000)); } + /// A rung follows its reservation's grant down, but never disables while no session + /// probes for it, since nothing could then raise the grant to resume it. A missing + /// grant holds the target. + #[test] + fn a_rate_control_disables_only_while_probed() { + let estimate = moq_net::bandwidth::Producer::new(); + let bandwidth = allocator(&estimate); + let broadcast = MoqBroadcastProducer::new().unwrap(); + + let track = named_track(&broadcast, "a"); + let _sub = track.consume(None).unwrap(); + let reserved = bandwidth.reserve(&track.demand().unwrap(), 6_000_000); + let control = MoqRateControl::new(MoqRatePolicy { + max_bps: 6_000_000, + lower_bps: Some(2_000_000), + }); + assert_eq!(control.target(), MoqRateTarget::Enabled { bitrate_bps: 6_000_000 }); + assert_eq!(control.update(&reserved), None); + + // Below the 3.33 Mbps boundary, but unprobed: follow it like any sender. + estimate + .set(Some(moq_net::bandwidth::Rate::from_bps(3_000_000))) + .unwrap(); + assert_eq!( + control.update(&reserved), + Some(MoqRateTarget::Enabled { bitrate_bps: 3_000_000 }) + ); + + estimate.set(None).unwrap(); + assert_eq!(control.update(&reserved), None); + assert_eq!(control.target(), MoqRateTarget::Enabled { bitrate_bps: 3_000_000 }); + } + #[cfg(all(feature = "video", not(target_arch = "wasm32")))] #[test] fn a_share_wakes_a_parked_thread() { diff --git a/rs/moq-ffi/src/media.rs b/rs/moq-ffi/src/media.rs index 4bb3ce1f75a..7e0e51107ef 100644 --- a/rs/moq-ffi/src/media.rs +++ b/rs/moq-ffi/src/media.rs @@ -329,6 +329,11 @@ pub struct MoqVideoInit { /// Catalog fields the stream cannot reveal itself. #[uniffi(default = None)] pub hint: Option, + /// The next lower video rendition of the same source, when this one is published beside + /// it. This track then sits one priority below it, so a bandwidth allocator funds the + /// lower rendition first. + #[uniffi(default = None)] + pub lower: Option>, } /// Shares one source's timestamp offset across containers without keeping its broadcast open. @@ -490,14 +495,23 @@ impl From for moq_mux::import::AudioInit { } } -impl From for moq_mux::import::VideoInit { - fn from(init: MoqVideoInit) -> Self { - let mut out = moq_mux::import::VideoInit::new(init.format.into(), init.data); - out.label = init.label; - if let Some(hint) = init.hint { +impl MoqVideoInit { + /// The importer's init, at one priority below the [`lower`](Self::lower) rendition's. + fn resolve(self) -> Result { + let mut out = moq_mux::import::VideoInit::new(self.format.into(), self.data); + out.label = self.label; + if let Some(hint) = self.hint { out.hint = hint.into(); } - out + if let Some(lower) = self.lower { + let guard = lower.inner.lock().unwrap(); + let below = guard.as_ref().ok_or(MoqError::Closed)?; + let priority = below.video.ok_or_else(|| MoqError::Codec { + reason: "the lower rendition is not a video track".into(), + })?; + out.priority = priority.saturating_sub(1); + } + Ok(out) } } @@ -801,6 +815,8 @@ struct MediaProducer { import: Box, /// Subscriber demand (name/used/unused) for the one track this publishes. demand: moq_net::track::Demand, + /// A video track's priority, which a higher rung takes one below; `None` for audio. + video: Option, } /// An importer for a container, which may publish several tracks. @@ -912,7 +928,7 @@ impl MoqMediaTrackProducer { reason: format!("init failed: {err}"), } })?; - MoqMediaTrackProducer::from_import(import, &state.children) + MoqMediaTrackProducer::from_import(import, None, &state.children) }) } } @@ -930,8 +946,9 @@ impl MoqMediaTrackProducer { init: MoqVideoInit, ) -> Result, MoqError> { let _guard = crate::ffi::enter(); + let init = init.resolve()?; broadcast.with_state(|state| { - let init: moq_mux::import::VideoInit = init.into(); + let priority = init.priority; let request = target.resolve(&state.broadcast, &init.format)?; let import = moq_mux::import::Track::video(request, state.catalog.reserve(), init).map_err(|err| { @@ -939,7 +956,7 @@ impl MoqMediaTrackProducer { reason: format!("init failed: {err}"), } })?; - MoqMediaTrackProducer::from_import(import, &state.children) + MoqMediaTrackProducer::from_import(import, Some(priority), &state.children) }) } } @@ -982,8 +999,8 @@ impl MoqMediaTrackStreamProducer { init: MoqVideoInit, ) -> Result, MoqError> { let _guard = crate::ffi::enter(); + let init = init.resolve()?; broadcast.with_state(|state| { - let init: moq_mux::import::VideoInit = init.into(); let request = target.resolve(&state.broadcast, &init.format)?; let import = @@ -1004,13 +1021,19 @@ impl MoqMediaTrackStreamProducer { } impl MoqMediaTrackProducer { - /// Wrap a single-codec importer, capturing the demand handle its track exposes. - fn from_import(import: moq_mux::import::Track, children: &Children) -> Result, MoqError> { + /// Wrap a single-codec importer, capturing the demand handle its track exposes and, for + /// video, the priority it was published at. + fn from_import( + import: moq_mux::import::Track, + video: Option, + children: &Children, + ) -> Result, MoqError> { let demand = import.demand(); children.adopt(Arc::new(Self { inner: std::sync::Mutex::new(Some(MediaProducer { import: Box::new(import), demand, + video, })), children: children.clone(), })) @@ -1062,6 +1085,24 @@ impl MoqMediaTrackProducer { Ok(()) } + /// Advertise whether this video rendition is being produced, as its catalog `enabled`. + /// + /// Disable it when its sender stops for lack of bandwidth (a + /// [`MoqRateTarget::Disabled`](crate::bandwidth::MoqRateTarget::Disabled)) and mark a + /// [`discontinuity`](Self::discontinuity), so viewers select a lower rendition instead of + /// waiting for frames. Enable it again just before writing the keyframe that resumes it. + /// The rest of the rendition, its maximum bitrate included, stays as advertised. An audio + /// track refuses. + pub fn set_enabled(&self, enabled: bool) -> Result<(), MoqError> { + let _guard = crate::ffi::enter(); + let mut guard = self.inner.lock().unwrap(); + let media = guard.as_mut().ok_or(MoqError::Closed)?; + media.import.set_enabled(enabled).map_err(|err| MoqError::Codec { + reason: format!("set enabled failed: {err}"), + })?; + Ok(()) + } + /// Draw a group boundary here. /// /// Audio has no boundary of its own (every packet is independently decodable), so this is the diff --git a/rs/moq-ffi/src/test.rs b/rs/moq-ffi/src/test.rs index a2a5974cbdf..736b6153d6a 100644 --- a/rs/moq-ffi/src/test.rs +++ b/rs/moq-ffi/src/test.rs @@ -215,6 +215,7 @@ fn video_init(format: MoqVideoFormat, data: Vec) -> MoqVideoInit { data, label: None, hint: None, + lower: None, } } @@ -2756,6 +2757,51 @@ async fn video_publish_named_track() { assert!(catalog.video.contains_key("hd"), "catalog: {:?}", catalog.video.keys()); } +/// A pre-encoded rung above a lower rendition sits one priority below it, so an allocator +/// funds the lower one first, and disables in the catalog without leaving it. +#[tokio::test] +async fn video_rung_sits_below_its_lower_and_disables() { + let broadcast = MoqBroadcastProducer::new().unwrap(); + let consumer = broadcast.consume().unwrap(); + let catalog_consumer = MoqMediaCatalogConsumer::subscribe(&consumer).await.unwrap(); + + let named = |name: &str| MoqMediaTarget::Named { + name: Some(name.into()), + }; + let low = MoqMediaTrackProducer::video(&broadcast, named("low"), video_init(MoqVideoFormat::Avc3, h264_init())) + .unwrap(); + let high = MoqMediaTrackProducer::video( + &broadcast, + named("high"), + MoqVideoInit { + lower: Some(low.clone()), + ..video_init(MoqVideoFormat::Avc3, h264_init()) + }, + ) + .unwrap(); + + let priority = |name: &str| { + let consumer = consumer.clone(); + let name = name.to_string(); + async move { consumer.subscribe_track(name, None).await.unwrap().info().priority } + }; + assert_eq!(priority("low").await, hang::catalog::PRIORITY.video); + assert_eq!(priority("high").await, hang::catalog::PRIORITY.video - 1); + + high.set_enabled(false).unwrap(); + let disabled = loop { + let catalog = tokio::time::timeout(TIMEOUT, catalog_consumer.next()) + .await + .expect("timed out waiting for catalog") + .unwrap() + .expect("expected a catalog"); + if catalog.video.get("high").is_some_and(|high| !high.enabled) { + break catalog; + } + }; + assert!(disabled.video["low"].enabled); +} + /// The raw-stream producer frames Annex-B written in arbitrary chunks, and /// `finish` publishes the last access unit even with no start code after it. #[tokio::test] diff --git a/rs/moq-mux/src/codec/av1/import.rs b/rs/moq-mux/src/codec/av1/import.rs index f5f568b384c..a9896756a72 100644 --- a/rs/moq-mux/src/codec/av1/import.rs +++ b/rs/moq-mux/src/codec/av1/import.rs @@ -233,6 +233,13 @@ impl Import { self.track.discontinuity() } + /// Advertise whether this rendition is being produced, as the catalog's + /// [`enabled`](hang::catalog::VideoConfig::enabled): a publisher that stops encoding it + /// disables it, so viewers select another rendition rather than wait for frames. + pub fn set_enabled(&mut self, enabled: bool) -> Result<()> { + self.catalog.set_enabled(&mut self.track, enabled) + } + /// Close the current group and open the next one at `sequence`. pub fn seek(&mut self, sequence: u64) -> Result<()> { self.track.seek(sequence)?; diff --git a/rs/moq-mux/src/codec/vp8/import.rs b/rs/moq-mux/src/codec/vp8/import.rs index 4d96d603979..0d1c17ae208 100644 --- a/rs/moq-mux/src/codec/vp8/import.rs +++ b/rs/moq-mux/src/codec/vp8/import.rs @@ -120,6 +120,13 @@ impl Import { self.track.discontinuity() } + /// Advertise whether this rendition is being produced, as the catalog's + /// [`enabled`](hang::catalog::VideoConfig::enabled): a publisher that stops encoding it + /// disables it, so viewers select another rendition rather than wait for frames. + pub fn set_enabled(&mut self, enabled: bool) -> crate::Result<()> { + self.catalog.set_enabled(&mut self.track, enabled) + } + /// Close the current group and open the next one at `sequence`. pub fn seek(&mut self, sequence: u64) -> crate::Result<()> { self.track.seek(sequence)?; diff --git a/rs/moq-mux/src/codec/vp9/import.rs b/rs/moq-mux/src/codec/vp9/import.rs index e01ec13f99b..b33ca72d003 100644 --- a/rs/moq-mux/src/codec/vp9/import.rs +++ b/rs/moq-mux/src/codec/vp9/import.rs @@ -121,6 +121,13 @@ impl Import { self.track.discontinuity() } + /// Advertise whether this rendition is being produced, as the catalog's + /// [`enabled`](hang::catalog::VideoConfig::enabled): a publisher that stops encoding it + /// disables it, so viewers select another rendition rather than wait for frames. + pub fn set_enabled(&mut self, enabled: bool) -> crate::Result<()> { + self.catalog.set_enabled(&mut self.track, enabled) + } + /// Close the current group and open the next one at `sequence`. pub fn seek(&mut self, sequence: u64) -> crate::Result<()> { self.track.seek(sequence)?; diff --git a/rs/moq-mux/src/import/init.rs b/rs/moq-mux/src/import/init.rs index 954460340c2..5f48f0b3e1a 100644 --- a/rs/moq-mux/src/import/init.rs +++ b/rs/moq-mux/src/import/init.rs @@ -52,6 +52,11 @@ pub struct VideoInit { pub label: Option, /// Catalog fields the stream cannot reveal itself. pub hint: VideoHint, + /// The track's priority, which a [`bandwidth::Allocator`](moq_net::bandwidth::Allocator) + /// funds in order. A rendition published above a lower one of the same source sits one + /// below it, so the lower one is funded first. Defaults to + /// [`PRIORITY.video`](hang::catalog::PRIORITY). + pub priority: i8, } impl VideoInit { @@ -62,6 +67,7 @@ impl VideoInit { data: data.into(), label: None, hint: VideoHint::default(), + priority: hang::catalog::PRIORITY.video, } } } diff --git a/rs/moq-mux/src/import/track.rs b/rs/moq-mux/src/import/track.rs index 227d8fe09e0..a8aacc154f1 100644 --- a/rs/moq-mux/src/import/track.rs +++ b/rs/moq-mux/src/import/track.rs @@ -134,6 +134,15 @@ fn build_av1( Ok((split, import)) } +/// An audio format handed to a video-only operation. +fn not_video(format: AudioFormat) -> crate::Error { + crate::Error::WrongKind { + format: format.to_string(), + actual: "audio", + wanted: "video", + } +} + enum TrackKind { /// H.264 avc3 (Annex-B, inline SPS/PPS). The split owns byte parsing; the /// import publishes. @@ -268,11 +277,7 @@ impl Track { ) -> Result { use hang::catalog::VideoCodec; - let track = request.accept(track_info( - &reserved, - hang::catalog::PRIORITY.video, - &init.hint.container, - )?); + let track = request.accept(track_info(&reserved, init.priority, &init.hint.container)?); let data = init.data.as_ref(); let kind = match init.format { VideoFormat::Avc1 => { @@ -457,6 +462,30 @@ impl Track { import!(&mut self.kind, import => import.discontinuity()) } + /// Advertise whether this video rendition is being produced, as the catalog's + /// [`enabled`](hang::catalog::VideoConfig::enabled). + /// + /// Disable it when the sender stops for lack of bandwidth (a + /// [`rate::Target::Disabled`](crate::rate::Target::Disabled)) and mark the + /// [`discontinuity`](Self::discontinuity), so viewers select a lower rendition instead + /// of waiting for frames. Enable it again with the keyframe that resumes it. Audio has + /// no lower rendition to fall back to, so an audio track refuses with + /// [`Error::WrongKind`](crate::Error::WrongKind). + pub fn set_enabled(&mut self, enabled: bool) -> Result<()> { + match &mut self.kind { + TrackKind::Avc3 { import, .. } | TrackKind::Avc1 { import, .. } => import.set_enabled(enabled)?, + TrackKind::Hev1 { import, .. } | TrackKind::Hvc1 { import, .. } => import.set_enabled(enabled)?, + TrackKind::Av01 { import, .. } => import.set_enabled(enabled)?, + TrackKind::Vp8(import) => import.set_enabled(enabled)?, + TrackKind::Vp9(import) => import.set_enabled(enabled)?, + TrackKind::Aac(_) => return Err(not_video(AudioFormat::Aac)), + TrackKind::Opus(_) => return Err(not_video(AudioFormat::Opus)), + TrackKind::Mp3(_) => return Err(not_video(AudioFormat::Mp3)), + TrackKind::Flac(_) => return Err(not_video(AudioFormat::Flac)), + } + Ok(()) + } + /// Close the current group and open the next one at `sequence`. pub fn seek(&mut self, sequence: u64) -> Result<()> { self.group_start = None; @@ -572,11 +601,7 @@ impl TrackStream { reserved: crate::catalog::Reserved, init: VideoInit, ) -> Result { - let track = request.accept(track_info( - &reserved, - hang::catalog::PRIORITY.video, - &init.hint.container, - )?); + let track = request.accept(track_info(&reserved, init.priority, &init.hint.container)?); let hint = video_hint(&init, None); // Only the self-delimiting codecs can be recovered from a raw byte stream. let kind = match init.format { @@ -1150,6 +1175,49 @@ mod tests { let video = consumer.track("video").unwrap().info().await.unwrap(); assert_eq!(video.priority, PRIORITY.video); + + // A higher rung asks for its own priority, below the rendition it sits above. + let request = broadcast.reserve_track("top").unwrap(); + let _top = Track::video( + request, + catalog.reserve(), + VideoInit { + priority: PRIORITY.video - 1, + ..VideoInit::new(VideoFormat::Vp8, Vec::new()) + }, + ) + .unwrap(); + let top = consumer.track("top").unwrap().info().await.unwrap(); + assert_eq!(top.priority, PRIORITY.video - 1); + } + + /// Disabling a video rendition flips its catalog `enabled` and keeps the rest of it; + /// audio has no lower rendition to fall back to, so it refuses. + #[tokio::test(start_paused = true)] + async fn video_disables_and_audio_refuses() { + let (broadcast, catalog) = new_broadcast(); + let request = broadcast.reserve_track("video").unwrap(); + let mut video = Track::video(request, catalog.reserve(), VideoInit::new(VideoFormat::Vp8, Vec::new())).unwrap(); + + video.set_enabled(false).unwrap(); + let rendition = catalog.snapshot().video.renditions.get("video").cloned().unwrap(); + assert!(!rendition.enabled); + assert_eq!(rendition.codec.to_string(), "vp8"); + + video.set_enabled(true).unwrap(); + assert!(catalog.snapshot().video.renditions["video"].enabled); + + let request = broadcast.reserve_track("audio").unwrap(); + let mut audio = Track::audio( + request, + catalog.reserve(), + AudioInit::new(AudioFormat::Opus, opus_head()), + ) + .unwrap(); + assert!(matches!( + audio.set_enabled(false), + Err(crate::Error::WrongKind { actual: "audio", .. }) + )); } /// An audio format with no init bytes errors up front (audio can't resolve its config from frames), diff --git a/sh/obs/unit.sh b/sh/obs/unit.sh index 6a21fb5e4a9..46ba0649fe7 100755 --- a/sh/obs/unit.sh +++ b/sh/obs/unit.sh @@ -64,7 +64,7 @@ for name in moq-audio-encoder moq-output moq-settings moq-source obs-moq; do -o "$out/$name-test" "${sources[@]}" "$out/moq.o" $moq_libs TSAN_OPTIONS="halt_on_error=1 suppressions=$PWD/test/tsan.supp" "$out/$name-test" done -for name in moq-quality-defaults moq-dock-stop moq-error moq-encoder-latency moq-dial moq-spark moq-rate; do +for name in moq-quality-defaults moq-dock-stop moq-error moq-encoder-latency moq-dial moq-spark; do "$cxx" "${flags[@]}" "${includes[@]}" \ -o "$out/$name-test" "test/$name-test.cpp" TSAN_OPTIONS="halt_on_error=1 suppressions=$PWD/test/tsan.supp" "$out/$name-test" From fc15a13e01877256923d6fa88be6d6d7d2c8887c Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 28 Oct 2026 00:14:19 -0700 Subject: [PATCH 03/14] feat(bindings): wrap MoqRateControl and pre-encoded rung disable Python, Swift, and Go wrap MoqRateControl, MediaTrackProducer.set_enabled, and the video init's lower rendition; Kotlin and Dart alias them. The docs describe the pre-encoded path and the OBS ladder, and the quest is rewritten to the remaining shaped-uplink proof, which requires BBR3 aging. Co-Authored-By: Claude Opus 5.5 --- cpp/obs/src/moq-output.cpp | 15 +- cpp/obs/test/moq-output-test.cpp | 14 +- dart/moq/lib/src/aliases.dart | 9 + dart/moq_ffi/lib/src/moq.dart | 477 +++++++++++++++++- doc/bin/obs.md | 7 + doc/lib/cpp/index.md | 2 +- doc/lib/dart/index.md | 2 +- doc/lib/go/index.md | 2 +- doc/lib/kt/index.md | 2 +- doc/lib/py/index.md | 2 +- doc/lib/swift/index.md | 2 +- go/wrapper/MIGRATE.md | 4 + go/wrapper/media/media.go | 37 +- go/wrapper/session.go | 38 ++ .../kotlin/dev/moq/Aliases.kt | 6 + py/moq-rs/moq/__init__.py | 13 +- py/moq-rs/moq/media.py | 51 +- py/moq-rs/moq/session.py | 29 ++ quest/m1/README.md | 2 +- quest/m1/simulcast-rung-disable.md | 70 ++- quest/m2/obs-multitrack.md | 10 +- quest/m2/probe/README.md | 2 +- rs/moq-ffi/src/test.rs | 4 +- swift/Sources/Moq/Bandwidth.swift | 30 ++ swift/Sources/Moq/Media.swift | 25 +- 25 files changed, 775 insertions(+), 80 deletions(-) diff --git a/cpp/obs/src/moq-output.cpp b/cpp/obs/src/moq-output.cpp index 519bfd242f1..9b4b21f110b 100644 --- a/cpp/obs/src/moq-output.cpp +++ b/cpp/obs/src/moq-output.cpp @@ -810,8 +810,15 @@ void MoQOutput::VideoInit(obs_encoder_t *encoder) obs_encoder_t *other = obs_output_get_video_encoder2(output, idx); if (!other || other == encoder) continue; - OBSDataAutoRelease other_settings = obs_encoder_get_settings(other); - const int kbps = other_settings ? (int)obs_data_get_int(other_settings, "bitrate") : 0; + // An encoder already publishing may have been retuned, so read the bitrate it + // was configured with. + int kbps = 0; + if (auto known = video_tracks.find(other); known != video_tracks.end()) { + kbps = (int)(known->second.bitrate_bps / 1000); + } else { + OBSDataAutoRelease other_settings = obs_encoder_get_settings(other); + kbps = other_settings ? (int)obs_data_get_int(other_settings, "bitrate") : 0; + } if (kbps > lower_kbps && kbps < video_bitrate_kbps) { lower = other; lower_kbps = kbps; @@ -854,8 +861,8 @@ void MoQOutput::VideoInit(obs_encoder_t *encoder) video_tracks[encoder] = std::move(entry); if (lower_bps) - LOG_INFO("Video track initialized (%ux%u, %d kbps, above a %d kbps rendition)", video_width, video_height, - video_bitrate_kbps, lower_kbps); + LOG_INFO("Video track initialized (%ux%u, %d kbps, above a %d kbps rendition)", video_width, + video_height, video_bitrate_kbps, lower_kbps); else LOG_INFO("Video track initialized (%ux%u, %d kbps)", video_width, video_height, video_bitrate_kbps); } diff --git a/cpp/obs/test/moq-output-test.cpp b/cpp/obs/test/moq-output-test.cpp index a0579408c03..cf0910e811e 100644 --- a/cpp/obs/test/moq-output-test.cpp +++ b/cpp/obs/test/moq-output-test.cpp @@ -352,10 +352,11 @@ std::shared_ptr relayBroadcast(TestRelay &relay) } // The broadcast's first catalog `ready` accepts, by default one with video and audio. -std::optional relayCatalog(const std::shared_ptr &broadcast, - std::function ready = [](const moq::Catalog &catalog) { - return !catalog.video.empty() && !catalog.audio.empty(); - }) +std::optional relayCatalog( + const std::shared_ptr &broadcast, + std::function ready = [](const moq::Catalog &catalog) { + return !catalog.video.empty() && !catalog.audio.empty(); + }) { if (!broadcast) return std::nullopt; @@ -713,10 +714,11 @@ int main() o.Data(&resumed); sent += keyframe.size(); CHECK(o.GetTotalBytes() == sent); - CHECK(relayCatalog(broadcast, [&](const moq::Catalog &catalog) { + auto enabled = relayCatalog(broadcast, [&](const moq::Catalog &catalog) { auto it = catalog.video.find(high->first); return it != catalog.video.end() && it->second.enabled; - }).has_value()); + }); + CHECK(enabled.has_value()); } o.Stop(); } diff --git a/dart/moq/lib/src/aliases.dart b/dart/moq/lib/src/aliases.dart index b4b13ca5e50..c0387225748 100644 --- a/dart/moq/lib/src/aliases.dart +++ b/dart/moq/lib/src/aliases.dart @@ -150,6 +150,15 @@ typedef BandwidthConsumer = MoqBandwidthConsumer; /// One track's standing claim on an [Allocator]. typedef Reservation = MoqReservation; +/// Turns a [Reservation]'s grant into a sender's [RateTarget]; call `update` on every frame. +typedef RateControl = MoqRateControl; + +/// How a [RateControl] maps a grant onto a bitrate, and the lower rendition that disables it. +typedef RatePolicy = MoqRatePolicy; + +/// What a sender following a [RateControl] should be doing: an enabled bitrate, or disabled. +typedef RateTarget = MoqRateTarget; + /// A snapshot of transport connection statistics. typedef ConnectionStats = MoqConnectionStats; diff --git a/dart/moq_ffi/lib/src/moq.dart b/dart/moq_ffi/lib/src/moq.dart index 1729c8a4564..afe616db992 100644 --- a/dart/moq_ffi/lib/src/moq.dart +++ b/dart/moq_ffi/lib/src/moq.dart @@ -8,7 +8,9 @@ import "dart:ffi"; import "dart:io" show Platform, File, Directory; import "dart:isolate"; import "dart:typed_data"; + import "package:ffi/ffi.dart"; + import "uniffi_runtime.dart"; export "uniffi_runtime.dart"; @@ -634,6 +636,65 @@ class FfiConverterMoqGrant { } } +class MoqRatePolicy { + final int maxBps; + final int? lowerBps; + MoqRatePolicy({required this.maxBps, this.lowerBps = null}); +} + +class FfiConverterMoqRatePolicy { + static MoqRatePolicy lift(RustBuffer buf) { + return FfiConverterMoqRatePolicy.read(buf.asUint8List()).value; + } + + static LiftRetVal read(Uint8List buf) { + int new_offset = buf.offsetInBytes; + final maxBps_lifted = FfiConverterUInt64.read( + Uint8List.view(buf.buffer, new_offset), + ); + final maxBps = maxBps_lifted.value; + new_offset += maxBps_lifted.bytesRead; + final lowerBps_lifted = FfiConverterOptionalUInt64.read( + Uint8List.view(buf.buffer, new_offset), + ); + final lowerBps = lowerBps_lifted.value; + new_offset += lowerBps_lifted.bytesRead; + return LiftRetVal( + MoqRatePolicy(maxBps: maxBps, lowerBps: lowerBps), + new_offset - buf.offsetInBytes, + ); + } + + static RustBuffer lower(MoqRatePolicy value) { + final total_length = + FfiConverterUInt64.allocationSize(value.maxBps) + + FfiConverterOptionalUInt64.allocationSize(value.lowerBps) + + 0; + final buf = Uint8List(total_length); + write(value, buf); + return toRustBuffer(buf); + } + + static int write(MoqRatePolicy value, Uint8List buf) { + int new_offset = buf.offsetInBytes; + new_offset += FfiConverterUInt64.write( + value.maxBps, + Uint8List.view(buf.buffer, new_offset), + ); + new_offset += FfiConverterOptionalUInt64.write( + value.lowerBps, + Uint8List.view(buf.buffer, new_offset), + ); + return new_offset - buf.offsetInBytes; + } + + static int allocationSize(MoqRatePolicy value) { + return FfiConverterUInt64.allocationSize(value.maxBps) + + FfiConverterOptionalUInt64.allocationSize(value.lowerBps) + + 0; + } +} + class MoqFetchGroupOptions { final int priority; final Duration? maxDelay; @@ -2500,11 +2561,13 @@ class MoqVideoInit { final Uint8List data; final String? label; final MoqVideoHint? hint; + final MoqMediaTrackProducer? lower; MoqVideoInit({ required this.format, required this.data, this.label = null, this.hint = null, + this.lower = null, }); } @@ -2535,8 +2598,19 @@ class FfiConverterMoqVideoInit { ); final hint = hint_lifted.value; new_offset += hint_lifted.bytesRead; + final lower_lifted = FfiConverterOptionalMoqMediaTrackProducer.read( + Uint8List.view(buf.buffer, new_offset), + ); + final lower = lower_lifted.value; + new_offset += lower_lifted.bytesRead; return LiftRetVal( - MoqVideoInit(format: format, data: data, label: label, hint: hint), + MoqVideoInit( + format: format, + data: data, + label: label, + hint: hint, + lower: lower, + ), new_offset - buf.offsetInBytes, ); } @@ -2547,6 +2621,7 @@ class FfiConverterMoqVideoInit { FfiConverterUint8List.allocationSize(value.data) + FfiConverterOptionalString.allocationSize(value.label) + FfiConverterOptionalMoqVideoHint.allocationSize(value.hint) + + FfiConverterOptionalMoqMediaTrackProducer.allocationSize(value.lower) + 0; final buf = Uint8List(total_length); write(value, buf); @@ -2571,6 +2646,10 @@ class FfiConverterMoqVideoInit { value.hint, Uint8List.view(buf.buffer, new_offset), ); + new_offset += FfiConverterOptionalMoqMediaTrackProducer.write( + value.lower, + Uint8List.view(buf.buffer, new_offset), + ); return new_offset - buf.offsetInBytes; } @@ -2579,6 +2658,7 @@ class FfiConverterMoqVideoInit { FfiConverterUint8List.allocationSize(value.data) + FfiConverterOptionalString.allocationSize(value.label) + FfiConverterOptionalMoqVideoHint.allocationSize(value.hint) + + FfiConverterOptionalMoqMediaTrackProducer.allocationSize(value.lower) + 0; } } @@ -5063,6 +5143,120 @@ class FfiConverterMoqAudioSampleFormat { } } +abstract class MoqRateTarget { + RustBuffer lower(); + int allocationSize(); + int write(Uint8List buf); +} + +class FfiConverterMoqRateTarget { + static MoqRateTarget lift(RustBuffer buffer) { + return FfiConverterMoqRateTarget.read(buffer.asUint8List()).value; + } + + static LiftRetVal read(Uint8List buf) { + final index = buf.buffer.asByteData(buf.offsetInBytes).getInt32(0); + final subview = Uint8List.view(buf.buffer, buf.offsetInBytes + 4); + switch (index) { + case 1: + final lifted = EnabledMoqRateTarget.read(subview); + return LiftRetVal( + lifted.value, + lifted.bytesRead - subview.offsetInBytes + 4, + ); + case 2: + final lifted = DisabledMoqRateTarget.read(subview); + return LiftRetVal( + lifted.value, + lifted.bytesRead - subview.offsetInBytes + 4, + ); + default: + throw UniffiInternalError( + UniffiInternalError.unexpectedEnumCase, + "Unable to determine enum variant", + ); + } + } + + static RustBuffer lower(MoqRateTarget value) { + return value.lower(); + } + + static int allocationSize(MoqRateTarget value) { + return value.allocationSize(); + } + + static int write(MoqRateTarget value, Uint8List buf) { + return value.write(buf) - buf.offsetInBytes; + } +} + +class EnabledMoqRateTarget extends MoqRateTarget { + final int bitrateBps; + EnabledMoqRateTarget(int this.bitrateBps); + EnabledMoqRateTarget._(int this.bitrateBps); + static LiftRetVal read(Uint8List buf) { + int new_offset = buf.offsetInBytes; + final bitrateBps_lifted = FfiConverterUInt64.read( + Uint8List.view(buf.buffer, new_offset), + ); + final bitrateBps = bitrateBps_lifted.value; + new_offset += bitrateBps_lifted.bytesRead; + return LiftRetVal(EnabledMoqRateTarget._(bitrateBps), new_offset); + } + + @override + RustBuffer lower() { + final buf = Uint8List(allocationSize()); + write(buf); + return toRustBuffer(buf); + } + + @override + int allocationSize() { + return FfiConverterUInt64.allocationSize(bitrateBps) + 4; + } + + @override + int write(Uint8List buf) { + buf.buffer.asByteData(buf.offsetInBytes).setInt32(0, 1); + int new_offset = buf.offsetInBytes + 4; + new_offset += FfiConverterUInt64.write( + bitrateBps, + Uint8List.view(buf.buffer, new_offset), + ); + return new_offset; + } +} + +class DisabledMoqRateTarget extends MoqRateTarget { + DisabledMoqRateTarget(); + DisabledMoqRateTarget._(); + static LiftRetVal read(Uint8List buf) { + int new_offset = buf.offsetInBytes; + return LiftRetVal(DisabledMoqRateTarget._(), new_offset); + } + + @override + RustBuffer lower() { + final buf = Uint8List(allocationSize()); + write(buf); + return toRustBuffer(buf); + } + + @override + int allocationSize() { + return 4; + } + + @override + int write(Uint8List buf) { + buf.buffer.asByteData(buf.offsetInBytes).setInt32(0, 2); + int new_offset = buf.offsetInBytes + 4; + return new_offset; + } +} + abstract class MoqAnchor { RustBuffer lower(); int allocationSize(); @@ -9410,6 +9604,102 @@ class FfiConverterMoqBandwidthConsumer { } } +abstract class MoqRateControlInterface { + MoqRateTarget target(); + MoqRateTarget? update({required MoqReservation reservation}); +} + +final _MoqRateControlFinalizer = Finalizer>((ptr) { + rustCall((status) => uniffi_moq_ffi_fn_free_moqratecontrol(ptr, status)); +}); + +class MoqRateControl implements MoqRateControlInterface { + late final Pointer _ptr; + bool _isDisposed = false; + MoqRateControl._(this._ptr) { + _MoqRateControlFinalizer.attach(this, _ptr, detach: this); + } + MoqRateControl({required MoqRatePolicy policy}) + : _ptr = rustCall( + (status) => uniffi_moq_ffi_fn_constructor_moqratecontrol_new( + FfiConverterMoqRatePolicy.lower(policy), + status, + ), + null, + ) { + _MoqRateControlFinalizer.attach(this, _ptr, detach: this); + } + factory MoqRateControl.lift(Pointer ptr) { + return MoqRateControl._(ptr); + } + Pointer uniffiClonePointer() { + if (_isDisposed) { + throw StateError("Object has already been disposed"); + } + return rustCall( + (status) => uniffi_moq_ffi_fn_clone_moqratecontrol(_ptr, status), + ); + } + + void dispose() { + if (_isDisposed) { + return; + } + _isDisposed = true; + _MoqRateControlFinalizer.detach(this); + rustCall((status) => uniffi_moq_ffi_fn_free_moqratecontrol(_ptr, status)); + } + + MoqRateTarget target() { + return rustCallWithLifter( + (status) => uniffi_moq_ffi_fn_method_moqratecontrol_target( + uniffiClonePointer(), + status, + ), + FfiConverterMoqRateTarget.lift, + null, + ); + } + + MoqRateTarget? update({required MoqReservation reservation}) { + return rustCallWithLifter( + (status) => uniffi_moq_ffi_fn_method_moqratecontrol_update( + uniffiClonePointer(), + FfiConverterMoqReservation.lower(reservation), + status, + ), + FfiConverterOptionalMoqRateTarget.lift, + null, + ); + } +} + +class FfiConverterMoqRateControl { + static MoqRateControl lift(Pointer ptr) { + return MoqRateControl.lift(ptr); + } + + static Pointer lower(MoqRateControl value) { + return value.uniffiClonePointer(); + } + + static int allocationSize(MoqRateControl value) { + return 8; + } + + static LiftRetVal read(Uint8List buf) { + final handle = buf.buffer.asByteData(buf.offsetInBytes).getInt64(0); + final pointer = Pointer.fromAddress(handle); + return LiftRetVal(MoqRateControl.lift(pointer), 8); + } + + static int write(MoqRateControl value, Uint8List buf) { + final handle = lower(value); + buf.buffer.asByteData(buf.offsetInBytes).setInt64(0, handle.address); + return 8; + } +} + abstract class MoqReservationInterface { int? rate(); void update({required int maxBps}); @@ -11554,6 +11844,7 @@ abstract class MoqMediaTrackProducerInterface { void discontinuity(); void finish(); void seek({required int sequence}); + void setEnabled({required bool enabled}); void writeFrame({required MoqFrame frame}); } @@ -11670,6 +11961,16 @@ class MoqMediaTrackProducer implements MoqMediaTrackProducerInterface { }, moqExceptionErrorHandler); } + void setEnabled({required bool enabled}) { + return rustCall((status) { + uniffi_moq_ffi_fn_method_moqmediatrackproducer_set_enabled( + uniffiClonePointer(), + FfiConverterBool.lower(enabled), + status, + ); + }, moqExceptionErrorHandler); + } + void writeFrame({required MoqFrame frame}) { return rustCall((status) { uniffi_moq_ffi_fn_method_moqmediatrackproducer_write_frame( @@ -15975,6 +16276,58 @@ class FfiConverterOptionalMoqMediaTimebase { } } +class FfiConverterOptionalMoqMediaTrackProducer { + static MoqMediaTrackProducer? lift(RustBuffer buf) { + return FfiConverterOptionalMoqMediaTrackProducer.read( + buf.asUint8List(), + ).value; + } + + static LiftRetVal read(Uint8List buf) { + if (ByteData.view(buf.buffer, buf.offsetInBytes).getInt8(0) == 0) { + return LiftRetVal(null, 1); + } + final result = FfiConverterMoqMediaTrackProducer.read( + Uint8List.view(buf.buffer, buf.offsetInBytes + 1), + ); + return LiftRetVal( + result.value, + result.bytesRead + 1, + ); + } + + static int allocationSize([MoqMediaTrackProducer? value]) { + if (value == null) { + return 1; + } + return FfiConverterMoqMediaTrackProducer.allocationSize(value) + 1; + } + + static RustBuffer lower(MoqMediaTrackProducer? value) { + if (value == null) { + return toRustBuffer(Uint8List.fromList([0])); + } + final buf = Uint8List( + FfiConverterOptionalMoqMediaTrackProducer.allocationSize(value), + ); + FfiConverterOptionalMoqMediaTrackProducer.write(value, buf); + return toRustBuffer(buf); + } + + static int write(MoqMediaTrackProducer? value, Uint8List buf) { + if (value == null) { + buf[0] = 0; + return 1; + } + buf[0] = 1; + return FfiConverterMoqMediaTrackProducer.write( + value, + Uint8List.view(buf.buffer, buf.offsetInBytes + 1), + ) + + 1; + } +} + class FfiConverterOptionalMoqOrder { static MoqOrder? lift(RustBuffer buf) { return FfiConverterOptionalMoqOrder.read(buf.asUint8List()).value; @@ -16067,6 +16420,53 @@ class FfiConverterOptionalMoqOriginProducer { } } +class FfiConverterOptionalMoqRateTarget { + static MoqRateTarget? lift(RustBuffer buf) { + return FfiConverterOptionalMoqRateTarget.read(buf.asUint8List()).value; + } + + static LiftRetVal read(Uint8List buf) { + if (ByteData.view(buf.buffer, buf.offsetInBytes).getInt8(0) == 0) { + return LiftRetVal(null, 1); + } + final result = FfiConverterMoqRateTarget.read( + Uint8List.view(buf.buffer, buf.offsetInBytes + 1), + ); + return LiftRetVal(result.value, result.bytesRead + 1); + } + + static int allocationSize([MoqRateTarget? value]) { + if (value == null) { + return 1; + } + return FfiConverterMoqRateTarget.allocationSize(value) + 1; + } + + static RustBuffer lower(MoqRateTarget? value) { + if (value == null) { + return toRustBuffer(Uint8List.fromList([0])); + } + final buf = Uint8List( + FfiConverterOptionalMoqRateTarget.allocationSize(value), + ); + FfiConverterOptionalMoqRateTarget.write(value, buf); + return toRustBuffer(buf); + } + + static int write(MoqRateTarget? value, Uint8List buf) { + if (value == null) { + buf[0] = 0; + return 1; + } + buf[0] = 1; + return FfiConverterMoqRateTarget.write( + value, + Uint8List.view(buf.buffer, buf.offsetInBytes + 1), + ) + + 1; + } +} + class FfiConverterOptionalMoqRequest { static MoqRequest? lift(RustBuffer buf) { return FfiConverterOptionalMoqRequest.read(buf.asUint8List()).value; @@ -17525,6 +17925,47 @@ external RustBuffer uniffi_moq_ffi_fn_method_moqbandwidthconsumer_rate( external Pointer uniffi_moq_ffi_fn_method_moqbandwidthconsumer_rate_changed(Pointer ptr); +@Native Function(Pointer, Pointer)>( + assetId: _uniffiAssetId, +) +external Pointer uniffi_moq_ffi_fn_clone_moqratecontrol( + Pointer handle, + Pointer uniffiStatus, +); + +@Native, Pointer)>( + assetId: _uniffiAssetId, +) +external void uniffi_moq_ffi_fn_free_moqratecontrol( + Pointer handle, + Pointer uniffiStatus, +); + +@Native Function(RustBuffer, Pointer)>( + assetId: _uniffiAssetId, +) +external Pointer uniffi_moq_ffi_fn_constructor_moqratecontrol_new( + RustBuffer policy, + Pointer uniffiStatus, +); + +@Native, Pointer)>( + assetId: _uniffiAssetId, +) +external RustBuffer uniffi_moq_ffi_fn_method_moqratecontrol_target( + Pointer ptr, + Pointer uniffiStatus, +); + +@Native< + RustBuffer Function(Pointer, Pointer, Pointer) +>(assetId: _uniffiAssetId) +external RustBuffer uniffi_moq_ffi_fn_method_moqratecontrol_update( + Pointer ptr, + Pointer reservation, + Pointer uniffiStatus, +); + @Native Function(Pointer, Pointer)>( assetId: _uniffiAssetId, ) @@ -18512,6 +18953,15 @@ external void uniffi_moq_ffi_fn_method_moqmediatrackproducer_seek( Pointer uniffiStatus, ); +@Native, Int8, Pointer)>( + assetId: _uniffiAssetId, +) +external void uniffi_moq_ffi_fn_method_moqmediatrackproducer_set_enabled( + Pointer ptr, + int enabled, + Pointer uniffiStatus, +); + @Native, RustBuffer, Pointer)>( assetId: _uniffiAssetId, ) @@ -20292,6 +20742,12 @@ external int uniffi_moq_ffi_checksum_method_moqbandwidthconsumer_rate(); @Native(assetId: _uniffiAssetId) external int uniffi_moq_ffi_checksum_method_moqbandwidthconsumer_rate_changed(); +@Native(assetId: _uniffiAssetId) +external int uniffi_moq_ffi_checksum_method_moqratecontrol_target(); + +@Native(assetId: _uniffiAssetId) +external int uniffi_moq_ffi_checksum_method_moqratecontrol_update(); + @Native(assetId: _uniffiAssetId) external int uniffi_moq_ffi_checksum_method_moqreservation_rate(); @@ -20480,6 +20936,9 @@ external int uniffi_moq_ffi_checksum_method_moqmediatrackproducer_finish(); @Native(assetId: _uniffiAssetId) external int uniffi_moq_ffi_checksum_method_moqmediatrackproducer_seek(); +@Native(assetId: _uniffiAssetId) +external int uniffi_moq_ffi_checksum_method_moqmediatrackproducer_set_enabled(); + @Native(assetId: _uniffiAssetId) external int uniffi_moq_ffi_checksum_method_moqmediatrackproducer_write_frame(); @@ -20833,6 +21292,9 @@ external int uniffi_moq_ffi_checksum_constructor_moqaudioencoder_new(); @Native(assetId: _uniffiAssetId) external int uniffi_moq_ffi_checksum_constructor_moqaudioproducer_new(); +@Native(assetId: _uniffiAssetId) +external int uniffi_moq_ffi_checksum_constructor_moqratecontrol_new(); + @Native(assetId: _uniffiAssetId) external int uniffi_moq_ffi_checksum_constructor_moqflatesnapshotproducer_new(); @@ -21002,6 +21464,12 @@ void _checkApiChecksums() { 38021) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } + if (uniffi_moq_ffi_checksum_method_moqratecontrol_target() != 6016) { + throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); + } + if (uniffi_moq_ffi_checksum_method_moqratecontrol_update() != 28669) { + throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); + } if (uniffi_moq_ffi_checksum_method_moqreservation_rate() != 65114) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } @@ -21204,6 +21672,10 @@ void _checkApiChecksums() { if (uniffi_moq_ffi_checksum_method_moqmediatrackproducer_seek() != 37244) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } + if (uniffi_moq_ffi_checksum_method_moqmediatrackproducer_set_enabled() != + 1046) { + throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); + } if (uniffi_moq_ffi_checksum_method_moqmediatrackproducer_write_frame() != 7616) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); @@ -21567,6 +22039,9 @@ void _checkApiChecksums() { if (uniffi_moq_ffi_checksum_constructor_moqaudioproducer_new() != 51486) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } + if (uniffi_moq_ffi_checksum_constructor_moqratecontrol_new() != 63053) { + throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); + } if (uniffi_moq_ffi_checksum_constructor_moqflatesnapshotproducer_new() != 40377) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); diff --git a/doc/bin/obs.md b/doc/bin/obs.md index f2de02b73b3..d12d7a0e7e7 100644 --- a/doc/bin/obs.md +++ b/doc/bin/obs.md @@ -44,6 +44,13 @@ shows the current video target beside the configured rate. An encoder shared with a recording follows too, so record with a separate encoder to keep the recording at the full rate. +With several video encoders on the output, each one is a rendition, and one +above a lower-bitrate encoder stops sending while its share of the estimate +cannot fund it, advertised as disabled so viewers switch down. It keeps +encoding, since OBS cannot pause one encoder of a live output, and resumes at +that encoder's next keyframe once the estimate recovers. This needs the +session to probe for bandwidth, which takes moq-quic with BBR3. + ## Source quality and moq-transcode OBS publishes **one** rendition and does not encode a viewer ladder. When a diff --git a/doc/lib/cpp/index.md b/doc/lib/cpp/index.md index 4728f2ba9d1..f1a1275536c 100644 --- a/doc/lib/cpp/index.md +++ b/doc/lib/cpp/index.md @@ -138,7 +138,7 @@ ok(broadcast->announce(route)); ok(broadcast->close()); // keep the producer alive while publishing, then close explicitly ``` -Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Producer options hold input, codec settings, track naming, and optional session bandwidth ownership; standalone encoders take only codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. +Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Producer options hold input, codec settings, track naming, and optional session bandwidth ownership; standalone encoders take only codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded video track takes `lower` in its `VideoInit` too, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `set_enabled(false)` after a `discontinuity()`, and `set_enabled(true)` on the keyframe that resumes it. ## Futures diff --git a/doc/lib/dart/index.md b/doc/lib/dart/index.md index 0889e49c7cf..d8012377a15 100644 --- a/doc/lib/dart/index.md +++ b/doc/lib/dart/index.md @@ -132,7 +132,7 @@ mine.close(); await moq.close(); ``` -Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Producer options hold input, codec settings, track naming, and optional session bandwidth ownership; standalone encoders take only codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. +Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Producer options hold input, codec settings, track naming, and optional session bandwidth ownership; standalone encoders take only codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded video track takes `lower` in its `VideoInit` too, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `setEnabled(false)` after a `discontinuity()`, and `setEnabled(true)` on the keyframe that resumes it. ## Things to know diff --git a/doc/lib/go/index.md b/doc/lib/go/index.md index 38ab0f5f69b..4f8f520ea5d 100644 --- a/doc/lib/go/index.md +++ b/doc/lib/go/index.md @@ -117,7 +117,7 @@ _ = video.Finish() broadcast.Close() // keep the producer reachable while publishing, then close explicitly ``` -Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Producer options hold input, codec settings, track naming, and optional session bandwidth ownership; standalone encoders take only codec settings. For simulcast, give each rendition the next one down as `Lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. +Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Producer options hold input, codec settings, track naming, and optional session bandwidth ownership; standalone encoders take only codec settings. For simulcast, give each rendition the next one down as `Lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded video track takes `Lower` in its `media.VideoInit` too, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `SetEnabled(false)` after a `Discontinuity()`, and `SetEnabled(true)` on the keyframe that resumes it. ## Things to know diff --git a/doc/lib/kt/index.md b/doc/lib/kt/index.md index 8b399b92fe1..d609db706ec 100644 --- a/doc/lib/kt/index.md +++ b/doc/lib/kt/index.md @@ -89,7 +89,7 @@ Moq.connect("https://relay.example.com").use { moq -> } ``` -Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Producer options hold input, codec settings, track naming, and optional session bandwidth ownership; standalone encoders take only codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. +Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Producer options hold input, codec settings, track naming, and optional session bandwidth ownership; standalone encoders take only codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded video track takes `lower` in its `VideoInit` too, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `setEnabled(false)` after a `discontinuity()`, and `setEnabled(true)` on the keyframe that resumes it. ## Things to know diff --git a/doc/lib/py/index.md b/doc/lib/py/index.md index 5cf642d5fe2..d8e3e0cecf0 100644 --- a/doc/lib/py/index.md +++ b/doc/lib/py/index.md @@ -106,7 +106,7 @@ async def main(): asyncio.run(main()) ``` -Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Pass the publication track name to the producer, alongside its input and codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. +Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Pass the publication track name to the producer, alongside its input and codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded `media.TrackProducer.video` takes `lower` too, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `set_enabled(False)` after a `discontinuity()`, and `set_enabled(True)` on the keyframe that resumes it. ## Things to know diff --git a/doc/lib/swift/index.md b/doc/lib/swift/index.md index 6743997dbba..3d6d08549a0 100644 --- a/doc/lib/swift/index.md +++ b/doc/lib/swift/index.md @@ -90,7 +90,7 @@ session.shutdown() try await session.closed() ``` -Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Pass the publication track name to the producer, alongside its input and codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. +Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Pass the publication track name to the producer, alongside its input and codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded `Media.TrackProducer.video` takes `lower` too, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `setEnabled(false)` after a `discontinuity()`, and `setEnabled(true)` on the keyframe that resumes it. ## Things to know diff --git a/go/wrapper/MIGRATE.md b/go/wrapper/MIGRATE.md index 32cc37e56f7..e0e01e6ad64 100644 --- a/go/wrapper/MIGRATE.md +++ b/go/wrapper/MIGRATE.md @@ -4,6 +4,10 @@ How to update your code for each breaking change, newest first. ## Unreleased +### `media.VideoInit` is a struct + +`media.VideoInit` is its own struct rather than an alias of the generated record, so its new `Lower` field takes a `*media.TrackProducer`. A composite literal with `Format`, `Data`, `Label`, and `Hint` compiles unchanged; build one that way where you converted a generated `MoqVideoInit`. + ### Paths with an empty segment are refused ([#6648](https://github.com/moq-dev/moq/pull/6648)) Publishing, serving, or requesting a broadcast at a path with a leading, trailing, or doubled `/` returns an error instead of collapsing it. Pass the canonical path, such as `room/cam`. diff --git a/go/wrapper/media/media.go b/go/wrapper/media/media.go index a88ea9bb01a..d44b1c1452f 100644 --- a/go/wrapper/media/media.go +++ b/go/wrapper/media/media.go @@ -55,10 +55,31 @@ type ( // AudioInit describes an audio codec and its initialization bytes. AudioInit = ffi.MoqAudioInit - // VideoInit describes a video codec, initialization bytes, and hints. - VideoInit = ffi.MoqVideoInit ) +// VideoInit describes a video codec, initialization bytes, and hints. +type VideoInit struct { + // Format selects the video codec. + Format VideoFormat + // Data supplies codec init bytes (an avcC, an hvcC, ...); may be empty for a format that resolves in band. + Data []byte + // Label is a human-readable rendition name for a track picker. + Label *string + // Hint carries catalog fields the stream cannot reveal itself. + Hint *VideoHint + // Lower is the next lower rendition of the same source: this track then sits one + // priority below it, so a bandwidth allocator funds that one first. + Lower *TrackProducer +} + +func (init VideoInit) ffi() ffi.MoqVideoInit { + var lower **ffi.MoqMediaTrackProducer + if init.Lower != nil { + lower = &init.Lower.inner + } + return ffi.MoqVideoInit{Format: init.Format, Data: init.Data, Label: init.Label, Hint: init.Hint, Lower: lower} +} + // Timebase shares a source's timestamp offset without keeping its broadcast open. type Timebase struct{ inner *ffi.MoqMediaTimebase } @@ -185,7 +206,7 @@ func NewVideoTrackProducer(broadcast *moq.BroadcastProducer, target Target, init if err != nil { return nil, err } - inner, err := ffi.MoqMediaTrackProducerVideo(bridge.BroadcastProducer(broadcast), resolved, init) + inner, err := ffi.MoqMediaTrackProducerVideo(bridge.BroadcastProducer(broadcast), resolved, init.ffi()) if err != nil { return nil, err } @@ -198,7 +219,7 @@ func NewVideoTrackStreamProducer(broadcast *moq.BroadcastProducer, target Target if err != nil { return nil, err } - inner, err := ffi.MoqMediaTrackStreamProducerVideo(bridge.BroadcastProducer(broadcast), resolved, init) + inner, err := ffi.MoqMediaTrackStreamProducerVideo(bridge.BroadcastProducer(broadcast), resolved, init.ffi()) if err != nil { return nil, err } @@ -386,6 +407,14 @@ func (m *TrackProducer) Discontinuity() error { return m.inner.Discontinuity() } +// SetEnabled advertises whether this video rendition is being produced, as its catalog +// enabled flag. Disable it when its sender stops for lack of bandwidth and mark a +// Discontinuity; enable it again just before the keyframe that resumes it. An audio +// track refuses. +func (m *TrackProducer) SetEnabled(enabled bool) error { + return m.inner.SetEnabled(enabled) +} + // Cut draws a group boundary here. // // Audio has no boundary of its own (every packet is independently decodable), so this is diff --git a/go/wrapper/session.go b/go/wrapper/session.go index 119ab1317f3..73df812ac0d 100644 --- a/go/wrapper/session.go +++ b/go/wrapper/session.go @@ -176,6 +176,44 @@ func (r *Reservation) Update(maxBps uint64) { r.inner.Update(maxBps) } +// RatePolicy maps a reservation's grant onto a sender's bitrate: MaxBps caps it, +// and LowerBps, the next lower rendition's MaxBps, disables this rendition while +// the grant is below (MaxBps + 2*LowerBps) / 3 and a session probes for it. +type RatePolicy = ffi.MoqRatePolicy + +// RateTarget is what a sender following a RateControl should be doing: a +// RateTargetEnabled bitrate, or RateTargetDisabled. +type RateTarget = ffi.MoqRateTarget + +// RateTargetEnabled produces at BitrateBps and advertises the rendition as enabled. +type RateTargetEnabled = ffi.MoqRateTargetEnabled + +// RateTargetDisabled stops sending and advertises the rendition as disabled. +type RateTargetDisabled = ffi.MoqRateTargetDisabled + +// RateControl turns a Reservation's grant into a sender's RateTarget, with the +// policy moq's own encoders follow: drops apply at once, raises climb at most 25% +// per second, and moves under 5% are ignored. For a sender that encodes on its +// own, call Update on every frame and apply what it returns. +type RateControl struct { + inner *ffi.MoqRateControl +} + +// NewRateControl starts at policy.MaxBps until a grant says otherwise. +func NewRateControl(policy RatePolicy) *RateControl { + return &RateControl{inner: ffi.NewMoqRateControl(policy)} +} + +// Target returns what the sender should currently be doing. +func (c *RateControl) Target() RateTarget { + return c.inner.Target() +} + +// Update feeds reservation's grant, returning a new target worth applying, or nil. +func (c *RateControl) Update(reservation *Reservation) *RateTarget { + return c.inner.Update(reservation.inner) +} + // Publish returns the origin used to advertise local broadcasts to the remote. func (s *Session) Publish() *OriginProducer { return &OriginProducer{inner: s.inner.Publish()} diff --git a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Aliases.kt b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Aliases.kt index 19d384bf4c8..25c17018577 100644 --- a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Aliases.kt +++ b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Aliases.kt @@ -155,6 +155,12 @@ typealias Allocator = uniffi.moq.MoqAllocator typealias BandwidthConsumer = uniffi.moq.MoqBandwidthConsumer /** One track's standing claim on an [Allocator]. */ typealias Reservation = uniffi.moq.MoqReservation +/** Turns a [Reservation]'s grant into a sender's [RateTarget]; call `update` on every frame. */ +typealias RateControl = uniffi.moq.MoqRateControl +/** How a [RateControl] maps a grant onto a bitrate, and the lower rendition that disables it. */ +typealias RatePolicy = uniffi.moq.MoqRatePolicy +/** What a sender following a [RateControl] should be doing: an enabled bitrate, or disabled. */ +typealias RateTarget = uniffi.moq.MoqRateTarget /** A snapshot of transport connection statistics. */ typealias ConnectionStats = uniffi.moq.MoqConnectionStats /** Follows the grant a session's peer gave it; cancel it to release a pending wait. */ diff --git a/py/moq-rs/moq/__init__.py b/py/moq-rs/moq/__init__.py index 169fa743a5d..bbe8d0d2147 100644 --- a/py/moq-rs/moq/__init__.py +++ b/py/moq-rs/moq/__init__.py @@ -41,7 +41,16 @@ TrackRequests, ) from .server import Request, Server, Transport -from .session import Allocator, BandwidthConsumer, GrantConsumer, Reservation, Session, Tokens +from .session import ( + Allocator, + BandwidthConsumer, + GrantConsumer, + RateControl, + RateTarget, + Reservation, + Session, + Tokens, +) from .subscribe import ( BroadcastConsumer, GroupConsumer, @@ -126,6 +135,8 @@ "OriginProducer", "Order", "Range", + "RateControl", + "RateTarget", "Request", "Reservation", "Route", diff --git a/py/moq-rs/moq/media.py b/py/moq-rs/moq/media.py index 0ac4923cffb..781534d6ec3 100644 --- a/py/moq-rs/moq/media.py +++ b/py/moq-rs/moq/media.py @@ -2,6 +2,7 @@ from __future__ import annotations +import copy import json from dataclasses import dataclass, field from datetime import timedelta @@ -166,9 +167,20 @@ def audio(cls, broadcast: BroadcastProducer, init: AudioInit, *, target: Target return cls(MoqMediaTrackProducer.audio(broadcast._inner, target._ffi(), init)) @classmethod - def video(cls, broadcast: BroadcastProducer, init: VideoInit, *, target: Target = Named()) -> TrackProducer: - """Import complete encoded video frames on a named or requested track.""" - return cls(MoqMediaTrackProducer.video(broadcast._inner, target._ffi(), init)) + def video( + cls, + broadcast: BroadcastProducer, + init: VideoInit, + *, + target: Target = Named(), + lower: TrackProducer | None = None, + ) -> TrackProducer: + """Import complete encoded video frames on a named or requested track. + + Pass ``lower``, the next lower rendition of the same source, to publish this + track one priority below it, so a bandwidth allocator funds that one first. + """ + return cls(MoqMediaTrackProducer.video(broadcast._inner, target._ffi(), _lower(init, lower))) def __init__(self, inner: MoqMediaTrackProducer) -> None: self._inner = inner @@ -185,6 +197,15 @@ def discontinuity(self) -> None: """Mark a timeline break, preserving the advertised jitter and delay.""" self._inner.discontinuity() + def set_enabled(self, enabled: bool) -> None: + """Advertise whether this video rendition is being produced, as its catalog ``enabled``. + + Disable it when its sender stops for lack of bandwidth and mark a + :meth:`discontinuity`; enable it again just before the keyframe that + resumes it. An audio track refuses. + """ + self._inner.set_enabled(enabled) + def cut(self) -> None: """Draw a group boundary here. @@ -252,9 +273,19 @@ class TrackStreamProducer: """ @classmethod - def video(cls, broadcast: BroadcastProducer, init: VideoInit, *, target: Target = Named()) -> TrackStreamProducer: - """Import a video byte stream, inferring frame boundaries.""" - return cls(MoqMediaTrackStreamProducer.video(broadcast._inner, target._ffi(), init)) + def video( + cls, + broadcast: BroadcastProducer, + init: VideoInit, + *, + target: Target = Named(), + lower: TrackProducer | None = None, + ) -> TrackStreamProducer: + """Import a video byte stream, inferring frame boundaries. + + ``lower`` is as for :meth:`TrackProducer.video`. + """ + return cls(MoqMediaTrackStreamProducer.video(broadcast._inner, target._ffi(), _lower(init, lower))) def __init__(self, inner: MoqMediaTrackStreamProducer) -> None: self._inner = inner @@ -272,6 +303,14 @@ def finish(self) -> None: self._inner.finish() +def _lower(init: VideoInit, lower: TrackProducer | None) -> VideoInit: + if lower is None: + return init + init = copy.copy(init) + init.lower = lower._inner + return init + + def _rendition(id: str, rendition: Video | Audio) -> MoqRendition: if isinstance(rendition, Video): return cast(MoqRendition, MoqRendition.VIDEO(id=id, config=rendition)) diff --git a/py/moq-rs/moq/session.py b/py/moq-rs/moq/session.py index 33619e8ff04..e066dbf06de 100644 --- a/py/moq-rs/moq/session.py +++ b/py/moq-rs/moq/session.py @@ -8,10 +8,13 @@ MoqAllocator, MoqBandwidthConsumer, MoqGrantConsumer, + MoqRateControl, + MoqRatePolicy, MoqReservation, MoqSession, MoqTokens, ) +from moq_ffi import MoqRateTarget as RateTarget from ._records import _strs from .errors import is_shutdown @@ -269,3 +272,29 @@ def rate(self) -> int | None: def update(self, max_bps: int) -> None: """Change the ceiling, keeping the same claim.""" self._inner.update(max_bps) + + +class RateControl: + """Turns a :class:`Reservation`'s grant into a sender's :data:`RateTarget`. + + The same policy moq's own encoders follow: drops apply at once, raises climb + at most 25% per second, and moves under 5% are ignored. For a sender that + encodes on its own, such as a pre-encoded track: call :meth:`update` on every + frame and apply what it returns. + + Pass ``lower_bps``, the next lower rendition's ``max_bps``, for a rendition + encoded beside lower ones of the same source: a grant below + ``(max_bps + 2 * lower_bps) / 3`` then disables it while a session probes for + the reservation. + """ + + def __init__(self, max_bps: int, *, lower_bps: int | None = None) -> None: + self._inner = MoqRateControl(MoqRatePolicy(max_bps=max_bps, lower_bps=lower_bps)) + + def target(self) -> RateTarget: + """What the sender should currently be doing.""" + return self._inner.target() + + def update(self, reservation: Reservation) -> RateTarget | None: + """Feed ``reservation``'s grant, returning a new target worth applying, or ``None``.""" + return self._inner.update(reservation._inner) diff --git a/quest/m1/README.md b/quest/m1/README.md index 8990c91aa21..8854cb5725b 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -142,7 +142,7 @@ surface and zero-copy texture import, land before the cut. - [Rust identity enforcement](/quest/m1/catalog-identity-rs.md) - `moq-mux` refuses a redefined rendition, and every Rust publisher mints a new rendition name for a decoder identity change and updates everything else in place - [IETF takeover](/quest/m1/ietf-takeover.md) - carry the shared takeover contract through the negotiated IETF epoch extension - [Identity and recovery guide](/quest/m1/identity-guide.md) - explain epochs, pinned catalogs, lazy jobs, restart and version compatibility as they exist; takeover adds its own section -- [Native rung disable](/quest/m1/simulcast-rung-disable.md) - OBS stops sending the top renditions its grant cannot fund, as the Rust/FFI publishers do, and a shaped uplink through moq-ffi proves the resume +- [Native rung disable proof](/quest/m1/simulcast-rung-disable.md) - a shaped uplink through moq-ffi proves the top rendition stops and resumes over real BBR3 probing - [Audio jitter target](/quest/m1/audio-jitter-target/README.md) - the playout target is default in both languages; a manual browser proof and a native trace replay remain - [A/V clock](/quest/m1/av-clock.md) - the audio playhead drives Sync.reference while audio plays, through per-track sync handles - [Container reads drop backfill](/quest/m1/read-backfill.md) - the JS container consumer drops groups below its read cursor like Rust diff --git a/quest/m1/simulcast-rung-disable.md b/quest/m1/simulcast-rung-disable.md index a2e156595b2..afc47c9da69 100644 --- a/quest/m1/simulcast-rung-disable.md +++ b/quest/m1/simulcast-rung-disable.md @@ -1,49 +1,41 @@ -# [M] OBS disables simulcast rungs the grant cannot fund, proven on a shaped uplink +# [M] A shaped uplink proves native rung disable and resume ## Goal -The OBS plugin stops sending its top video renditions when their bandwidth -grant cannot fund them, advertising each as `enabled: false`, and re-enables -them with hysteresis once the grant recovers, as the Rust and FFI publishers -already do. One shaped-uplink run through moq-ffi proves the rule end to end -over real BBR3 probing, including resuming a rung after the grant fell below -the lowest rung. The JS publisher is +One shaped-uplink run through moq-ffi proves the rung rule end to end over +real BBR3 probing: as the bottleneck narrows below a rung's band, the top +rendition stops and is advertised `enabled: false`, and once the bottleneck +opens back up it resumes with hysteresis, including after the grant fell below +the lowest rung. It runs nightly. The JS publisher is [JS rung disable](/quest/m2/probe/rung-disable-js.md). ## Plan -Landed in [moq-dev/moq#6781](https://github.com/moq-dev/moq/pull/6781): the -rule lives in `moq_mux::rate` (`Policy::lower` sets the band boundary -`(max + 2 * lower) / 3`, `Follower` yields `rate::Target::Disabled`), only -once a session probes for the reservation (`bandwidth::Consumer::probed`). -An idle reservation reads the share it would get, so recovery does not depend -on demand. The FFI video producer takes `lower` (the next rendition down), -sits one track priority below it, drains its encoder while disabled, and -reopens on a keyframe; `moq_encode_video` takes the same `lower` handle in -its `moq_video_encode_options`. - -Depends on BBR3 aging its estimate: the grant this rule reads can stay high -for 20 s or more after the path drops, since `update_max_bw` in -`rs/moq-congestion/src/bbr3/mod.rs` never ages an app-limited `max_bw` -(measured in [moq-dev/moq#6635](https://github.com/moq-dev/moq/pull/6635)), -so a rung keeps sending into a narrowed path until it does. A separate quest -fixes that; the shaped-uplink proof needs it to disable on time. - -- OBS: each OBS video encoder is its own rendition (`video_tracks` in - `cpp/obs/src/moq-output.cpp`) publishing pre-encoded packets through a - media track, and `MoQRate` (`cpp/obs/src/moq-rate.h`) mirrors - `rate::Control` without the rung rule. It needs, through the generated C++ - package: a way to flip a media track's catalog `enabled`, the reservation's - `probed` state (dropped from `MoqReservation` in #6781 for lack of a - consumer), and the lower encoder's ceiling and priority. OBS cannot stop an - encoder or force a keyframe per rung mid-stream, so decide between dropping - packets while disabled and resuming at the next natural keyframe, or - restarting the encoder. -- Shaped uplink: `moq-shaper` never steps its rate limit (queued datagrams - would reorder), so a run whose bottleneck opens back up needs either a - rate step the shaper can apply without reordering, or a second shaper the - first retargets to. Run it nightly with an `#[ignore]` test and a recipe in - `rs/justfile`, beside `quic-sim`. +The rule and every native publisher landed: `moq_mux::rate` in +[moq-dev/moq#6781](https://github.com/moq-dev/moq/pull/6781) (`Policy::lower` +sets the band boundary `(max + 2 * lower) / 3`; a rung disables only while a +session probes for its reservation), the FFI and C video producers there too, +and OBS through moq-ffi's `MoqRateControl`, `MoqVideoInit.lower`, and +`MoqMediaTrackProducer::set_enabled`, which a pre-encoded publisher follows +the same way. + +- Blocked on [BBR3 aging](/quest/m1/bbr-app-limited.md): the grant can stay + high for 20 s or more after the path drops, since `update_max_bw` in + `rs/moq-congestion/src/bbr3/mod.rs` never ages an app-limited `max_bw` + (measured in [moq-dev/moq#6635](https://github.com/moq-dev/moq/pull/6635)), + so a rung keeps sending into a narrowed path until that lands. +- `moq-shaper` never steps its rate limit (queued datagrams would reorder), + so a run whose bottleneck opens back up needs either a rate step the shaper + can apply without reordering, or a second shaper the first retargets to. +- Shape it as bufferbloat: a rate limit below the top rung behind a deep but + finite queue whose overflow drops packets. Assert the disable, the catalog + `enabled` flips, and the resume, not exact timings. +- Run it nightly with an `#[ignore]` test and a recipe in `rs/justfile`, + beside `quic-sim`. + +## Required + +- [BBR3 aging](/quest/m1/bbr-app-limited.md) - the estimate this run reads falls within seconds of the path narrowing ## Related diff --git a/quest/m2/obs-multitrack.md b/quest/m2/obs-multitrack.md index 80ae07948ba..49363b537d4 100644 --- a/quest/m2/obs-multitrack.md +++ b/quest/m2/obs-multitrack.md @@ -5,10 +5,10 @@ The obs-moq output publishes OBS's native multitrack video encoders as simulcast renditions of one broadcast, one rendition per track, each with its own size and bitrate in the catalog. A user configures them the way OBS -multitrack video is configured, including over obs-websocket. Disabling the -top renditions as the bandwidth grant falls is -[simulcast rung disable](/quest/m1/simulcast-rung-disable.md)'s work, not -this quest's. +multitrack video is configured, including over obs-websocket. The output +already disables a rendition above a lower-bitrate encoder while its grant +cannot fund it (`VideoInit` and `VideoData` in `cpp/obs/src/moq-output.cpp`); +keep that working for every rendition this configures. ## Plan @@ -34,6 +34,6 @@ at least two video tracks watched in ``. ## Related -- [Simulcast rung disable](/quest/m1/simulcast-rung-disable.md) - the grant-driven rung disable these renditions follow +- [Simulcast rung disable](/quest/m1/simulcast-rung-disable.md) - the shaped-uplink proof of the rung disable these renditions follow - [MoQ in obs-studio](/quest/m2/obs-studio/README.md) - the native output this configuration carries into - [OBS native codecs](/quest/m2/obs-moq-video/README.md) - the encoders behind each track when MoQ encoders are chosen diff --git a/quest/m2/probe/README.md b/quest/m2/probe/README.md index 4b594d07d6c..e2140c25f04 100644 --- a/quest/m2/probe/README.md +++ b/quest/m2/probe/README.md @@ -82,7 +82,7 @@ This README's own work, once the children land: ## Required -- [Native rung disable](/quest/m1/simulcast-rung-disable.md) - OBS stops sending the top renditions its grant cannot fund, as the Rust/FFI publishers do, and a shaped uplink through moq-ffi proves the resume +- [Native rung disable proof](/quest/m1/simulcast-rung-disable.md) - a shaped uplink through moq-ffi proves the top rendition stops and resumes over real BBR3 probing - [Padding probes](/quest/m2/probe/padding.md) - a browser probes with a lowest-priority padding stream on an ALR schedule - [JS rung disable](/quest/m2/probe/rung-disable-js.md) - `@moq/publish` applies the native rung rule, recovering through padding - [Viewer targets](/quest/m2/probe/viewer-target.md) - a session sends its viewer handles as its PROBE target, probes to the peer's, splits the receive estimate into viewer grants, and advertises `Increase` where a sender exists diff --git a/rs/moq-ffi/src/test.rs b/rs/moq-ffi/src/test.rs index 736b6153d6a..76429133509 100644 --- a/rs/moq-ffi/src/test.rs +++ b/rs/moq-ffi/src/test.rs @@ -2768,8 +2768,8 @@ async fn video_rung_sits_below_its_lower_and_disables() { let named = |name: &str| MoqMediaTarget::Named { name: Some(name.into()), }; - let low = MoqMediaTrackProducer::video(&broadcast, named("low"), video_init(MoqVideoFormat::Avc3, h264_init())) - .unwrap(); + let low = + MoqMediaTrackProducer::video(&broadcast, named("low"), video_init(MoqVideoFormat::Avc3, h264_init())).unwrap(); let high = MoqMediaTrackProducer::video( &broadcast, named("high"), diff --git a/swift/Sources/Moq/Bandwidth.swift b/swift/Sources/Moq/Bandwidth.swift index 917b4199046..71abf536bee 100644 --- a/swift/Sources/Moq/Bandwidth.swift +++ b/swift/Sources/Moq/Bandwidth.swift @@ -58,3 +58,33 @@ public final class Reservation: Sendable { ffi.update(maxBps: maxBps) } } + +/// What a sender following a ``RateControl`` should be doing. +public typealias RateTarget = MoqRateTarget + +/// Turns a ``Reservation``'s grant into a sender's ``RateTarget``. +/// +/// The same policy moq's own encoders follow: drops apply at once, raises climb at most 25% per +/// second, and moves under 5% are ignored. For a sender that encodes on its own, such as a +/// pre-encoded track: call `update(_:)` on every frame and apply what it returns. +/// +/// Pass `lowerBps`, the next lower rendition's `maxBps`, for a rendition encoded beside lower +/// ones of the same source: a grant below `(maxBps + 2 * lowerBps) / 3` then disables it while +/// a session probes for the reservation. +public final class RateControl: Sendable { + let ffi: MoqRateControl + + public init(maxBps: UInt64, lowerBps: UInt64? = nil) { + ffi = MoqRateControl(policy: MoqRatePolicy(maxBps: maxBps, lowerBps: lowerBps)) + } + + /// What the sender should currently be doing. + public func target() -> RateTarget { + ffi.target() + } + + /// Feed `reservation`'s grant, returning a new target worth applying, or nil. + public func update(_ reservation: Reservation) -> RateTarget? { + ffi.update(reservation: reservation.ffi) + } +} diff --git a/swift/Sources/Moq/Media.swift b/swift/Sources/Moq/Media.swift index 9b995cb9efe..56d2af2e202 100644 --- a/swift/Sources/Moq/Media.swift +++ b/swift/Sources/Moq/Media.swift @@ -226,8 +226,13 @@ public enum Media { TrackProducer(try MoqMediaTrackProducer.audio(broadcast: broadcast.ffi, target: target.ffi, init: initData)) } /// Import complete video frames on a named or requested track. - public static func video(broadcast: BroadcastProducer, initData: VideoInit, target: Target = .named(name: nil)) throws -> TrackProducer { - TrackProducer(try MoqMediaTrackProducer.video(broadcast: broadcast.ffi, target: target.ffi, init: initData)) + /// + /// Pass `lower`, the next lower rendition of the same source, to publish this track one + /// priority below it, so a bandwidth allocator funds that one first. + public static func video(broadcast: BroadcastProducer, initData: VideoInit, target: Target = .named(name: nil), lower: TrackProducer? = nil) throws -> TrackProducer { + var initData = initData + if let lower { initData.lower = lower.ffi } + return TrackProducer(try MoqMediaTrackProducer.video(broadcast: broadcast.ffi, target: target.ffi, init: initData)) } init(_ ffi: MoqMediaTrackProducer) { self.ffi = ffi @@ -251,6 +256,14 @@ public enum Media { try ffi.discontinuity() } + /// Advertise whether this video rendition is being produced, as its catalog `enabled`. + /// + /// Disable it when its sender stops for lack of bandwidth and mark a ``discontinuity()``; + /// enable it again just before the keyframe that resumes it. An audio track refuses. + public func setEnabled(_ enabled: Bool) throws { + try ffi.setEnabled(enabled: enabled) + } + /// Draw a group boundary here. /// /// Audio has no boundary of its own (every packet is independently decodable), so this is the @@ -319,8 +332,12 @@ public enum Media { let ffi: MoqMediaTrackStreamProducer /// Import a video byte stream, inferring frame boundaries. - public static func video(broadcast: BroadcastProducer, initData: VideoInit, target: Target = .named(name: nil)) throws -> TrackStreamProducer { - TrackStreamProducer(try MoqMediaTrackStreamProducer.video(broadcast: broadcast.ffi, target: target.ffi, init: initData)) + /// + /// `lower` is as for ``TrackProducer/video(broadcast:initData:target:lower:)``. + public static func video(broadcast: BroadcastProducer, initData: VideoInit, target: Target = .named(name: nil), lower: TrackProducer? = nil) throws -> TrackStreamProducer { + var initData = initData + if let lower { initData.lower = lower.ffi } + return TrackStreamProducer(try MoqMediaTrackStreamProducer.video(broadcast: broadcast.ffi, target: target.ffi, init: initData)) } init(_ ffi: MoqMediaTrackStreamProducer) { self.ffi = ffi From 20a56a358e7fd4ab2eefc9c9536145dd99a5adbf Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 29 Oct 2026 18:12:32 -0700 Subject: [PATCH 04/14] fix(obs): wait for a lower rung's headers before starting its track A higher rung's first packet started its lower rung's track, before an encoder that supplies parameter sets only after its first frame had them, so the lower track was cached without them. Wait for the lower's own first packet instead, and open every track on a keyframe. Co-Authored-By: Claude Opus 5.5 --- cpp/obs/src/moq-output.cpp | 13 +++++++- cpp/obs/test/moq-output-test.cpp | 52 ++++++++++++++++++++++++++++++++ 2 files changed, 64 insertions(+), 1 deletion(-) diff --git a/cpp/obs/src/moq-output.cpp b/cpp/obs/src/moq-output.cpp index 8461998562b..6fdde6dde60 100644 --- a/cpp/obs/src/moq-output.cpp +++ b/cpp/obs/src/moq-output.cpp @@ -667,6 +667,9 @@ void MoQOutput::VideoData(struct encoder_packet *packet) auto it = video_tracks.find(encoder); if (it == video_tracks.end()) { + // A track opens on a keyframe; deltas before it have no group to join. + if (!packet->keyframe) + return; VideoInit(encoder); it = video_tracks.find(encoder); } @@ -827,8 +830,16 @@ void MoQOutput::VideoInit(obs_encoder_t *encoder) } std::optional lower_bps; if (lower) { - if (video_tracks.find(lower) == video_tracks.end()) + if (video_tracks.find(lower) == video_tracks.end()) { + // The lower track keeps the headers it starts with, and some encoders only have + // them after their first frame. Wait for its first packet to start it instead; + // this rendition opens on its next keyframe after that. + uint8_t *lower_data = nullptr; + size_t lower_size = 0; + if (!obs_encoder_get_extra_data(lower, &lower_data, &lower_size) || lower_size == 0) + return; VideoInit(lower); + } auto below = video_tracks.find(lower); if (below != video_tracks.end() && below->second.producer) { config.lower = below->second.producer; diff --git a/cpp/obs/test/moq-output-test.cpp b/cpp/obs/test/moq-output-test.cpp index e3e7e70342e..67b2b009d60 100644 --- a/cpp/obs/test/moq-output-test.cpp +++ b/cpp/obs/test/moq-output-test.cpp @@ -80,6 +80,8 @@ obs_encoder_t *const AUDIO_ENCODER = reinterpret_cast(0x3); // A second, 2 Mbps video encoder below VIDEO_ENCODER's 8 Mbps, while g_ladder is set. obs_encoder_t *const LOW_ENCODER = reinterpret_cast(0x5); std::atomic g_ladder{false}; +// LOW_ENCODER has no headers yet, as an encoder that only has them after its first frame. +std::atomic g_low_headers_pending{false}; } // namespace extern "C" { @@ -156,6 +158,11 @@ void obs_register_output_s(const struct obs_output_info *, size_t) {} bool obs_encoder_get_extra_data(const obs_encoder_t *encoder, uint8_t **data, size_t *size) { + if (encoder == LOW_ENCODER && g_low_headers_pending) { + *data = nullptr; + *size = 0; + return false; + } const auto &extra = encoder == AUDIO_ENCODER ? g_opus_head : g_h264_init; *data = const_cast(extra.data()); *size = extra.size(); @@ -305,6 +312,7 @@ void reset(std::string url) g_rate_control = "CBR"; g_encoder_caps = OBS_ENCODER_CAP_DYN_BITRATE; g_ladder = false; + g_low_headers_pending = false; { std::lock_guard lock(g_updates_mutex); g_updates.clear(); @@ -724,6 +732,50 @@ int main() } printf("a ladder rung disables, drops packets, and resumes on a keyframe: ok\n"); + // A higher rung whose lower encoder has no headers yet waits for the lower's first + // packet rather than start the lower track without them, then opens on its own + // next keyframe one priority below it. + { + TestRelay relay; + reset(relay.Url()); + g_ladder = true; + g_encoder_caps = 0; + g_low_headers_pending = true; + MoQOutput o(nullptr, OUTPUT); + CHECK(o.Start()); + CHECK(WaitFor([&] { return o.IsLiveSession(); })); + + auto keyframe = TestH264Keyframe(); + std::vector delta = {0x00, 0x00, 0x00, 0x01, 0x41, 0x9a, 0x00, 0x33}; + auto early = videoPacket(keyframe, 0); + o.Data(&early); + CHECK(o.GetTotalBytes() == 0); + CHECK(!MoQOutputStatsTest::Published(o, LOW_ENCODER).has_value()); + + g_low_headers_pending = false; + auto first = videoPacket(keyframe, 0); + first.encoder = LOW_ENCODER; + o.Data(&first); + size_t sent = keyframe.size(); + CHECK(o.GetTotalBytes() == sent); + + auto skipped = videoPacket(delta, 1); + skipped.keyframe = false; + o.Data(&skipped); + CHECK(o.GetTotalBytes() == sent); + auto opened = videoPacket(keyframe, 2); + o.Data(&opened); + sent += keyframe.size(); + CHECK(o.GetTotalBytes() == sent); + + const auto high = MoQOutputStatsTest::Published(o, VIDEO_ENCODER); + const auto low = MoQOutputStatsTest::Published(o, LOW_ENCODER); + CHECK(high.has_value() && low.has_value()); + CHECK(high && low && high->second == low->second - 1); + o.Stop(); + } + printf("a ladder rung waits for its lower rung's headers: ok\n"); + // Stop drains the session, even when the output is destroyed straight after: a // large frame written just before it still reaches the relay, and the track then // ends cleanly instead of being cut off with the session. From ba63bd17c71dc8ca51e92ad1e05988c445be459f Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 29 Oct 2026 18:22:12 -0700 Subject: [PATCH 05/14] fix(obs): a rung waits while its lower rung waits on one further down Co-Authored-By: Claude Opus 5.5 --- cpp/obs/src/moq-output.cpp | 5 ++++- cpp/obs/test/moq-output-test.cpp | 25 ++++++++++++++++++------- 2 files changed, 22 insertions(+), 8 deletions(-) diff --git a/cpp/obs/src/moq-output.cpp b/cpp/obs/src/moq-output.cpp index 6fdde6dde60..571401db910 100644 --- a/cpp/obs/src/moq-output.cpp +++ b/cpp/obs/src/moq-output.cpp @@ -840,8 +840,11 @@ void MoQOutput::VideoInit(obs_encoder_t *encoder) return; VideoInit(lower); } + // The lower rendition waits on one further down, so this one waits with it. auto below = video_tracks.find(lower); - if (below != video_tracks.end() && below->second.producer) { + if (below == video_tracks.end()) + return; + if (below->second.producer) { config.lower = below->second.producer; lower_bps = (uint64_t)lower_kbps * 1000ULL; } diff --git a/cpp/obs/test/moq-output-test.cpp b/cpp/obs/test/moq-output-test.cpp index 67b2b009d60..6a5fbb25df2 100644 --- a/cpp/obs/test/moq-output-test.cpp +++ b/cpp/obs/test/moq-output-test.cpp @@ -80,6 +80,9 @@ obs_encoder_t *const AUDIO_ENCODER = reinterpret_cast(0x3); // A second, 2 Mbps video encoder below VIDEO_ENCODER's 8 Mbps, while g_ladder is set. obs_encoder_t *const LOW_ENCODER = reinterpret_cast(0x5); std::atomic g_ladder{false}; +// A third, 4 Mbps video encoder between the two, while g_ladder and g_mid are set. +obs_encoder_t *const MID_ENCODER = reinterpret_cast(0x6); +std::atomic g_mid{false}; // LOW_ENCODER has no headers yet, as an encoder that only has them after its first frame. std::atomic g_low_headers_pending{false}; } // namespace @@ -121,7 +124,9 @@ obs_encoder_t *obs_output_get_video_encoder2(const obs_output_t *, size_t idx) { if (idx == 0) return VIDEO_ENCODER; - return idx == 1 && g_ladder ? LOW_ENCODER : nullptr; + if (idx == 1 && g_ladder) + return LOW_ENCODER; + return idx == 2 && g_ladder && g_mid ? MID_ENCODER : nullptr; } bool obs_output_begin_data_capture(obs_output_t *, uint32_t) @@ -199,7 +204,9 @@ const char *obs_data_get_string(obs_data_t *, const char *) long long obs_data_get_int(obs_data_t *settings, const char *) { - return settings == reinterpret_cast(LOW_ENCODER) ? 2000 : 8000; + if (settings == reinterpret_cast(LOW_ENCODER)) + return 2000; + return settings == reinterpret_cast(MID_ENCODER) ? 4000 : 8000; } // The settings a retune builds carry only the bitrate, so the stub is that bitrate. @@ -312,6 +319,7 @@ void reset(std::string url) g_rate_control = "CBR"; g_encoder_caps = OBS_ENCODER_CAP_DYN_BITRATE; g_ladder = false; + g_mid = false; g_low_headers_pending = false; { std::lock_guard lock(g_updates_mutex); @@ -732,13 +740,14 @@ int main() } printf("a ladder rung disables, drops packets, and resumes on a keyframe: ok\n"); - // A higher rung whose lower encoder has no headers yet waits for the lower's first - // packet rather than start the lower track without them, then opens on its own - // next keyframe one priority below it. + // A rung whose lower encoder has no headers yet waits for the lower's first packet + // rather than start the lower track without them, and so does every rung above it. + // Each then opens on its next keyframe one priority below its lower. { TestRelay relay; reset(relay.Url()); g_ladder = true; + g_mid = true; g_encoder_caps = 0; g_low_headers_pending = true; MoQOutput o(nullptr, OUTPUT); @@ -751,6 +760,7 @@ int main() o.Data(&early); CHECK(o.GetTotalBytes() == 0); CHECK(!MoQOutputStatsTest::Published(o, LOW_ENCODER).has_value()); + CHECK(!MoQOutputStatsTest::Published(o, MID_ENCODER).has_value()); g_low_headers_pending = false; auto first = videoPacket(keyframe, 0); @@ -769,9 +779,10 @@ int main() CHECK(o.GetTotalBytes() == sent); const auto high = MoQOutputStatsTest::Published(o, VIDEO_ENCODER); + const auto mid = MoQOutputStatsTest::Published(o, MID_ENCODER); const auto low = MoQOutputStatsTest::Published(o, LOW_ENCODER); - CHECK(high.has_value() && low.has_value()); - CHECK(high && low && high->second == low->second - 1); + CHECK(high.has_value() && mid.has_value() && low.has_value()); + CHECK(high && mid && low && high->second == mid->second - 1 && mid->second == low->second - 1); o.Stop(); } printf("a ladder rung waits for its lower rung's headers: ok\n"); From c8a11afcb4f12303075221aac26293418bac6549 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 30 Oct 2026 20:55:36 -0700 Subject: [PATCH 06/14] fix(py): re-export RateTarget through types so pyright sees it public session.py imported MoqRateTarget under an alias without an __all__, so pyright's reportPrivateImportUsage rejected moq/__init__.py importing it. types.py already re-exports the other moq-ffi records with an __all__. Co-Authored-By: Claude Opus 5.5 --- py/moq-rs/moq/__init__.py | 2 +- py/moq-rs/moq/session.py | 3 +-- py/moq-rs/moq/types.py | 4 ++++ 3 files changed, 6 insertions(+), 3 deletions(-) diff --git a/py/moq-rs/moq/__init__.py b/py/moq-rs/moq/__init__.py index b48699392c6..bce75ad93e6 100644 --- a/py/moq-rs/moq/__init__.py +++ b/py/moq-rs/moq/__init__.py @@ -46,7 +46,6 @@ BandwidthConsumer, GrantConsumer, RateControl, - RateTarget, Reservation, Session, Tokens, @@ -77,6 +76,7 @@ ProtocolKind, QuicConfig, Range, + RateTarget, Route, Subscription, TrackInfo, diff --git a/py/moq-rs/moq/session.py b/py/moq-rs/moq/session.py index e066dbf06de..4a047a7e158 100644 --- a/py/moq-rs/moq/session.py +++ b/py/moq-rs/moq/session.py @@ -14,13 +14,12 @@ MoqSession, MoqTokens, ) -from moq_ffi import MoqRateTarget as RateTarget from ._records import _strs from .errors import is_shutdown from .origin import OriginConsumer, OriginProducer from .publish import TrackDemand -from .types import ConnectionState, ConnectionStats, Grant +from .types import ConnectionState, ConnectionStats, Grant, RateTarget class Session: diff --git a/py/moq-rs/moq/types.py b/py/moq-rs/moq/types.py index d65eaa159cf..414a2b7455f 100644 --- a/py/moq-rs/moq/types.py +++ b/py/moq-rs/moq/types.py @@ -42,6 +42,9 @@ from moq_ffi import ( MoqRange as Range, ) +from moq_ffi import ( + MoqRateTarget as RateTarget, +) from moq_ffi import ( MoqRoute as Route, ) @@ -101,6 +104,7 @@ "ProtocolKind", "QuicConfig", "Range", + "RateTarget", "Route", "Subscription", "TrackInfo", From 11481709945dee87e083959d24364cac06660c57 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 31 Oct 2026 02:42:14 -0700 Subject: [PATCH 07/14] doc(obs): only a rung that follows the estimate stops sending Co-Authored-By: Claude Opus 5.5 --- doc/bin/obs.md | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/doc/bin/obs.md b/doc/bin/obs.md index d12d7a0e7e7..37beea9d06a 100644 --- a/doc/bin/obs.md +++ b/doc/bin/obs.md @@ -46,10 +46,12 @@ recording at the full rate. With several video encoders on the output, each one is a rendition, and one above a lower-bitrate encoder stops sending while its share of the estimate -cannot fund it, advertised as disabled so viewers switch down. It keeps -encoding, since OBS cannot pause one encoder of a live output, and resumes at -that encoder's next keyframe once the estimate recovers. This needs the -session to probe for bandwidth, which takes moq-quic with BBR3. +cannot fund it, advertised as disabled so viewers switch down. Only an encoder +that follows the estimate, as above, stops; one that cannot retune while live, +or uses CQP or CRF, keeps sending. A stopped encoder keeps encoding, since OBS +cannot pause one encoder of a live output, and resumes at that encoder's next +keyframe once the estimate recovers. This needs the session to probe for +bandwidth, which takes moq-quic with BBR3. ## Source quality and moq-transcode From 92046c6bfd0563db7e4215b5a631d91dee11d8de Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 31 Oct 2026 03:01:35 -0700 Subject: [PATCH 08/14] fix(obs): keep the init's parameter sets when a rung disables before its first frame The first keyframe opens the track from the encoder's extra data, whose SPS/PPS the splitter holds until a keyframe adopts them. A Disabled target on that same packet marked a discontinuity, which resets the splitter and drops them, so an encoder that sends headers only as extra data resumed with bare keyframes. Skip the marker until a frame is written: there is no timeline to break yet. Co-Authored-By: Claude Opus 5.5 --- cpp/obs/src/moq-output.cpp | 11 +++++-- cpp/obs/src/moq-output.h | 3 ++ cpp/obs/test/moq-output-test.cpp | 54 ++++++++++++++++++++++++++++++++ 3 files changed, 65 insertions(+), 3 deletions(-) diff --git a/cpp/obs/src/moq-output.cpp b/cpp/obs/src/moq-output.cpp index 571401db910..3990c6b93f0 100644 --- a/cpp/obs/src/moq-output.cpp +++ b/cpp/obs/src/moq-output.cpp @@ -708,6 +708,7 @@ void MoQOutput::VideoData(struct encoder_packet *packet) LOG_ERROR("Failed to write video frame: %s", result.error().to_string().c_str()); return; } + track.started = true; total_bytes_sent += packet->size; } @@ -727,9 +728,13 @@ void MoQOutput::Retarget(obs_encoder_t *encoder, Track &track, const moq::RateTa return; track.sending = false; // The lower renditions keep their share; viewers switch to one instead of - // waiting on frames that stopped coming. - if (auto result = track.producer->discontinuity(); !result) - LOG_ERROR("Failed to mark a disabled video track: %s", result.error().to_string().c_str()); + // waiting on frames that stopped coming. Before the first frame there is no + // timeline to break, and the marker would drop the init's parameter sets, which + // an encoder that sends them only as extra data never repeats. + if (track.started) { + if (auto result = track.producer->discontinuity(); !result) + LOG_ERROR("Failed to mark a disabled video track: %s", result.error().to_string().c_str()); + } if (auto result = track.producer->set_enabled(false); !result) LOG_ERROR("Failed to disable video track: %s", result.error().to_string().c_str()); } diff --git a/cpp/obs/src/moq-output.h b/cpp/obs/src/moq-output.h index 50220f265cb..5c222c37f3e 100644 --- a/cpp/obs/src/moq-output.h +++ b/cpp/obs/src/moq-output.h @@ -140,6 +140,9 @@ class MoQOutput { bool enabled = true; // False from a disable until the keyframe that resumes it, dropping packets. bool sending = true; + // Whether a frame was written. Until one is, the importer still holds the + // init's parameter sets, which a discontinuity would discard. + bool started = false; }; void VideoInit(obs_encoder_t *encoder); diff --git a/cpp/obs/test/moq-output-test.cpp b/cpp/obs/test/moq-output-test.cpp index 6a5fbb25df2..041fe2fc018 100644 --- a/cpp/obs/test/moq-output-test.cpp +++ b/cpp/obs/test/moq-output-test.cpp @@ -9,6 +9,7 @@ // // Run with `just obs test` (ThreadSanitizer) or `just obs ci` (plain). This is not // part of the plugin build. +#include #include #include #include @@ -460,6 +461,13 @@ struct MoQOutputStatsTest { } return true; } + // Open `encoder`'s track from its extra data, as its first keyframe does just + // before the track's rate control is first read. + static void Init(MoQOutput &output, obs_encoder_t *encoder) + { + std::lock_guard lock(output.media_mutex); + output.VideoInit(encoder); + } // Apply `target` to `encoder`'s track, as a grant the track's rate control returned. static void Retarget(MoQOutput &output, obs_encoder_t *encoder, const moq::RateTarget &target) { @@ -740,6 +748,52 @@ int main() } printf("a ladder rung disables, drops packets, and resumes on a keyframe: ok\n"); + // A rung disabled before its first frame keeps the extra data's parameter sets, + // so a keyframe without them still resumes it decodable. + { + TestRelay relay; + reset(relay.Url()); + g_encoder_caps = 0; + MoQOutput o(nullptr, OUTPUT); + CHECK(o.Start()); + CHECK(WaitFor([&] { return o.IsLiveSession(); })); + + MoQOutputStatsTest::Init(o, VIDEO_ENCODER); + MoQOutputStatsTest::Retarget(o, VIDEO_ENCODER, moq::RateTarget::kDisabled{}); + MoQOutputStatsTest::Retarget(o, VIDEO_ENCODER, moq::RateTarget::kEnabled{4'000'000}); + std::vector idr = {0x00, 0x00, 0x00, 0x01, 0x65, 0x88, 0x84, 0x00, 0x33, 0xff}; + auto resumed = videoPacket(idr, 0); + o.Data(&resumed); + CHECK(o.GetTotalBytes() == idr.size()); + + auto broadcast = relayBroadcast(relay); + auto catalog = + relayCatalog(broadcast, [](const moq::Catalog &catalog) { return !catalog.video.empty(); }); + CHECK(catalog.has_value() && catalog->video.size() == 1); + if (catalog && catalog->video.size() == 1) { + const auto &[name, rendition] = *catalog->video.begin(); + auto media = TestOk( + moq::MediaTrackConsumer::subscribe(broadcast, {moq::Rendition::kVideo{name, rendition}}) + .get(), + "container consumer"); + auto next = media->next(); + CHECK(next.wait_for(std::chrono::seconds(10)) == std::future_status::ready); + auto first = TestOk(next.get(), "first frame"); + const auto *frame = first ? std::get_if(&first->get_variant()) + : nullptr; + CHECK(frame != nullptr); + if (frame) { + const auto &payload = frame->frame.payload; + // The start of the extra data's SPS, past its start code. + const auto init = TestH264Init(); + CHECK(std::search(payload.begin(), payload.end(), init.begin() + 4, + init.begin() + 12) != payload.end()); + } + } + o.Stop(); + } + printf("a rung disabled before its first frame keeps its parameter sets: ok\n"); + // A rung whose lower encoder has no headers yet waits for the lower's first packet // rather than start the lower track without them, and so does every rung above it. // Each then opens on its next keyframe one priority below its lower. From a7959e2e74b80b97b70628e701d3acf0858b41d3 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 31 Oct 2026 23:43:32 -0700 Subject: [PATCH 09/14] fix(obs): roll back a rung resume whose keyframe fails to write Validate the resume keyframe's timestamp before re-enabling the track, and disable it again if the write fails, so the rung retries on the next keyframe instead of advertising enabled with no open group. Co-Authored-By: Claude Opus 5.5 --- cpp/obs/src/moq-output.cpp | 26 ++++++++++++++++++-------- 1 file changed, 18 insertions(+), 8 deletions(-) diff --git a/cpp/obs/src/moq-output.cpp b/cpp/obs/src/moq-output.cpp index 3990c6b93f0..c101de2c3c2 100644 --- a/cpp/obs/src/moq-output.cpp +++ b/cpp/obs/src/moq-output.cpp @@ -688,24 +688,34 @@ void MoQOutput::VideoData(struct encoder_packet *packet) // OBS can't stop one encoder of a running output, or force it to keyframe, so a // disabled rung keeps encoding and its packets are dropped here. It resumes on // the encoder's next natural keyframe once the grant funds it again. + if (!track.sending && (!track.enabled || !packet->keyframe)) + return; + + auto timestamp = PacketTimestamp(packet); + if (!timestamp) { + LOG_WARNING("Dropping video frame with negative PTS: %lld", (long long)packet->pts); + return; + } + + // A resume that fails to write its keyframe stays disabled, so it retries on the + // next keyframe instead of advertising a rung whose deltas have no group to join. + bool resumed = false; if (!track.sending) { - if (!track.enabled || !packet->keyframe) - return; if (auto result = track.producer->set_enabled(true); !result) { LOG_ERROR("Failed to enable video track: %s", result.error().to_string().c_str()); return; } track.sending = true; - } - - auto timestamp = PacketTimestamp(packet); - if (!timestamp) { - LOG_WARNING("Dropping video frame with negative PTS: %lld", (long long)packet->pts); - return; + resumed = true; } if (auto result = track.producer->write_frame(PacketFrame(packet, *timestamp)); !result) { LOG_ERROR("Failed to write video frame: %s", result.error().to_string().c_str()); + if (resumed) { + track.sending = false; + if (auto disable = track.producer->set_enabled(false); !disable) + LOG_ERROR("Failed to disable video track: %s", disable.error().to_string().c_str()); + } return; } track.started = true; From d7e63636f659ad1e7dca2bbe254e4c0b69648a08 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 31 Oct 2026 23:53:00 -0700 Subject: [PATCH 10/14] test(moq-ffi): wait on the ambient clock in the rung catalog test Main's workspace clippy.toml now bans tokio::time::timeout. Co-Authored-By: Claude Opus 5.5 --- rs/moq-ffi/src/test.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/rs/moq-ffi/src/test.rs b/rs/moq-ffi/src/test.rs index 1fa076523ea..55c0fdbb7cf 100644 --- a/rs/moq-ffi/src/test.rs +++ b/rs/moq-ffi/src/test.rs @@ -2823,7 +2823,7 @@ async fn video_rung_sits_below_its_lower_and_disables() { high.set_enabled(false).unwrap(); let disabled = loop { - let catalog = tokio::time::timeout(TIMEOUT, catalog_consumer.next()) + let catalog = moq_clock::timeout(TIMEOUT, catalog_consumer.next()) .await .expect("timed out waiting for catalog") .unwrap() From e8f805ad4e88992ae37bbeaf80081d614539a172 Mon Sep 17 00:00:00 2001 From: quest fixer Date: Sun, 1 Nov 2026 00:41:08 -0700 Subject: [PATCH 11/14] fix(ffi): follow base clock move into kio --- rs/moq-ffi/src/bandwidth.rs | 2 +- rs/moq-ffi/src/test.rs | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/rs/moq-ffi/src/bandwidth.rs b/rs/moq-ffi/src/bandwidth.rs index 9b1e4d958bd..3deefa4af9f 100644 --- a/rs/moq-ffi/src/bandwidth.rs +++ b/rs/moq-ffi/src/bandwidth.rs @@ -162,7 +162,7 @@ impl MoqRateControl { let grant = reservation.inner.consume(); let mut control = self.inner.lock().unwrap(); control.set_probed(grant.probed()); - control.update(grant.peek(), moq_clock::now()).map(Into::into) + control.update(grant.peek(), kio::clock::now()).map(Into::into) } } diff --git a/rs/moq-ffi/src/test.rs b/rs/moq-ffi/src/test.rs index 3d1e8877665..8cbee55db8e 100644 --- a/rs/moq-ffi/src/test.rs +++ b/rs/moq-ffi/src/test.rs @@ -2823,7 +2823,7 @@ async fn video_rung_sits_below_its_lower_and_disables() { high.set_enabled(false).unwrap(); let disabled = loop { - let catalog = moq_clock::timeout(TIMEOUT, catalog_consumer.next()) + let catalog = kio::clock::timeout(TIMEOUT, catalog_consumer.next()) .await .expect("timed out waiting for catalog") .unwrap() From faf9fd307fec91f5673fd191f518900bffd80d9a Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 1 Nov 2026 08:46:11 -0800 Subject: [PATCH 12/14] style(swift): format the merged Media.swift video constructor Co-Authored-By: Claude Opus 5.5 --- swift/Sources/Moq/Media.swift | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/swift/Sources/Moq/Media.swift b/swift/Sources/Moq/Media.swift index 937d0e5aee0..5c9b2830bfb 100644 --- a/swift/Sources/Moq/Media.swift +++ b/swift/Sources/Moq/Media.swift @@ -246,7 +246,8 @@ public enum Media { ) throws -> TrackProducer { var initData = initData if let lower { initData.lower = lower.ffi } - return TrackProducer(try MoqMediaTrackProducer.video(broadcast: broadcast.ffi, target: target.ffi, init: initData)) + return TrackProducer( + try MoqMediaTrackProducer.video(broadcast: broadcast.ffi, target: target.ffi, init: initData)) } init(_ ffi: MoqMediaTrackProducer) { self.ffi = ffi From c672fbec308ebbd810390804914d335d22906993 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 1 Nov 2026 08:52:09 -0800 Subject: [PATCH 13/14] docs(obs): drop the restated rendition intro from the rung-disable paragraph Main's simulcast section (#7333) already says every encoder is a rendition. Co-Authored-By: Claude Opus 5.5 --- doc/bin/obs.md | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/doc/bin/obs.md b/doc/bin/obs.md index 0d352c028a9..b7ed6ffff38 100644 --- a/doc/bin/obs.md +++ b/doc/bin/obs.md @@ -67,9 +67,8 @@ Saving Settings > Stream afterwards replaces the MoQ service. The dock's Go Live publishes one rendition. OBS's Enhanced Broadcasting (multitrack video) always streams through its RTMP output, so it cannot drive MoQ. -With several video encoders on the output, each one is a rendition, and one -above a lower-bitrate encoder stops sending while its share of the estimate -cannot fund it, advertised as disabled so viewers switch down. Only an encoder +A rendition above a lower-bitrate one stops sending while its share of the +estimate cannot fund it, advertised as disabled so viewers switch down. Only an encoder that follows the estimate, as above, stops; one that cannot retune while live, or uses CQP or CRF, keeps sending. A stopped encoder keeps encoding, since OBS cannot pause one encoder of a live output, and resumes at that encoder's next From 0238c502f2f6ea01ad0705bf74b81f2632bfd085 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 2 Nov 2026 20:11:59 -0800 Subject: [PATCH 14/14] chore(dart): regenerate moq_ffi bindings after merging main Co-Authored-By: Claude Opus 5.5 --- dart/moq_ffi/lib/src/moq.dart | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/dart/moq_ffi/lib/src/moq.dart b/dart/moq_ffi/lib/src/moq.dart index 344bf443312..a0c2169c173 100644 --- a/dart/moq_ffi/lib/src/moq.dart +++ b/dart/moq_ffi/lib/src/moq.dart @@ -9886,9 +9886,15 @@ class MoqRateControl implements MoqRateControlInterface { if (_isDisposed) { return; } + final status = calloc(); _isDisposed = true; _MoqRateControlFinalizer.detach(this); - rustCall((status) => uniffi_moq_ffi_fn_free_moqratecontrol(_ptr, status)); + try { + uniffi_moq_ffi_fn_free_moqratecontrol(_ptr, status); + checkCallStatus(NullRustCallStatusErrorHandler(), status); + } finally { + calloc.free(status); + } } MoqRateTarget target() {