- commit
- 0f1134c9251d804860bdf938ec01afd68455bfda
- parent
- 0d9d3bb5001cc4ee35477eb9c68bc4f880484769
- author
- Dami <dami.etoile@gmail.com>
@@ -8,7 +8,7 @@ use std::{
88 };99 1010 use anyhow::{Context, Result, bail, ensure};11-use clap::Parser;11+use clap::{Parser, ValueEnum};1212 use iroh::{Endpoint, EndpointAddr, EndpointId, SecretKey, endpoint::presets};1313 use iroh_moq::{Moq, MoqSession};1414 use moq_net::{BroadcastConsumer, GroupConsumer, TrackConsumer};@@ -47,6 +47,13 @@ use moq_catalog::{
4747 4848 const MEDIA_OBJECT_CAPACITY: usize = 4;4949 50+51+enum AdaptiveProfile {52+ Clean,53+ FloorBreach,54+ Stale,55+}56+5057 5158 5259 struct Args {@@ -93,6 +100,15 @@ struct Args {
93100 /// observing this many successor generations.94101 95102 focus_handoffs: u32,103+ /// Drive a bounded per-viewer adaptive profile on control/2. Clean viewers104+ /// prove survivor recovery; floor-breach viewers prove isolated detach.105+ 106+107+108+109+110+111+ adaptive_profile: Option<AdaptiveProfile>,96112 /// Request a configured recovery keyframe after three accepted frames,97113 /// then prove no delta history is delivered before the recovery barrier.98114 @@ -765,6 +781,7 @@ async fn run_control_v2_smoke(
765781 spectator: bool,766782 disconnect_with_focus: bool,767783 focus_handoffs: u32,784+ adaptive_profile: Option<AdaptiveProfile>,768785 ) -> Result<()> {769786 let control_connection = endpoint770787 .connect(address.clone(), CONTROL_ALPN_V2)@@ -846,6 +863,78 @@ async fn run_control_v2_smoke(
846863 .await;847864 }848865 if spectator {866+ if let Some(profile) = adaptive_profile {867+ let mut accepted = 0_u32;868+ let mut last_sequence = None;869+ let feedback =870+ run_v2_adaptive_profile(endpoint, address.clone(), nonce, session_id, profile);871+ tokio::pin!(feedback);872+ let mut media_open = true;873+ let evidence = loop {874+ tokio::select! {875+ evidence = &mut feedback => break evidence?,876+ outcome = media.next(), if media_open => {877+ let Some(outcome) = outcome? else {878+ if profile == AdaptiveProfile::FloorBreach {879+ media_open = false;880+ continue;881+ }882+ bail!("adaptive-profile spectator media ended early");883+ };884+ let MoqProbeOutcome::Frame { frame, .. } = outcome else {885+ continue;886+ };887+ if let Some((width, height)) = expected_size {888+ ensure!(889+ frame.header.width == width && frame.header.height == height,890+ "adaptive-profile media dimensions did not match {width}x{height}"891+ );892+ }893+ if let Some(previous) = last_sequence {894+ ensure!(895+ frame.header.sequence > previous,896+ "adaptive-profile media sequence regressed"897+ );898+ }899+ last_sequence = Some(frame.header.sequence);900+ accepted += 1;901+ }902+ }903+ };904+ ensure!(905+ accepted >= frames.min(60),906+ "adaptive profile did not preserve media progress"907+ );908+ control_connection.close(0_u32.into(), b"adaptive profile complete");909+ println!("probe=ok");910+ println!("transport=iroh-moq");911+ println!("control_alpn=sigil/control/2");912+ println!("media_generation_id={generation_id}");913+ println!("media_broadcast_name={broadcast_name}");914+ println!("roster_viewers={}", initial.viewers.len());915+ println!("frames={accepted}");916+ println!("spectator=ok");917+ println!("adaptive_profile={}", profile.label());918+ println!("adaptive_minimum_kbps={}", evidence.minimum_kbps);919+ println!("adaptive_final_kbps={}", evidence.final_kbps);920+ println!(921+ "adaptive_survivor_recovery={}",922+ if evidence.survivor_recovery {923+ "ok"924+ } else {925+ "not-applicable"926+ }927+ );928+ println!(929+ "adaptive_floor_detach={}",930+ if evidence.floor_detach {931+ "ok"932+ } else {933+ "not-applicable"934+ }935+ );936+ return Ok(());937+ }849938 let mut accepted = 0_u32;850939 let mut last_sequence = None;851940 while accepted < frames {@@ -1493,6 +1582,134 @@ async fn run_feedback_smoke(
14931582 Ok((decision, path_mode, path_rtt_ms))14941583 }14951584 1585+impl AdaptiveProfile {1586+ const fn label(self) -> &'static str {1587+ match self {1588+ Self::Clean => "clean",1589+ Self::FloorBreach => "floor-breach",1590+ Self::Stale => "stale",1591+ }1592+ }1593+}1594+1595+struct AdaptiveProfileEvidence {1596+ minimum_kbps: u32,1597+ final_kbps: u32,1598+ survivor_recovery: bool,1599+ floor_detach: bool,1600+}1601+1602+async fn run_v2_adaptive_profile(1603+ endpoint: &Endpoint,1604+ address: EndpointAddr,1605+ nonce: [u8; 16],1606+ session_id: u64,1607+ profile: AdaptiveProfile,1608+) -> Result<AdaptiveProfileEvidence> {1609+ let connection = endpoint1610+ .connect(address, MEDIA_FEEDBACK_ALPN_V1)1611+ .await1612+ .context("connecting control-v2 adaptive feedback profile")?;1613+ let (mut send, mut recv) = connection1614+ .open_bi()1615+ .await1616+ .context("opening control-v2 adaptive feedback stream")?;1617+ let negotiation = negotiate(1618+ &mut send,1619+ &mut recv,1620+ nonce,1621+ vec![Capability::VideoH264],1622+ Capability::VideoH264,1623+ "control-v2 adaptive feedback",1624+ None,1625+ )1626+ .await?;1627+ ensure!(1628+ negotiation.session_id == session_id,1629+ "control-v2 media and adaptive feedback session IDs differ"1630+ );1631+1632+ let report_count = match profile {1633+ AdaptiveProfile::Clean => 36,1634+ AdaptiveProfile::FloorBreach => 32,1635+ AdaptiveProfile::Stale => 5,1636+ };1637+ let mut minimum_kbps = u32::MAX;1638+ let mut maximum_kbps = 0_u32;1639+ let mut final_kbps = 0_u32;1640+ let mut saw_stale = false;1641+ let mut floor_detach = false;1642+ let mut final_reason_bits = 0_u16;1643+ for report_id in 1..=report_count {1644+ let mut report = complete_feedback_report(report_id, MediaFeedbackFlags::NONE, report_id);1645+ match profile {1646+ AdaptiveProfile::Clean => {}1647+ AdaptiveProfile::FloorBreach => {1648+ report.flags = MediaFeedbackFlags::RESYNC_ACTIVE;1649+ report.transport_dropped_delta = 8;1650+ report.frontend_dropped_delta = 8;1651+ report.decoder_dropped_delta = 8;1652+ report.presenter_dropped_delta = 8;1653+ report.frontend_queue_depth = report.frontend_queue_capacity;1654+ report.decode_queue_depth = report.decode_queue_capacity;1655+ report.presenter_queue_depth = report.presenter_queue_capacity;1656+ report.transport_delivery_p95_ms = Some(250);1657+ report.decode_p95_ms = Some(125);1658+ report.presentation_p95_ms = Some(125);1659+ }1660+ AdaptiveProfile::Stale => {1661+ report.interval_ms = 5_000;1662+ }1663+ }1664+ let decision = match exchange_feedback_report(&mut send, &mut recv, &report).await {1665+ Ok(decision) => decision,1666+ Err(error) if profile == AdaptiveProfile::FloorBreach && minimum_kbps <= 1_000 => {1667+ floor_detach = true;1668+ eprintln!("[probe] floor-breach feedback stream detached as expected: {error}");1669+ break;1670+ }1671+ Err(error) => return Err(error).context("running control-v2 adaptive profile"),1672+ };1673+ minimum_kbps = minimum_kbps.min(decision.target_kbps);1674+ maximum_kbps = maximum_kbps.max(decision.target_kbps);1675+ final_kbps = decision.target_kbps;1676+ final_reason_bits = decision.reasons.bits();1677+ saw_stale |= decision1678+ .reasons1679+ .contains(AdaptiveBitrateReasonFlagsV1::FEEDBACK_STALE);1680+ tokio::time::sleep(Duration::from_millis(800)).await;1681+ }1682+ ensure!(1683+ minimum_kbps != u32::MAX,1684+ "adaptive profile received no decisions"1685+ );1686+ let survivor_recovery = profile == AdaptiveProfile::Clean1687+ && minimum_kbps < maximum_kbps1688+ && final_kbps > minimum_kbps;1689+ match profile {1690+ AdaptiveProfile::Clean => ensure!(1691+ survivor_recovery,1692+ "clean survivor did not observe aggregate downshift and recovery: min={minimum_kbps}, max={maximum_kbps}, final={final_kbps}, reasons={final_reason_bits}"1693+ ),1694+ AdaptiveProfile::FloorBreach => ensure!(1695+ floor_detach && minimum_kbps <= 1_000,1696+ "floor-breach viewer was not detached after bounded recovery"1697+ ),1698+ AdaptiveProfile::Stale => ensure!(saw_stale, "stale profile was not classified stale"),1699+ }1700+ if profile != AdaptiveProfile::FloorBreach {1701+ send.finish()1702+ .context("finishing control-v2 adaptive feedback profile")?;1703+ connection.close(0_u32.into(), b"adaptive profile complete");1704+ }1705+ Ok(AdaptiveProfileEvidence {1706+ minimum_kbps,1707+ final_kbps,1708+ survivor_recovery,1709+ floor_detach,1710+ })1711+}1712+14961713 async fn next_resolution_stall_moq_frame(receiver: &mut MoqProbeReceiver) -> Result<AcceptedMedia> {14971714 loop {14981715 match receiver.next().await? {@@ -1835,6 +2052,7 @@ async fn main() -> Result<()> {
18352052 args.spectator,18362053 args.disconnect_with_focus,18372054 args.focus_handoffs,2055+ args.adaptive_profile,18382056 )18392057 .await;18402058 }@@ -1170,7 +1170,9 @@ async fn serve_command(args: ServeArgs) -> Result<()> {
11701170 MediaFeedbackHandler {11711171 config: config.clone(),11721172 sessions: Arc::clone(&sessions),1173+ generations: Arc::clone(&media_generations),11731174 authorization: authorization.clone(),1175+ input_operations: Arc::clone(&input_operations),11741176 },11751177 )11761178 .accept(@@ -1,6 +1,9 @@
1+use std::collections::HashMap;2+use std::sync::Mutex;3+14 use super::session::{2- AdaptiveEncoderProposal, FeedbackLease, MediaV3TelemetrySnapshot, ResolutionEncoderProposal,3- SessionRegistry,5+ AdaptiveEncoderProposal, FeedbackLease, FeedbackScope, MediaV3TelemetrySnapshot,6+ ResolutionEncoderProposal, SessionRegistry,47 };58 use super::*;69 @@ -25,6 +28,12 @@ const FEEDBACK_WRITE_TIMEOUT: Duration = Duration::from_secs(1);
2528 // consumer observes a stale 5-second window instead of a deceptively fresh2629 // latest report.2730 const MEDIA_FEEDBACK_INTERVAL_MAX_MS: u64 = 5_000;31+const AGGREGATE_TICK_INTERVAL: Duration = Duration::from_millis(250);32+const FLOOR_BREACH_WINDOWS: u8 = 3;33+const ISOLATED_RECOVERY_CLEAN_WINDOWS: u8 = 3;34+const ISOLATED_RECOVERY_PRESSURE_WINDOWS: u8 = 3;35+const ISOLATED_RECOVERY_MAX_ATTEMPTS: u8 = 2;36+const ISOLATED_RECOVERY_BACKOFF: Duration = Duration::from_secs(2);2837 2938 3039 enum FeedbackSeverity {@@ -440,6 +449,546 @@ fn cumulative_feedback_delta(total: u64, baseline: u64) -> u32 {
440449 .expect("feedback delta is clamped to u32")441450 }442451 452+453+struct GenerationContributorKey {454+ remote: EndpointId,455+ session_id: u64,456+}457+458+459+enum ContributorRecoveryState {460+ Active {461+ floor_breach_windows: u8,462+ },463+ Recovering {464+ attempt: u8,465+ retry_at: Instant,466+ clean_windows: u8,467+ pressure_windows: u8,468+ },469+ Detaching,470+}471+472+impl Default for ContributorRecoveryState {473+ fn default() -> Self {474+ Self::Active {475+ floor_breach_windows: 0,476+ }477+ }478+}479+480+impl ContributorRecoveryState {481+ fn contributes(self) -> bool {482+ matches!(self, Self::Active { .. })483+ }484+485+ fn observe(486+ &mut self,487+ evaluation: &AdaptiveBitrateEvaluation,488+ now: Instant,489+ ) -> RecoveryMutation {490+ let pressured = evaluation.fresh491+ && matches!(492+ evaluation.pressure_severity,493+ FeedbackSeverity::Moderate | FeedbackSeverity::Severe494+ );495+ let severe = evaluation.fresh && evaluation.pressure_severity == FeedbackSeverity::Severe;496+ let clean = evaluation.fresh497+ && evaluation.pressure_severity == FeedbackSeverity::Clean498+ && evaluation.clean_recovery_evidence;499+ match *self {500+ Self::Active {501+ ref mut floor_breach_windows,502+ } => {503+ if severe && evaluation.decision.target_kbps == evaluation.decision.floor_kbps {504+ *floor_breach_windows = floor_breach_windows.saturating_add(1);505+ } else {506+ *floor_breach_windows = 0;507+ }508+ if *floor_breach_windows < FLOOR_BREACH_WINDOWS {509+ return RecoveryMutation::None;510+ }511+ *self = Self::Recovering {512+ attempt: 1,513+ retry_at: now + ISOLATED_RECOVERY_BACKOFF,514+ clean_windows: 0,515+ pressure_windows: 0,516+ };517+ RecoveryMutation::RequestRecovery518+ }519+ Self::Recovering {520+ attempt,521+ retry_at,522+ ref mut clean_windows,523+ ref mut pressure_windows,524+ } => {525+ if now < retry_at {526+ return RecoveryMutation::None;527+ }528+ if clean {529+ *clean_windows = clean_windows.saturating_add(1);530+ *pressure_windows = 0;531+ if *clean_windows >= ISOLATED_RECOVERY_CLEAN_WINDOWS {532+ *self = Self::default();533+ return RecoveryMutation::Recovered;534+ }535+ } else if pressured {536+ *pressure_windows = pressure_windows.saturating_add(1);537+ *clean_windows = 0;538+ if *pressure_windows >= ISOLATED_RECOVERY_PRESSURE_WINDOWS {539+ if attempt >= ISOLATED_RECOVERY_MAX_ATTEMPTS {540+ *self = Self::Detaching;541+ return RecoveryMutation::Detach;542+ }543+ let next_attempt = attempt.saturating_add(1);544+ *self = Self::Recovering {545+ attempt: next_attempt,546+ retry_at: now547+ + ISOLATED_RECOVERY_BACKOFF.saturating_mul(u32::from(next_attempt)),548+ clean_windows: 0,549+ pressure_windows: 0,550+ };551+ return RecoveryMutation::RequestRecovery;552+ }553+ } else {554+ *clean_windows = 0;555+ *pressure_windows = 0;556+ }557+ RecoveryMutation::None558+ }559+ Self::Detaching => RecoveryMutation::None,560+ }561+ }562+}563+564+565+enum RecoveryMutation {566+ None,567+ RequestRecovery,568+ Recovered,569+ Detach,570+}571+572+struct GenerationContributor {573+ authorization_revision: u64,574+ ingress: CumulativeFeedbackIngress,575+ cursor: FeedbackIngressCursor,576+ controller: ShadowBitrateController,577+ latest_report: Option<MediaFeedbackReportV1>,578+ latest_evaluation: Option<AdaptiveBitrateEvaluation>,579+ last_received_at: Option<Instant>,580+ recovery: ContributorRecoveryState,581+ telemetry: Arc<super::session::MediaV3Telemetry>,582+ decisions: tokio::sync::watch::Sender<Option<AdaptiveBitrateDecisionV1>>,583+ detach: tokio::sync::watch::Sender<bool>,584+}585+586+struct GenerationAggregateState {587+ contributors: HashMap<GenerationContributorKey, GenerationContributor>,588+ next_decision_id: u64,589+ target_kbps: u32,590+}591+592+pub(super) struct GenerationAdaptiveCoordinator {593+ generation_id: u64,594+ ceiling_kbps: u32,595+ state: Mutex<GenerationAggregateState>,596+ changed: tokio::sync::Notify,597+}598+599+impl std::fmt::Debug for GenerationAdaptiveCoordinator {600+ fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {601+ formatter602+ .debug_struct("GenerationAdaptiveCoordinator")603+ .field("generation_id", &self.generation_id)604+ .field("ceiling_kbps", &self.ceiling_kbps)605+ .finish_non_exhaustive()606+ }607+}608+609+pub(super) struct GenerationFeedbackRegistration {610+ coordinator: Arc<GenerationAdaptiveCoordinator>,611+ key: GenerationContributorKey,612+ authorization_revision: u64,613+ pub(super) decisions: tokio::sync::watch::Receiver<Option<AdaptiveBitrateDecisionV1>>,614+ pub(super) detach: tokio::sync::watch::Receiver<bool>,615+}616+617+impl GenerationFeedbackRegistration {618+ pub(super) fn observe(&self, report: MediaFeedbackReportV1) -> Result<()> {619+ report.validate()?;620+ let mut state = self621+ .coordinator622+ .state623+ .lock()624+ .map_err(|_| anyhow::anyhow!("generation adaptive state poisoned"))?;625+ let contributor = state626+ .contributors627+ .get_mut(&self.key)628+ .filter(|contributor| contributor.authorization_revision == self.authorization_revision)629+ .context("feedback contributor no longer matches the active viewer generation")?;630+ contributor.ingress.observe(report);631+ drop(state);632+ self.coordinator.changed.notify_one();633+ Ok(())634+ }635+}636+637+impl Drop for GenerationFeedbackRegistration {638+ fn drop(&mut self) {639+ self.coordinator640+ .unregister(self.key, self.authorization_revision);641+ }642+}643+644+impl GenerationAdaptiveCoordinator {645+ pub(super) fn start(646+ config: HostConfig,647+ generation_id: u64,648+ encoder_control: Option<EncoderControl>,649+ producer_telemetry: Arc<super::session::MediaV3Telemetry>,650+ keyframe_requests: tokio::sync::watch::Sender<Option<MediaControlRequestV3>>,651+ shutdown: tokio::sync::watch::Receiver<bool>,652+ ) -> Result<(Arc<Self>, tokio::task::JoinHandle<Result<()>>)> {653+ let ceiling_kbps = adaptive_bitrate_ceiling_kbps(&config)?;654+ let coordinator = Arc::new(Self {655+ generation_id,656+ ceiling_kbps,657+ state: Mutex::new(GenerationAggregateState {658+ contributors: HashMap::with_capacity(crate::config::MAX_VIEWERS.into()),659+ next_decision_id: 1,660+ target_kbps: ceiling_kbps,661+ }),662+ changed: tokio::sync::Notify::new(),663+ });664+ let task_coordinator = Arc::clone(&coordinator);665+ let task = tokio::spawn(async move {666+ task_coordinator667+ .run(668+ config,669+ encoder_control,670+ producer_telemetry,671+ keyframe_requests,672+ shutdown,673+ )674+ .await675+ });676+ Ok((coordinator, task))677+ }678+679+ pub(super) fn register(680+ self: &Arc<Self>,681+ remote: EndpointId,682+ session_id: u64,683+ authorization_revision: u64,684+ media_generation_id: u64,685+ telemetry: Arc<super::session::MediaV3Telemetry>,686+ ) -> Result<GenerationFeedbackRegistration> {687+ ensure!(688+ media_generation_id == self.generation_id,689+ "feedback media generation does not match the shared generation"690+ );691+ ensure!(692+ authorization_revision != 0,693+ "feedback authorization revision must be non-zero"694+ );695+ let key = GenerationContributorKey { remote, session_id };696+ let (decisions, decision_rx) = tokio::sync::watch::channel(None);697+ let (detach, detach_rx) = tokio::sync::watch::channel(false);698+ let mut state = self699+ .state700+ .lock()701+ .map_err(|_| anyhow::anyhow!("generation adaptive state poisoned"))?;702+ ensure!(703+ state.contributors.len() < usize::from(crate::config::MAX_VIEWERS)704+ || state.contributors.contains_key(&key),705+ "generation feedback contributor capacity is full"706+ );707+ state.contributors.insert(708+ key,709+ GenerationContributor {710+ authorization_revision,711+ ingress: CumulativeFeedbackIngress::default(),712+ cursor: FeedbackIngressCursor::default(),713+ controller: ShadowBitrateController::new(self.ceiling_kbps, telemetry.snapshot()),714+ latest_report: None,715+ latest_evaluation: None,716+ last_received_at: None,717+ recovery: ContributorRecoveryState::default(),718+ telemetry,719+ decisions,720+ detach,721+ },722+ );723+ drop(state);724+ self.changed.notify_one();725+ Ok(GenerationFeedbackRegistration {726+ coordinator: Arc::clone(self),727+ key,728+ authorization_revision,729+ decisions: decision_rx,730+ detach: detach_rx,731+ })732+ }733+734+ fn unregister(&self, key: GenerationContributorKey, authorization_revision: u64) {735+ let Ok(mut state) = self.state.lock() else {736+ return;737+ };738+ if state739+ .contributors740+ .get(&key)741+ .is_some_and(|contributor| contributor.authorization_revision == authorization_revision)742+ {743+ state.contributors.remove(&key);744+ drop(state);745+ self.changed.notify_one();746+ }747+ }748+749+ async fn run(750+ self: Arc<Self>,751+ config: HostConfig,752+ encoder_control: Option<EncoderControl>,753+ producer_telemetry: Arc<super::session::MediaV3Telemetry>,754+ keyframe_requests: tokio::sync::watch::Sender<Option<MediaControlRequestV3>>,755+ mut shutdown: tokio::sync::watch::Receiver<bool>,756+ ) -> Result<()> {757+ let actuation_enabled =758+ adaptive_bitrate_actuation_enabled(&config) && encoder_control.is_some();759+ let mut resolution_controller = if actuation_enabled {760+ encoder_control761+ .as_ref()762+ .map(motion_resolution_policy_for_encoder)763+ .transpose()?764+ .flatten()765+ } else {766+ None767+ };768+ let initial_resolution = resolution_controller769+ .as_ref()770+ .map(MotionResolutionPolicy::target);771+ let mut actuator = GenerationAdaptiveActuator::new(772+ encoder_control,773+ actuation_enabled,774+ self.ceiling_kbps,775+ initial_resolution,776+ );777+ let mut tick = tokio::time::interval(AGGREGATE_TICK_INTERVAL);778+ tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);779+ loop {780+ tokio::select! {781+ biased;782+ changed = shutdown.changed() => {783+ if changed.is_err() || *shutdown.borrow_and_update() {784+ break;785+ }786+ }787+ acknowledgement = actuator.acknowledgements.join_next(), if actuator.pending => {788+ actuator.complete(acknowledgement)?;789+ actuator.start_queued();790+ }791+ _ = self.changed.notified() => {}792+ _ = tick.tick() => {}793+ }794+ if let Some(update) = self.evaluate(795+ Instant::now(),796+ producer_telemetry.snapshot(),797+ actuator.bitrate_is_applied(),798+ resolution_controller.as_mut(),799+ )? {800+ if update.request_recovery {801+ let request_id = update.decision_id;802+ keyframe_requests.send_replace(Some(MediaControlRequestV3::request_keyframe(803+ request_id,804+ update.last_sequence,805+ KeyframeRequestReasonV3::FrontendBackpressure,806+ )));807+ }808+ actuator.submit(update.plan);809+ }810+ }811+ actuator.abort_and_drain().await;812+ Ok(())813+ }814+815+ fn evaluate(816+ &self,817+ now: Instant,818+ producer_telemetry: MediaV3TelemetrySnapshot,819+ bitrate_applied: bool,820+ resolution_controller: Option<&mut MotionResolutionPolicy>,821+ ) -> Result<Option<GenerationAggregateUpdate>> {822+ let mut state = self823+ .state824+ .lock()825+ .map_err(|_| anyhow::anyhow!("generation adaptive state poisoned"))?;826+ let mut changed_keys = Vec::new();827+ let mut request_recovery = false;828+ for (key, contributor) in &mut state.contributors {829+ let Some((report, next_cursor)) =830+ contributor.ingress.report_since(contributor.cursor)?831+ else {832+ continue;833+ };834+ contributor.cursor = next_cursor;835+ // Configured GOP supersession and producer-wide send state are not836+ // charged to every viewer as local transport pressure. The shared837+ // values remain available to generation diagnostics and actuation.838+ let telemetry = contributor.telemetry.snapshot();839+ let evaluation = contributor.controller.evaluate(&report, telemetry, now)?;840+ match contributor.recovery.observe(&evaluation, now) {841+ RecoveryMutation::RequestRecovery => request_recovery = true,842+ RecoveryMutation::Detach => {843+ contributor.detach.send_replace(true);844+ }845+ RecoveryMutation::None | RecoveryMutation::Recovered => {}846+ }847+ contributor.latest_report = Some(report);848+ contributor.latest_evaluation = Some(evaluation);849+ contributor.last_received_at = Some(now);850+ changed_keys.push(*key);851+ }852+853+ let active = state854+ .contributors855+ .values()856+ .filter(|contributor| {857+ contributor.recovery.contributes()858+ && contributor.last_received_at.is_some_and(|received| {859+ now.saturating_duration_since(received) <= FEEDBACK_STALE_ARRIVAL_GAP860+ })861+ && contributor862+ .latest_evaluation863+ .is_some_and(|evaluation| evaluation.fresh)864+ })865+ .filter_map(|contributor| contributor.latest_evaluation.zip(contributor.latest_report))866+ .collect::<Vec<_>>();867+ if changed_keys.is_empty() && active.is_empty() {868+ return Ok(None);869+ }870+ let (target_kbps, all_clean) =871+ aggregate_target_and_clean(&active, state.target_kbps, self.ceiling_kbps);872+ let previous_target = state.target_kbps;873+ if changed_keys.is_empty() && target_kbps == previous_target {874+ return Ok(None);875+ }876+ state.target_kbps = target_kbps;877+ let global_state = match target_kbps.cmp(&previous_target) {878+ std::cmp::Ordering::Less => AdaptiveBitrateStateV1::Decrease,879+ std::cmp::Ordering::Greater => AdaptiveBitrateStateV1::Increase,880+ std::cmp::Ordering::Equal => AdaptiveBitrateStateV1::Hold,881+ };882+ let worst = active883+ .iter()884+ .max_by_key(|(evaluation, _)| feedback_severity_rank(evaluation.pressure_severity));885+ let (aggregate_evaluation, aggregate_report, aggregate_reasons) = match worst {886+ Some((evaluation, report)) => {887+ let mut aggregate = *evaluation;888+ aggregate.clean_recovery_evidence = all_clean;889+ (Some(aggregate), Some(*report), evaluation.decision.reasons)890+ }891+ None => (None, None, AdaptiveBitrateReasonFlagsV1::FEEDBACK_STALE),892+ };893+ let resolution_decision = resolution_controller.and_then(|policy| {894+ aggregate_evaluation895+ .zip(aggregate_report)896+ .map(|(evaluation, report)| policy.observe_window(&report, &evaluation, now))897+ });898+ let plan_decision_id = state.next_decision_id;899+ let plan_report_id = aggregate_report.map_or(1, |report| report.report_id);900+ let plan = AdaptiveCommitPlan {901+ decision: AdaptiveBitrateDecisionV1 {902+ decision_id: plan_decision_id,903+ report_id: plan_report_id,904+ target_kbps,905+ floor_kbps: ADAPTIVE_BITRATE_FLOOR_KBPS,906+ ceiling_kbps: self.ceiling_kbps,907+ state: global_state,908+ reasons: aggregate_reasons,909+ applied: bitrate_applied,910+ },911+ telemetry: producer_telemetry,912+ resolution_decision,913+ force_keyframe: request_recovery914+ || aggregate_report915+ .is_some_and(|report| report.flags.contains(MediaFeedbackFlags::RESYNC_ACTIVE)),916+ };917+918+ for key in changed_keys {919+ let Some((evaluation, decisions)) =920+ state.contributors.get(&key).and_then(|contributor| {921+ contributor922+ .latest_evaluation923+ .map(|evaluation| (evaluation, contributor.decisions.clone()))924+ })925+ else {926+ continue;927+ };928+ let decision_id = state.next_decision_id;929+ state.next_decision_id = state930+ .next_decision_id931+ .checked_add(1)932+ .context("generation adaptive decision ID exhausted")?;933+ let decision = AdaptiveBitrateDecisionV1 {934+ decision_id,935+ report_id: evaluation.decision.report_id,936+ target_kbps,937+ floor_kbps: ADAPTIVE_BITRATE_FLOOR_KBPS,938+ ceiling_kbps: self.ceiling_kbps,939+ state: global_state,940+ reasons: evaluation.decision.reasons,941+ applied: bitrate_applied,942+ };943+ decision.validate()?;944+ decisions.send_replace(Some(decision));945+ }946+ plan.decision.validate()?;947+ Ok(Some(GenerationAggregateUpdate {948+ plan,949+ request_recovery,950+ decision_id: plan_decision_id,951+ last_sequence: aggregate_report.and_then(|report| report.last_sequence),952+ }))953+ }954+}955+956+fn feedback_severity_rank(severity: FeedbackSeverity) -> u8 {957+ match severity {958+ FeedbackSeverity::Clean => 0,959+ FeedbackSeverity::InsufficientEvidence => 1,960+ FeedbackSeverity::Moderate => 2,961+ FeedbackSeverity::Severe => 3,962+ FeedbackSeverity::Stale => 4,963+ }964+}965+966+fn aggregate_target_and_clean(967+ active: &[(AdaptiveBitrateEvaluation, MediaFeedbackReportV1)],968+ current_target_kbps: u32,969+ ceiling_kbps: u32,970+) -> (u32, bool) {971+ let target = active972+ .iter()973+ .map(|(evaluation, _)| evaluation.decision.target_kbps)974+ .min()975+ .unwrap_or(current_target_kbps)976+ .clamp(ADAPTIVE_BITRATE_FLOOR_KBPS, ceiling_kbps);977+ let all_clean = !active.is_empty()978+ && active.iter().all(|(evaluation, _)| {979+ evaluation.pressure_severity == FeedbackSeverity::Clean980+ && evaluation.clean_recovery_evidence981+ });982+ (target, all_clean)983+}984+985+struct GenerationAggregateUpdate {986+ plan: AdaptiveCommitPlan,987+ request_recovery: bool,988+ decision_id: u64,989+ last_sequence: Option<u64>,990+}991+443992 444993 struct ShadowBitrateController {445994 floor_kbps: u32,@@ -681,7 +1230,7 @@ impl ShadowBitrateController {
6811230 FeedbackSeverity::Severe,6821231 AdaptiveBitrateReasonFlagsV1::RTT_INFLATION,6831232 );684- } else if rtt >= baseline.saturating_mul(3) / 2 {1233+ } else if rtt >= (baseline.saturating_mul(3) / 2).max(40_000) {6851234 add_feedback_signal(6861235 &mut severity,6871236 &mut reasons,@@ -897,6 +1446,200 @@ struct AdaptiveCommitCoordinator {
8971446 acknowledgements: tokio::task::JoinSet<AdaptiveCommitAcknowledgement>,8981447 }8991448 1449+1450+struct GenerationAdaptiveAcknowledgement {1451+ bitrate_kbps: Option<u32>,1452+ resolution: Option<VideoDimensions>,1453+}1454+1455+struct GenerationAdaptiveActuator {1456+ control: Option<EncoderControl>,1457+ actuation_enabled: bool,1458+ desired_bitrate_kbps: u32,1459+ requested_bitrate_kbps: u32,1460+ committed_bitrate_kbps: u32,1461+ desired_resolution: Option<VideoDimensions>,1462+ requested_resolution: Option<VideoDimensions>,1463+ committed_resolution: Option<VideoDimensions>,1464+ pending: bool,1465+ queued: Option<AdaptiveCommitPlan>,1466+ acknowledgements: tokio::task::JoinSet<Result<GenerationAdaptiveAcknowledgement>>,1467+}1468+1469+impl GenerationAdaptiveActuator {1470+ fn new(1471+ control: Option<EncoderControl>,1472+ actuation_enabled: bool,1473+ bitrate_kbps: u32,1474+ resolution: Option<VideoDimensions>,1475+ ) -> Self {1476+ Self {1477+ control,1478+ actuation_enabled,1479+ desired_bitrate_kbps: bitrate_kbps,1480+ requested_bitrate_kbps: bitrate_kbps,1481+ committed_bitrate_kbps: bitrate_kbps,1482+ desired_resolution: resolution,1483+ requested_resolution: resolution,1484+ committed_resolution: resolution,1485+ pending: false,1486+ queued: None,1487+ acknowledgements: tokio::task::JoinSet::new(),1488+ }1489+ }1490+1491+ fn bitrate_is_applied(&self) -> bool {1492+ self.actuation_enabled1493+ && !self.pending1494+ && self.desired_bitrate_kbps == self.committed_bitrate_kbps1495+ }1496+1497+ fn submit(&mut self, mut plan: AdaptiveCommitPlan) {1498+ self.desired_bitrate_kbps = plan.decision.target_kbps;1499+ if let Some(resolution) = plan.resolution_decision {1500+ self.desired_resolution = Some(resolution.target);1501+ }1502+ if self.pending {1503+ if let Some(queued) = self.queued {1504+ plan.force_keyframe |= queued.force_keyframe;1505+ }1506+ self.queued = Some(plan);1507+ return;1508+ }1509+ self.start(plan);1510+ }1511+1512+ fn start_queued(&mut self) {1513+ if let Some(plan) = self.queued.take() {1514+ self.start(plan);1515+ }1516+ }1517+1518+ fn start(&mut self, plan: AdaptiveCommitPlan) {1519+ if !self.actuation_enabled {1520+ return;1521+ }1522+ let Some(control) = self.control.clone() else {1523+ return;1524+ };1525+ let bitrate_target = plan.decision.target_kbps;1526+ let bitrate_revision = if bitrate_target != self.requested_bitrate_kbps1527+ || bitrate_target != self.committed_bitrate_kbps1528+ {1529+ match control.request_bitrate_kbps(bitrate_target) {1530+ Ok(revision) => {1531+ self.requested_bitrate_kbps = bitrate_target;1532+ Some(revision)1533+ }1534+ Err(error) => {1535+ warn!(%error, bitrate_target, "generation adaptive bitrate request failed");1536+ None1537+ }1538+ }1539+ } else {1540+ None1541+ };1542+ let resolution_target = plan.resolution_decision.map(|decision| decision.target);1543+ let resolution_revision = if resolution_target.is_some()1544+ && (resolution_target != self.requested_resolution1545+ || resolution_target != self.committed_resolution)1546+ {1547+ let target = resolution_target.expect("resolution target was checked");1548+ match control.request_resolution(target.width, target.height) {1549+ Ok(revision) => {1550+ self.requested_resolution = Some(target);1551+ Some((revision, target))1552+ }1553+ Err(error) => {1554+ warn!(%error, width = target.width, height = target.height, "generation adaptive resolution request failed");1555+ None1556+ }1557+ }1558+ } else {1559+ None1560+ };1561+ if bitrate_revision.is_none() && resolution_revision.is_none() {1562+ return;1563+ }1564+ self.pending = true;1565+ self.acknowledgements.spawn(async move {1566+ let bitrate_kbps = if let Some(revision) = bitrate_revision {1567+ let status = control1568+ .wait_for_bitrate_applied(revision, ENCODER_CONTROL_COMMIT_TIMEOUT)1569+ .await?;1570+ ensure!(1571+ status.applied_bitrate_revision == Some(revision)1572+ && status.applied_bitrate_kbps == Some(bitrate_target),1573+ "generation bitrate acknowledgement did not match revision {revision} at {bitrate_target} kbps"1574+ );1575+ Some(bitrate_target)1576+ } else {1577+ None1578+ };1579+ let resolution = if let Some((revision, target)) = resolution_revision {1580+ let status = control1581+ .wait_for_resolution_applied(1582+ revision,1583+ target.width,1584+ target.height,1585+ ENCODER_CONTROL_COMMIT_TIMEOUT,1586+ )1587+ .await?;1588+ ensure!(1589+ status.applied_resolution_revision == Some(revision)1590+ && status.applied_width == Some(target.width)1591+ && status.applied_height == Some(target.height),1592+ "generation resolution acknowledgement did not match revision {revision} at {}x{}",1593+ target.width,1594+ target.height1595+ );1596+ Some(target)1597+ } else {1598+ None1599+ };1600+ Ok(GenerationAdaptiveAcknowledgement {1601+ bitrate_kbps,1602+ resolution,1603+ })1604+ });1605+ }1606+1607+ fn complete(1608+ &mut self,1609+ result: Option<Result<Result<GenerationAdaptiveAcknowledgement>, tokio::task::JoinError>>,1610+ ) -> Result<()> {1611+ ensure!(1612+ self.pending,1613+ "generation adaptive acknowledgement was not pending"1614+ );1615+ self.pending = false;1616+ match result1617+ .context("generation adaptive task ended without a result")?1618+ .context("generation adaptive task failed")?1619+ {1620+ Ok(acknowledgement) => {1621+ if let Some(bitrate_kbps) = acknowledgement.bitrate_kbps {1622+ self.committed_bitrate_kbps = bitrate_kbps;1623+ }1624+ if let Some(resolution) = acknowledgement.resolution {1625+ self.committed_resolution = Some(resolution);1626+ }1627+ }1628+ Err(error) => {1629+ warn!(%error, "generation adaptive commit was not applied");1630+ }1631+ }1632+ Ok(())1633+ }1634+1635+ async fn abort_and_drain(&mut self) {1636+ self.pending = false;1637+ self.queued = None;1638+ self.acknowledgements.abort_all();1639+ while self.acknowledgements.join_next().await.is_some() {}1640+ }1641+}1642+9001643 impl AdaptiveCommitCoordinator {9011644 fn new(9021645 sessions: Arc<SessionRegistry>,@@ -1106,7 +1849,9 @@ pub(super) async fn serve_media_feedback(
11061849 connection: Connection,11071850 config: &HostConfig,11081851 sessions: &Arc<SessionRegistry>,1852+ generations: &Arc<MediaGenerationManager>,11091853 authorization: &AuthorizationPolicy,1854+ input_operations: &Arc<InputOperations>,11101855 ) -> Result<()> {11111856 let remote = connection.remote_id();11121857 let handshake_permit = sessions@@ -1155,6 +1900,40 @@ pub(super) async fn serve_media_feedback(
11551900 ),11561901 )11571902 .await?;1903+ if lease.scope == FeedbackScope::SharedGeneration {1904+ let adaptive = generations.adaptive(lease.media_generation_id).await?;1905+ let registration = adaptive.register(1906+ remote,1907+ lease.session_id,1908+ lease.authorization_revision,1909+ lease.media_generation_id,1910+ Arc::clone(&lease.telemetry),1911+ )?;1912+ info!(1913+ %remote,1914+ session_id = lease.session_id,1915+ media_generation_id = lease.media_generation_id,1916+ authorization_revision = lease.authorization_revision,1917+ ceiling_kbps,1918+ mode = if encoder_actuation_available { "aggregate-active" } else { "aggregate-shadow" },1919+ "media feedback contributor attached to shared generation"1920+ );1921+ let result = run_generation_feedback_session(1922+ &connection,1923+ &mut send,1924+ recv,1925+ remote,1926+ sessions,1927+ input_operations,1928+ &lease,1929+ registration,1930+ )1931+ .await;1932+ let _ = send.finish();1933+ drop(lease);1934+ info!(%remote, "shared-generation media feedback contributor released");1935+ return result;1936+ }11581937 info!(11591938 %remote,11601939 session_id = lease.session_id,@@ -1190,6 +1969,99 @@ pub(super) async fn serve_media_feedback(
11901969 result11911970 }11921971 1972+1973+async fn run_generation_feedback_session<W>(1974+ connection: &Connection,1975+ send: &mut W,1976+ recv: iroh::endpoint::RecvStream,1977+ remote: EndpointId,1978+ sessions: &Arc<SessionRegistry>,1979+ input_operations: &Arc<InputOperations>,1980+ lease: &FeedbackLease,1981+ mut registration: GenerationFeedbackRegistration,1982+) -> Result<()>1983+where1984+ W: tokio::io::AsyncWrite + Unpin,1985+{1986+ let (feedback_sender, mut feedback_receiver) =1987+ tokio::sync::watch::channel(CumulativeFeedbackIngress::default());1988+ let mut reader_task = tokio::spawn(forward_media_feedback_reports(recv, feedback_sender));1989+ let mut feedback_cursor = FeedbackIngressCursor::default();1990+ let result = loop {1991+ let session_changed = sessions.session_changed.notified();1992+ tokio::pin!(session_changed);1993+ session_changed.as_mut().enable();1994+ if !sessions.is_active(remote, lease.session_id) {1995+ break Ok(());1996+ }1997+ tokio::select! {1998+ reader_result = &mut reader_task => {1999+ break reader_result2000+ .context("shared-generation feedback reader task failed")?2001+ .context("reading shared-generation feedback reports");2002+ }2003+ changed = feedback_receiver.changed() => {2004+ if changed.is_err() {2005+ break Ok(());2006+ }2007+ let ingress = *feedback_receiver.borrow_and_update();2008+ let Some((report, next_cursor)) = ingress.report_since(feedback_cursor)? else {2009+ continue;2010+ };2011+ feedback_cursor = next_cursor;2012+ registration.observe(report)?;2013+ }2014+ changed = registration.decisions.changed() => {2015+ changed.context("shared-generation adaptive decision publisher stopped")?;2016+ let Some(decision) = *registration.decisions.borrow_and_update() else {2017+ continue;2018+ };2019+ tokio::time::timeout(2020+ FEEDBACK_WRITE_TIMEOUT,2021+ write_adaptive_bitrate_decision_v1(send, &decision),2022+ )2023+ .await2024+ .context("timed out writing shared-generation adaptive decision")??;2025+ }2026+ changed = registration.detach.changed() => {2027+ changed.context("shared-generation recovery state publisher stopped")?;2028+ if !*registration.detach.borrow_and_update() {2029+ continue;2030+ }2031+ if let Some(transition) = sessions.invalidate_v2_focus(2032+ remote,2033+ lease.session_id,2034+ FocusTransitionReasonV2::Disconnected,2035+ )? {2036+ input_operations.neutralize_focus_transition(2037+ sessions,2038+ transition.transition_id,2039+ FocusTransitionReasonV2::Disconnected,2040+ )?;2041+ }2042+ sessions.disconnect_v2_viewer(remote, lease.session_id)?;2043+ info!(2044+ %remote,2045+ session_id = lease.session_id,2046+ media_generation_id = lease.media_generation_id,2047+ "detached persistently below-floor viewer after bounded isolated recovery"2048+ );2049+ break Ok(());2050+ }2051+ _ = &mut session_changed => {}2052+ reason = connection.closed() => {2053+ debug!(%remote, ?reason, "shared-generation feedback connection closed");2054+ break Ok(());2055+ }2056+ }2057+ };2058+ if !reader_task.is_finished() {2059+ reader_task.abort();2060+ let _ = reader_task.await;2061+ }2062+ result2063+}2064+11932065 11942066 async fn run_media_feedback_loop<W, C>(11952067 send: &mut W,@@ -1756,6 +2628,151 @@ mod tests {
17562628 }17572629 }17582630 2631+ fn aggregate_evaluation(2632+ report_id: u64,2633+ target_kbps: u32,2634+ severity: FeedbackSeverity,2635+ clean: bool,2636+ ) -> AdaptiveBitrateEvaluation {2637+ AdaptiveBitrateEvaluation {2638+ decision: AdaptiveBitrateDecisionV1 {2639+ decision_id: report_id,2640+ report_id,2641+ target_kbps,2642+ floor_kbps: ADAPTIVE_BITRATE_FLOOR_KBPS,2643+ ceiling_kbps: 12_000,2644+ state: AdaptiveBitrateStateV1::Hold,2645+ reasons: if severity == FeedbackSeverity::Clean {2646+ AdaptiveBitrateReasonFlagsV1::CLEAN_RECOVERY2647+ } else {2648+ AdaptiveBitrateReasonFlagsV1::RECEIVER_QUEUE2649+ },2650+ applied: false,2651+ },2652+ pressure_severity: severity,2653+ clean_recovery_evidence: clean,2654+ fresh: severity != FeedbackSeverity::Stale,2655+ }2656+ }2657+2658+ 2659+ fn aggregate_uses_worst_fresh_target_and_requires_every_contributor_clean() {2660+ let clean = aggregate_evaluation(1, 10_000, FeedbackSeverity::Clean, true);2661+ let pressured = aggregate_evaluation(2, 6_000, FeedbackSeverity::Moderate, false);2662+ let active = vec![(clean, clean_feedback(1)), (pressured, clean_feedback(2))];2663+ assert_eq!(2664+ aggregate_target_and_clean(&active, 12_000, 12_000),2665+ (6_000, false)2666+ );2667+2668+ let recovered = vec![2669+ (clean, clean_feedback(3)),2670+ (2671+ aggregate_evaluation(4, 8_000, FeedbackSeverity::Clean, true),2672+ clean_feedback(4),2673+ ),2674+ ];2675+ assert_eq!(2676+ aggregate_target_and_clean(&recovered, 6_000, 12_000),2677+ (8_000, true)2678+ );2679+ }2680+2681+ 2682+ fn sub_millisecond_rtt_jitter_does_not_become_permanent_moderate_pressure() {2683+ let now = Instant::now();2684+ let baseline = MediaV3TelemetrySnapshot {2685+ selected_path_rtt_micros: 500,2686+ ..Default::default()2687+ };2688+ let mut controller = ShadowBitrateController::new(12_000, baseline);2689+ controller2690+ .evaluate(&clean_feedback(1), baseline, now)2691+ .unwrap();2692+ let jitter = MediaV3TelemetrySnapshot {2693+ selected_path_rtt_micros: 1_000,2694+ ..Default::default()2695+ };2696+ let evaluation = controller2697+ .evaluate(&clean_feedback(2), jitter, now + Duration::from_secs(1))2698+ .unwrap();2699+ assert!(2700+ !evaluation2701+ .decision2702+ .reasons2703+ .contains(AdaptiveBitrateReasonFlagsV1::RTT_INFLATION)2704+ );2705+ assert_eq!(evaluation.pressure_severity, FeedbackSeverity::Clean);2706+ }2707+2708+ 2709+ fn persistent_floor_breach_retries_then_detaches_without_holding_survivors_down() {2710+ let now = Instant::now();2711+ let severe = aggregate_evaluation(2712+ 1,2713+ ADAPTIVE_BITRATE_FLOOR_KBPS,2714+ FeedbackSeverity::Severe,2715+ false,2716+ );2717+ let mut recovery = ContributorRecoveryState::default();2718+ assert_eq!(recovery.observe(&severe, now), RecoveryMutation::None);2719+ assert_eq!(recovery.observe(&severe, now), RecoveryMutation::None);2720+ assert_eq!(2721+ recovery.observe(&severe, now),2722+ RecoveryMutation::RequestRecovery2723+ );2724+ assert!(!recovery.contributes());2725+2726+ let first_retry = now + ISOLATED_RECOVERY_BACKOFF;2727+ assert_eq!(2728+ recovery.observe(&severe, first_retry),2729+ RecoveryMutation::None2730+ );2731+ assert_eq!(2732+ recovery.observe(&severe, first_retry),2733+ RecoveryMutation::None2734+ );2735+ assert_eq!(2736+ recovery.observe(&severe, first_retry),2737+ RecoveryMutation::RequestRecovery2738+ );2739+ let second_retry = first_retry + ISOLATED_RECOVERY_BACKOFF.saturating_mul(2);2740+ assert_eq!(2741+ recovery.observe(&severe, second_retry),2742+ RecoveryMutation::None2743+ );2744+ assert_eq!(2745+ recovery.observe(&severe, second_retry),2746+ RecoveryMutation::None2747+ );2748+ assert_eq!(2749+ recovery.observe(&severe, second_retry),2750+ RecoveryMutation::Detach2751+ );2752+ assert_eq!(recovery, ContributorRecoveryState::Detaching);2753+ }2754+2755+ 2756+ fn isolated_viewer_rejoins_only_after_bounded_clean_recovery() {2757+ let now = Instant::now();2758+ let severe = aggregate_evaluation(2759+ 1,2760+ ADAPTIVE_BITRATE_FLOOR_KBPS,2761+ FeedbackSeverity::Severe,2762+ false,2763+ );2764+ let clean = aggregate_evaluation(2, 2_000, FeedbackSeverity::Clean, true);2765+ let mut recovery = ContributorRecoveryState::default();2766+ for _ in 0..FLOOR_BREACH_WINDOWS {2767+ let _ = recovery.observe(&severe, now);2768+ }2769+ let retry = now + ISOLATED_RECOVERY_BACKOFF;2770+ assert_eq!(recovery.observe(&clean, retry), RecoveryMutation::None);2771+ assert_eq!(recovery.observe(&clean, retry), RecoveryMutation::None);2772+ assert_eq!(recovery.observe(&clean, retry), RecoveryMutation::Recovered);2773+ assert!(recovery.contributes());2774+ }2775+17592776 fn controller_decide(17602777 controller: &mut ShadowBitrateController,17612778 report: &MediaFeedbackReportV1,@@ -12,6 +12,7 @@ use sigil_protocol::{
1212 use tokio::sync::{Notify, watch};1313 use tracing::{debug, info, warn};1414 15+use super::adaptive::GenerationAdaptiveCoordinator;1516 use super::moq::{run_generation_audio_publisher, run_generation_video_publisher};1617 use super::session::{MediaV3Telemetry, SourceTaskGuard};1718 use crate::audio::spawn_pipewire_audio;@@ -158,6 +159,22 @@ impl MediaGenerationManager {
158159 self.state_changed.notify_waiters();159160 }160161 162+ pub(super) async fn adaptive(163+ &self,164+ generation_id: u64,165+ ) -> Result<Arc<GenerationAdaptiveCoordinator>> {166+ let state = self.state.lock().await;167+ state168+ .active169+ .as_ref()170+ .filter(|active| {171+ state.lifecycle == MediaGenerationLifecycle::Active172+ && active.shared.generation_id == generation_id173+ })174+ .map(|active| Arc::clone(&active.shared.adaptive))175+ .context("feedback does not match the active media generation")176+ }177+161178 162179 async fn lifecycle(&self) -> MediaGenerationLifecycle {163180 self.state.lock().await.lifecycle@@ -175,6 +192,7 @@ pub(super) struct GenerationShared {
175192 pub(super) telemetry: Arc<MediaV3Telemetry>,176193 pub(super) audio_enabled: bool,177194 pub(super) keyframe_requests: watch::Sender<Option<sigil_protocol::MediaControlRequestV3>>,195+ pub(super) adaptive: Arc<GenerationAdaptiveCoordinator>,178196 }179197 180198 impl std::fmt::Debug for GenerationShared {@@ -247,6 +265,7 @@ struct ActiveGeneration {
247265 shutdown: watch::Sender<bool>,248266 video_publisher: tokio::task::JoinHandle<Result<()>>,249267 audio_publisher: Option<tokio::task::JoinHandle<Result<()>>>,268+ adaptive: tokio::task::JoinHandle<Result<()>>,250269 video_source: SourceTaskGuard,251270 audio_source: Option<SourceTaskGuard>,252271 _frame_receiver: watch::Receiver<Option<EncodedFrame>>,@@ -328,6 +347,14 @@ impl ActiveGeneration {
328347 let telemetry = Arc::new(MediaV3Telemetry::default());329348 let (keyframe_requests, keyframes) = watch::channel(None);330349 let (shutdown, shutdown_rx) = watch::channel(false);350+ let (adaptive, adaptive_task) = GenerationAdaptiveCoordinator::start(351+ config.clone(),352+ generation_id,353+ encoder_control.clone(),354+ Arc::clone(&telemetry),355+ keyframe_requests.clone(),356+ shutdown_rx.clone(),357+ )?;331358 let video_publisher = tokio::spawn(run_generation_video_publisher(332359 config.clone(),333360 current_gop,@@ -375,6 +402,7 @@ impl ActiveGeneration {
375402 telemetry,376403 audio_enabled,377404 keyframe_requests,405+ adaptive,378406 });379407 info!(380408 generation_id,@@ -390,6 +418,7 @@ impl ActiveGeneration {
390418 shutdown,391419 video_publisher,392420 audio_publisher,421+ adaptive: adaptive_task,393422 video_source,394423 audio_source,395424 _frame_receiver: frames,@@ -404,6 +433,7 @@ impl ActiveGeneration {
404433 *broadcast = None;405434 }406435 self.shutdown.send_replace(true);436+ wait_or_abort_task(&mut self.adaptive, "adaptive coordinator").await;407437 wait_or_abort_task(&mut self.video_publisher, "video publisher").await;408438 if let Some(mut audio_publisher) = self.audio_publisher.take() {409439 wait_or_abort_task(&mut audio_publisher, "audio publisher").await;@@ -78,7 +78,9 @@ pub struct ControlV2Handler {
7878 pub struct MediaFeedbackHandler {7979 pub config: HostConfig,8080 pub sessions: Arc<SessionRegistry>,81+ pub generations: Arc<MediaGenerationManager>,8182 pub authorization: AuthorizationPolicy,83+ pub input_operations: Arc<InputOperations>,8284 }8385 8486 /// Upstream MoQ admission guarded by an already-authenticated control lease.@@ -134,7 +136,9 @@ impl ProtocolHandler for MediaFeedbackHandler {
134136 connection,135137 &self.config,136138 &self.sessions,139+ &self.generations,137140 &self.authorization,141+ &self.input_operations,138142 )139143 .await140144 {@@ -602,7 +602,7 @@ impl SessionRegistry {
602602 publish_v2_snapshots(&state)?;603603 let snapshot = v2_snapshot_for(&state, remote)?;604604 let replaced_snapshots = replaced.map(|viewer| viewer.snapshots);605- Ok(V2SessionLease {605+ let lease = V2SessionLease {606606 registry: Arc::clone(self),607607 remote,608608 session_id,@@ -613,7 +613,10 @@ impl SessionRegistry {
613613 authorization_committed_revision: authorized.committed_revision,614614 replaced_snapshots,615615 replacement_transition,616- })616+ };617+ drop(state);618+ self.session_changed.notify_waiters();619+ Ok(lease)617620 }618621 619622 pub(super) fn bind_v2_generation(@@ -622,7 +625,7 @@ impl SessionRegistry {
622625 session_id: u64,623626 generation_id: u64,624627 session_clock: SessionClock,625- telemetry: Arc<MediaV3Telemetry>,628+ _telemetry: Arc<MediaV3Telemetry>,626629 encoder_control: Option<EncoderControl>,627630 ) -> Result<SessionSnapshotV2> {628631 let mut state = self.v2_state.lock().expect("v2 session state poisoned");@@ -647,7 +650,8 @@ impl SessionRegistry {
647650 viewer.session.session_clock = session_clock;648651 viewer.session.media_generation_id = generation_id;649652 viewer.session.media_broadcast_name = descriptor.broadcast_name;650- viewer.session.media_v3_telemetry = telemetry;653+ // Keep transport telemetry viewer-scoped. Producer telemetry belongs654+ // to the shared generation and is merged only by its adaptive owner.651655 viewer.session.encoder_control = encoder_control;652656 publish_v2_snapshots(&state)?;653657 let snapshot = v2_snapshot_for(&state, remote)?;@@ -1212,6 +1216,8 @@ impl SessionRegistry {
12121216 telemetry: Arc::clone(&session.media_v3_telemetry),12131217 encoder_control: session.encoder_control.clone(),12141218 authorization_revision: session.authorization_revision,1219+ media_generation_id: session.media_generation_id,1220+ scope: FeedbackScope::Legacy,12151221 });12161222 }12171223 drop(active);@@ -1226,6 +1232,11 @@ impl SessionRegistry {
12261232 session.grants.contains(InvitationGrants::VIEW),12271233 "active Portal session lacks feedback view permission"12281234 );1235+ ensure!(1236+ session.authorization_committed_revision1237+ == self.authorization_committed_revision.load(Ordering::SeqCst),1238+ "feedback claim used a stale committed authorization revision"1239+ );12291240 ensure!(12301241 !session.feedback_claimed,12311242 "active client already has a feedback connection"@@ -1238,6 +1249,8 @@ impl SessionRegistry {
12381249 telemetry: Arc::clone(&session.media_v3_telemetry),12391250 encoder_control: session.encoder_control.clone(),12401251 authorization_revision: session.authorization_revision,1252+ media_generation_id: session.media_generation_id,1253+ scope: FeedbackScope::SharedGeneration,12411254 })12421255 }12431256 @@ -1515,6 +1528,41 @@ impl SessionRegistry {
15151528 })15161529 }15171530 1531+ pub(super) fn disconnect_v2_viewer(&self, remote: EndpointId, session_id: u64) -> Result<bool> {1532+ let mut state = self.v2_state.lock().expect("v2 session state poisoned");1533+ let Some(viewer) = state1534+ .viewers1535+ .get(&remote)1536+ .filter(|viewer| viewer.session.session_id == session_id)1537+ else {1538+ return Ok(false);1539+ };1540+ ensure!(1541+ !matches!(1542+ state.focus.state(),1543+ FocusStateV2::Held {1544+ session_id: holder_session_id,1545+ ..1546+ } if *holder_session_id == session_id1547+ ),1548+ "focused viewer must be neutralized before adaptive detachment"1549+ );1550+ viewer.snapshots.send_replace(None);1551+ state.viewers.remove(&remote);1552+ if state.viewers.is_empty() {1553+ state.media = None;1554+ }1555+ advance_v2_revision(&mut state)?;1556+ publish_v2_snapshots(&state)?;1557+ drop(state);1558+ self.pending_moq1559+ .lock()1560+ .expect("MoQ registry poisoned")1561+ .remove(&(remote, session_id));1562+ self.session_changed.notify_waiters();1563+ Ok(true)1564+ }1565+15181566 fn release_input(&self, remote: EndpointId, session_id: u64) {15191567 let mut active = self.active.lock().expect("session registry poisoned");15201568 if let Some(session) = active.as_mut()@@ -1729,6 +1777,14 @@ pub(super) struct FeedbackLease {
17291777 pub(super) telemetry: Arc<MediaV3Telemetry>,17301778 pub(super) encoder_control: Option<EncoderControl>,17311779 pub(super) authorization_revision: u64,1780+ pub(super) media_generation_id: u64,1781+ pub(super) scope: FeedbackScope,1782+}1783+1784+1785+pub(super) enum FeedbackScope {1786+ Legacy,1787+ SharedGeneration,17321788 }17331789 17341790 @@ -237,6 +237,12 @@ export function normalizeAdaptiveDecisionEnvelope(payload, generation) {
237237 }238238 const decision = payload.decision;239239 if (!decision || typeof decision !== 'object') throw new TypeError('adaptive decision body is required');240+ const contribution = ['steady', 'clean', 'limiting', 'stale'].includes(decision.contribution)241+ ? decision.contribution242+ : 'unknown';243+ const recoveryState = ['active', 'floor-limited', 'detached'].includes(decision.recovery_state)244+ ? decision.recovery_state245+ : 'unknown';240246 return {241247 decision_id: boundedInteger(decision.decision_id, Number.MAX_SAFE_INTEGER, 'decision ID'),242248 report_id: boundedInteger(decision.report_id, Number.MAX_SAFE_INTEGER, 'report ID'),@@ -248,6 +254,9 @@ export function normalizeAdaptiveDecisionEnvelope(payload, generation) {
248254 ? decision.reasons.filter((reason) => typeof reason === 'string').slice(0, 8)249255 : [],250256 applied: decision.applied === true,257+ contribution,258+ recovery_state: recoveryState,259+ scope: decision.scope === 'local-viewer' ? 'local-viewer' : 'unknown',251260 };252261 }253262 @@ -256,5 +265,5 @@ export function formatAdaptiveDecision(decision, available) {
256265 if (!decision) return 'awaiting report · advisory only';257266 const reasons = decision.reasons.length ? decision.reasons.join(', ') : 'no pressure signals';258267 const disposition = decision.applied ? 'host reports applied' : 'advisory only (not applied)';259- return `${decision.target_kbps} kbps · ${decision.state} · ${reasons} · ${disposition}`;268+ return `${decision.target_kbps} kbps global · ${decision.state} · local ${decision.contribution} · recovery ${decision.recovery_state} · ${reasons} · ${disposition}`;260269 }@@ -235,8 +235,13 @@ test('decision diagnostics reject stale generations and say advisory not applied
235235 state: 'hold',236236 reasons: ['clean-recovery'],237237 applied: false,238+ contribution: 'clean',239+ recovery_state: 'active',240+ scope: 'local-viewer',238241 },239242 }, 7);240243 assert.match(formatAdaptiveDecision(decision, true), /advisory only \(not applied\)/);244+ assert.match(formatAdaptiveDecision(decision, true), /local clean · recovery active/);245+ assert.equal(decision.scope, 'local-viewer');241246 assert.equal(formatAdaptiveDecision(null, false), 'unavailable');242247 });@@ -859,7 +859,7 @@ function updateStreamStats() {
859859 document.getElementById('stream-network-audio').textContent = networkPresentation.audio;860860 document.getElementById('stream-network-input-ack').textContent = networkPresentation.inputAck;861861 document.getElementById('stream-adaptive-feedback').textContent = adaptiveFeedbackAvailable862- ? 'authenticated · 1 Hz bounded reports'862+ ? `authenticated · 1 Hz bounded reports · ${networkPresentation.adaptive}`863863 : adaptiveFeedbackError || 'unavailable';864864 document.getElementById('stream-adaptive-decision').textContent = formatAdaptiveDecision(865865 adaptiveDecision,@@ -195,6 +195,39 @@ function normalizeInputAck(value) {
195195 };196196 }197197 198+function normalizeAdaptive(value) {199+ const adaptive = record(value, 'network_diagnostics.adaptive');200+ const localPressure = new Set(['awaiting_report', 'clean', 'pressured', 'recovering']);201+ const aggregateState = new Set(['unavailable', 'hold', 'decrease', 'increase']);202+ const recoveryState = new Set(['active', 'floor_limited', 'detached']);203+ if (adaptive.scope !== 'local_viewer') {204+ throw new TypeError('network_diagnostics.adaptive.scope must be local_viewer');205+ }206+ if (!localPressure.has(adaptive.local_pressure)) {207+ throw new TypeError('network_diagnostics.adaptive.local_pressure is unsupported');208+ }209+ if (!aggregateState.has(adaptive.aggregate_state)) {210+ throw new TypeError('network_diagnostics.adaptive.aggregate_state is unsupported');211+ }212+ if (!recoveryState.has(adaptive.recovery_state)) {213+ throw new TypeError('network_diagnostics.adaptive.recovery_state is unsupported');214+ }215+ const aggregateTargetKbps = adaptive.aggregate_target_kbps === null216+ ? null217+ : exactUnsigned(218+ adaptive.aggregate_target_kbps,219+ 'network_diagnostics.adaptive.aggregate_target_kbps',220+ );221+ return {222+ scope: 'local-viewer',223+ localPressure: adaptive.local_pressure.replace('_', '-'),224+ aggregateTargetKbps,225+ aggregateState: adaptive.aggregate_state,226+ recoveryState: adaptive.recovery_state.replace('_', '-'),227+ applied: boolean(adaptive.applied, 'network_diagnostics.adaptive.applied'),228+ };229+}230+198231 /**199232 * Strictly validates the versioned network snapshot and returns a presentation-safe200233 * allowlist. Peer identities, addresses, and unknown future fields never cross this boundary.@@ -217,6 +250,7 @@ export function normalizeNetworkDiagnostics(value) {
217250 ? null218251 : normalizeLeg(diagnostics.audio, 'network_diagnostics.audio'),219252 inputAck: normalizeInputAck(diagnostics.input_ack),253+ adaptive: normalizeAdaptive(diagnostics.adaptive),220254 };221255 }222256 @@ -263,6 +297,7 @@ export function networkDiagnosticsPresentation(diagnostics) {
263297 input: 'unavailable',264298 audio: 'unavailable',265299 inputAck: 'unavailable',300+ adaptive: 'unavailable',266301 };267302 }268303 return {@@ -271,5 +306,6 @@ export function networkDiagnosticsPresentation(diagnostics) {
271306 input: formatNetworkLeg(diagnostics.input),272307 audio: formatNetworkLeg(diagnostics.audio),273308 inputAck: formatInputAck(diagnostics.inputAck),309+ adaptive: `${diagnostics.adaptive.scope} · local ${diagnostics.adaptive.localPressure} · aggregate ${diagnostics.adaptive.aggregateTargetKbps === null ? 'awaiting decision' : `${diagnostics.adaptive.aggregateTargetKbps} kbps ${diagnostics.adaptive.aggregateState}`} · recovery ${diagnostics.adaptive.recoveryState} · ${diagnostics.adaptive.applied ? 'applied' : 'advisory'}`,274310 };275311 }@@ -62,6 +62,14 @@ function snapshot(overrides = {}) {
6262 latency_overflow_total: 0,6363 complete: true,6464 },65+ adaptive: {66+ scope: 'local_viewer',67+ local_pressure: 'pressured',68+ aggregate_target_kbps: 6000,69+ aggregate_state: 'decrease',70+ recovery_state: 'active',71+ applied: true,72+ },6573 ...overrides,6674 };6775 }@@ -82,6 +90,14 @@ test('normalizes a bounded v1 snapshot and keeps transport legs separate', () =>
8290 overflowTotal: 0,8391 });8492 assert.equal(diagnostics.inputAck.pendingCount, 1);93+ assert.deepEqual(diagnostics.adaptive, {94+ scope: 'local-viewer',95+ localPressure: 'pressured',96+ aggregateTargetKbps: 6000,97+ aggregateState: 'decrease',98+ recoveryState: 'active',99+ applied: true,100+ });85101 });86102 87103 test('validates exact counters, classified samples, percentiles, and completeness', () => {@@ -155,6 +171,7 @@ test('renders legs and ACK health without exposing identifiers or addresses', ()
155171 assert.match(presentation.input, /^relay · RTT/);156172 assert.equal(presentation.audio, 'unavailable');157173 assert.match(presentation.inputAck, /^negotiated · 9\/10 acknowledged/);174+ assert.match(presentation.adaptive, /^local-viewer · local pressured · aggregate 6000 kbps decrease/);158175 assert.doesNotMatch(JSON.stringify(presentation), /secret|192\.0\.2\.1|endpoint/i);159176 });160177 @@ -165,5 +182,24 @@ test('formats a missing v4 snapshot as explicitly unavailable', () => {
165182 input: 'unavailable',166183 audio: 'unavailable',167184 inputAck: 'unavailable',185+ adaptive: 'unavailable',168186 });169187 });188+189+test('adaptive diagnostics expose only local contribution and aggregate decision', () => {190+ const diagnostics = normalizeNetworkDiagnostics(snapshot({191+ adaptive: {192+ scope: 'local_viewer',193+ local_pressure: 'recovering',194+ aggregate_target_kbps: 1000,195+ aggregate_state: 'hold',196+ recovery_state: 'floor_limited',197+ applied: false,198+ other_viewer_rtt_ms: 900,199+ contributor_peer: 'secret-peer',200+ },201+ }));202+ assert.equal(diagnostics.adaptive.localPressure, 'recovering');203+ assert.equal(diagnostics.adaptive.recoveryState, 'floor-limited');204+ assert.doesNotMatch(JSON.stringify(diagnostics.adaptive), /other|peer|rtt/i);205+});@@ -31,6 +31,8 @@ legacy_exclusive=false
3131 expect_second_rejected=false3232 focus_handoffs=03333 assert_neutral_before_successor=false34+stall_viewer=035+assert_survivor_recovery=false3436 3537 usage() {3638 cat <<'EOF'@@ -61,6 +63,9 @@ Options:
6163 --focus-handoffs COUNT Run bounded holder-approved control transfers6264 --assert-neutral-before-successor6365 Require host ordering evidence for every transfer66+ --stall-viewer INDEX Drive this control-v2 viewer persistently below67+ the shared bitrate floor68+ --assert-survivor-recovery Require clean viewers to recover after isolation6469 --help Show this help6570 EOF6671 }@@ -161,6 +166,15 @@ while [[ $# -gt 0 ]]; do
161166 assert_neutral_before_successor=true162167 shift163168 ;;169+ --stall-viewer)170+ [[ $# -ge 2 ]] || die "--stall-viewer requires a value"171+ stall_viewer="$2"172+ shift 2173+ ;;174+ --assert-survivor-recovery)175+ assert_survivor_recovery=true176+ shift177+ ;;164178 --help|-h)165179 usage166180 exit 0@@ -203,6 +217,21 @@ esac
203217 if [[ "$replace_viewer" -gt "$viewers" ]]; then204218 die "--replace-viewer must identify one of the configured viewers"205219 fi220+case "$stall_viewer" in221+ ''|*[!0-9]*) die "--stall-viewer must be a non-negative integer" ;;222+esac223+if [[ "$stall_viewer" -gt "$viewers" ]]; then224+ die "--stall-viewer must identify one of the configured viewers"225+fi226+if [[ "$stall_viewer" -gt 0 && ( "$control_v2" != true || "$viewers" -lt 2 ) ]]; then227+ die "--stall-viewer requires --control-v2 with at least two viewers"228+fi229+if [[ "$assert_survivor_recovery" == true && "$stall_viewer" -eq 0 ]]; then230+ die "--assert-survivor-recovery requires --stall-viewer"231+fi232+if [[ "$stall_viewer" -gt 0 && "$command_timeout_seconds" -lt 75 ]]; then233+ command_timeout_seconds=75234+fi206235 if [[ "$control_v2" == true && "$media_v3" == true ]]; then207236 die "--control-v2 conflicts with --media-v3"208237 fi@@ -607,7 +636,14 @@ if [[ "$control_v2" == true ]]; then
607636 viewer_identities+=("$viewer_identity")608637 viewer_log="$tmp_root/control-v2-viewer-${viewer}.log"609638 viewer_logs+=("$viewer_log")610- if [[ "$focus_handoffs" -gt 0 && "$viewer" -le 2 ]]; then639+ if [[ "$stall_viewer" -gt 0 ]]; then640+ adaptive_profile="clean"641+ if [[ "$viewer" -eq "$stall_viewer" ]]; then adaptive_profile="floor-breach"; fi642+ "$probe_bin" --control-v2 --spectator --adaptive-profile "$adaptive_profile" \643+ --identity "$viewer_identity" --node-id "$node_id" --frames "$primary_frames" \644+ --timeout-seconds "$probe_timeout_seconds" --expect-size 1280x800 \645+ >"$viewer_log" 2>&1 &646+ elif [[ "$focus_handoffs" -gt 0 && "$viewer" -le 2 ]]; then611647 "$probe_bin" --control-v2 --focus-handoffs "$focus_handoffs" \612648 --identity "$viewer_identity" --node-id "$node_id" --frames "$primary_frames" \613649 --timeout-seconds "$probe_timeout_seconds" --expect-size 1280x800 \@@ -624,7 +660,7 @@ if [[ "$control_v2" == true ]]; then
624660 >"$viewer_log" 2>&1 &625661 fi626662 viewer_pids+=("$!")627- if [[ "$viewer" -eq 1 && "$focus_handoffs" -eq 0 ]]; then663+ if [[ "$viewer" -eq 1 && "$focus_handoffs" -eq 0 && "$stall_viewer" -eq 0 ]]; then628664 wait_for_log_count "$host_log" 'input v2 client accepted' 1 "$host_pid" 'sigil' \629665 || die "slot-0 holder did not acquire focused input before spectators joined"630666 fi@@ -667,7 +703,25 @@ if [[ "$control_v2" == true ]]; then
667703 viewer=$((viewer + 1))668704 done669705 viewer_pids=()670- if [[ "$focus_handoffs" -gt 0 ]]; then706+ if [[ "$stall_viewer" -gt 0 ]]; then707+ viewer=1708+ while [[ "$viewer" -le "$viewers" ]]; do709+ if [[ "$viewer" -eq "$stall_viewer" ]]; then710+ grep -Fxq 'adaptive_profile=floor-breach' "${viewer_logs[$((viewer - 1))]}" \711+ || die "stalled viewer did not run the floor-breach profile"712+ grep -Fxq 'adaptive_floor_detach=ok' "${viewer_logs[$((viewer - 1))]}" \713+ || die "stalled viewer was not detached after bounded isolated recovery"714+ elif [[ "$assert_survivor_recovery" == true ]]; then715+ grep -Fxq 'adaptive_profile=clean' "${viewer_logs[$((viewer - 1))]}" \716+ || die "survivor did not run the clean adaptive profile"717+ grep -Fxq 'adaptive_survivor_recovery=ok' "${viewer_logs[$((viewer - 1))]}" \718+ || die "survivor did not recover after the stalled viewer was isolated"719+ fi720+ viewer=$((viewer + 1))721+ done722+ grep -Fq 'detached persistently below-floor viewer after bounded isolated recovery' "$host_log" \723+ || die "host did not record terminal slow-viewer isolation"724+ elif [[ "$focus_handoffs" -gt 0 ]]; then671725 participant=1672726 while [[ "$participant" -le 2 ]]; do673727 grep -Fxq "focus_handoffs_observed=${focus_handoffs}" "${viewer_logs[$((participant - 1))]}" \@@ -678,6 +732,7 @@ if [[ "$control_v2" == true ]]; then
678732 grep -Fxq 'slot_0_input=ok' "${viewer_logs[0]}" || die "slot-0 holder input failed"679733 fi680734 spectator=$([[ "$focus_handoffs" -gt 0 ]] && printf 3 || printf 2)735+ if [[ "$stall_viewer" -gt 0 ]]; then spectator=1; fi681736 while [[ "$spectator" -le "$viewers" ]]; do682737 if [[ "$spectator" -eq "$replace_viewer" ]]; then683738 spectator=$((spectator + 1))@@ -744,9 +799,11 @@ if [[ "$control_v2" == true ]]; then
744799 printf 'viewers=%s\n' "$viewers"745800 printf 'same_peer_replacement=%s\n' "$([[ "$replace_viewer" -gt 0 ]] && printf ok || printf not-requested)"746801 printf 'shared_generation=ok\n'747- printf 'slot_0_single_holder=ok\n'802+ printf 'slot_0_single_holder=%s\n' "$([[ "$stall_viewer" -gt 0 ]] && printf not-requested || printf ok)"748803 printf 'focus_handoffs=%s\n' "$focus_handoffs"749804 printf 'neutral_before_successor=%s\n' "$([[ "$assert_neutral_before_successor" == true ]] && printf ok || printf not-requested)"805+ printf 'stalled_viewer=%s\n' "$stall_viewer"806+ printf 'survivor_recovery=%s\n' "$([[ "$assert_survivor_recovery" == true ]] && printf ok || printf not-requested)"750807 printf 'spectator_progress=ok\n'751808 exit 0752809 fi@@ -1,4 +1,5 @@
11 use std::sync::atomic::{AtomicU64, Ordering};2+use std::sync::{Arc, Mutex as StdMutex};23 use std::time::Duration;34 45 use iroh::Endpoint;@@ -12,6 +13,7 @@ use tauri::{AppHandle, Emitter, Manager};
1213 1314 use crate::commands::state::{AccumulatedMediaFeedback, AppState};1415 use crate::media::frame_channel::take_generation_owned_triple;16+use crate::media::network_diagnostics::{NetworkSessionDiagnostics, lock_network_diagnostics};1517 use crate::media::transport::{connect_error_is_unsupported_alpn, negotiate_v1};1618 1719 pub(crate) const CLIENT_MEDIA_FEEDBACK_CONNECT_TIMEOUT: Duration = Duration::from_secs(3);@@ -68,6 +70,9 @@ struct AdaptiveBitrateDecisionDiagnostic {
6870 state: &'static str,6971 reasons: Vec<&'static str>,7072 applied: bool,73+ contribution: &'static str,74+ recovery_state: &'static str,75+ scope: &'static str,7176 }7277 7378 pub(crate) async fn open_negotiated_feedback_stream(@@ -163,6 +168,41 @@ fn adaptive_bitrate_reason_names(reasons: AdaptiveBitrateReasonFlagsV1) -> Vec<&
163168 fn adaptive_bitrate_decision_diagnostic(164169 decision: AdaptiveBitrateDecisionV1,165170 ) -> AdaptiveBitrateDecisionDiagnostic {171+ let pressured = decision172+ .reasons173+ .contains(AdaptiveBitrateReasonFlagsV1::RECEIVER_QUEUE)174+ || decision175+ .reasons176+ .contains(AdaptiveBitrateReasonFlagsV1::DECODE_BACKLOG)177+ || decision178+ .reasons179+ .contains(AdaptiveBitrateReasonFlagsV1::DELIVERY_LATENCY)180+ || decision181+ .reasons182+ .contains(AdaptiveBitrateReasonFlagsV1::LOSS_OR_CANCELLATION)183+ || decision184+ .reasons185+ .contains(AdaptiveBitrateReasonFlagsV1::RTT_INFLATION);186+ let contribution = if decision187+ .reasons188+ .contains(AdaptiveBitrateReasonFlagsV1::FEEDBACK_STALE)189+ {190+ "stale"191+ } else if pressured {192+ "limiting"193+ } else if decision194+ .reasons195+ .contains(AdaptiveBitrateReasonFlagsV1::CLEAN_RECOVERY)196+ {197+ "clean"198+ } else {199+ "steady"200+ };201+ let recovery_state = if pressured && decision.target_kbps == decision.floor_kbps {202+ "floor-limited"203+ } else {204+ "active"205+ };166206 AdaptiveBitrateDecisionDiagnostic {167207 decision_id: decision.decision_id,168208 report_id: decision.report_id,@@ -172,6 +212,9 @@ fn adaptive_bitrate_decision_diagnostic(
172212 state: adaptive_bitrate_state_name(decision.state),173213 reasons: adaptive_bitrate_reason_names(decision.reasons),174214 applied: decision.applied,215+ contribution,216+ recovery_state,217+ scope: "local-viewer",175218 }176219 }177220 @@ -242,7 +285,9 @@ pub(crate) async fn run_media_feedback_session(
242285 mut send: iroh::endpoint::SendStream,243286 mut recv: iroh::endpoint::RecvStream,244287 mut reports: tokio::sync::watch::Receiver<Option<AccumulatedMediaFeedback>>,288+ network_diagnostics: Arc<StdMutex<NetworkSessionDiagnostics>>,245289 ) {290+ let writer_diagnostics = Arc::clone(&network_diagnostics);246291 let (decision_tx, decision_rx) = tokio::sync::watch::channel(None);247292 let writer = async {248293 let mut last_written = None;@@ -256,6 +301,7 @@ pub(crate) async fn run_media_feedback_session(
256301 continue;257302 };258303 let report = accumulated.report_since(last_written);304+ lock_network_diagnostics(&writer_diagnostics).observe_adaptive_report(&report);259305 tokio::time::timeout(260306 CLIENT_MEDIA_FEEDBACK_IO_TIMEOUT,261307 write_media_feedback_report_v1(&mut send, &report),@@ -268,6 +314,7 @@ pub(crate) async fn run_media_feedback_session(
268314 269315 Ok::<(), String>(())270316 };317+ let reader_diagnostics = Arc::clone(&network_diagnostics);271318 let reader = async {272319 let mut sequence = AdaptiveDecisionSequence::default();273320 loop {@@ -276,6 +323,7 @@ pub(crate) async fn run_media_feedback_session(
276323 .map_err(|error| format!("adaptive decision read failed: {error}"))?277324 .ok_or_else(|| "adaptive decision stream closed".to_string())?;278325 sequence.accept(&decision)?;326+ lock_network_diagnostics(&reader_diagnostics).observe_adaptive_decision(&decision);279327 decision_tx.send_replace(Some(decision));280328 }281329 @@ -298,12 +346,14 @@ pub(crate) async fn run_media_feedback_session(
298346 terminal = decision_emitter => terminal,299347 };300348 if let Err(error) = terminal {349+ lock_network_diagnostics(&network_diagnostics).mark_adaptive_detached();301350 eprintln!("[client] {error}");302351 if let Err(emit_error) = app.emit(303352 "adaptive-feedback-state",304353 serde_json::json!({305354 "generation": generation,306355 "available": false,356+ "recovery_state": "detached",307357 "error": error,308358 }),309359 ) {@@ -486,6 +536,9 @@ mod tests {
486536 assert_eq!(diagnostic.state, "decrease");487537 assert_eq!(diagnostic.reasons, vec!["receiver-queue", "decode-backlog"]);488538 assert!(!diagnostic.applied);539+ assert_eq!(diagnostic.contribution, "limiting");540+ assert_eq!(diagnostic.recovery_state, "active");541+ assert_eq!(diagnostic.scope, "local-viewer");489542 }490543 491544 fn adaptive_decision(decision_id: u64, report_id: u64) -> AdaptiveBitrateDecisionV1 {@@ -418,6 +418,7 @@ pub(crate) async fn connect_client(
418418 send,419419 recv,420420 feedback_rx,421+ Arc::clone(&network_diagnostics),421422 ));422423 } else {423424 *state.media_feedback.lock().await = None;@@ -8,6 +8,10 @@ use std::{
88 99 use iroh::endpoint::{Connection, PathId};1010 use serde::Serialize;11+use sigil_protocol::{12+ AdaptiveBitrateDecisionV1, AdaptiveBitrateReasonFlagsV1, AdaptiveBitrateStateV1,13+ MediaFeedbackFlags, MediaFeedbackReportV1,14+};1115 1216 const LATENCY_BUCKETS_MS: usize = 5_001;1317 pub(crate) const INPUT_ACK_PENDING_CAPACITY: usize = 1_024;@@ -64,6 +68,77 @@ struct MediaAuthenticationDiagnostics {
6468 verification_failures: Arc<AtomicU64>,6569 }6670 71+72+pub(crate) struct AdaptiveViewerSnapshot {73+ scope: &'static str,74+ local_pressure: &'static str,75+ aggregate_target_kbps: Option<u32>,76+ aggregate_state: &'static str,77+ recovery_state: &'static str,78+ applied: bool,79+}80+81+impl Default for AdaptiveViewerSnapshot {82+ fn default() -> Self {83+ Self {84+ scope: "local_viewer",85+ local_pressure: "awaiting_report",86+ aggregate_target_kbps: None,87+ aggregate_state: "unavailable",88+ recovery_state: "active",89+ applied: false,90+ }91+ }92+}93+94+impl AdaptiveViewerSnapshot {95+ fn observe_report(&mut self, report: &MediaFeedbackReportV1) {96+ let drops = report97+ .transport_dropped_delta98+ .saturating_add(report.frontend_dropped_delta)99+ .saturating_add(report.decoder_dropped_delta)100+ .saturating_add(report.presenter_dropped_delta);101+ let queued = report.frontend_queue_depth > 0102+ || report.decode_queue_depth > 0103+ || report.presenter_queue_depth > 0;104+ self.local_pressure = if report.flags.contains(MediaFeedbackFlags::RESYNC_ACTIVE) {105+ "recovering"106+ } else if drops > 0 || queued {107+ "pressured"108+ } else {109+ "clean"110+ };111+ }112+113+ fn observe_decision(&mut self, decision: &AdaptiveBitrateDecisionV1) {114+ self.aggregate_target_kbps = Some(decision.target_kbps);115+ self.aggregate_state = match decision.state {116+ AdaptiveBitrateStateV1::Hold => "hold",117+ AdaptiveBitrateStateV1::Decrease => "decrease",118+ AdaptiveBitrateStateV1::Increase => "increase",119+ };120+ let local_pressure = decision121+ .reasons122+ .contains(AdaptiveBitrateReasonFlagsV1::RECEIVER_QUEUE)123+ || decision124+ .reasons125+ .contains(AdaptiveBitrateReasonFlagsV1::DECODE_BACKLOG)126+ || decision127+ .reasons128+ .contains(AdaptiveBitrateReasonFlagsV1::DELIVERY_LATENCY);129+ self.recovery_state = if local_pressure && decision.target_kbps == decision.floor_kbps {130+ "floor_limited"131+ } else {132+ "active"133+ };134+ self.applied = decision.applied;135+ }136+137+ fn detach(&mut self) {138+ self.recovery_state = "detached";139+ }140+}141+67142 impl Default for MediaAuthenticationDiagnostics {68143 fn default() -> Self {69144 Self {@@ -536,6 +611,7 @@ pub(crate) struct NetworkSessionDiagnostics {
536611 audio: Option<LegDiagnostics>,537612 input_ack: InputAckDiagnostics,538613 media_authentication: MediaAuthenticationDiagnostics,614+ adaptive: AdaptiveViewerSnapshot,539615 }540616 541617 impl NetworkSessionDiagnostics {@@ -547,6 +623,7 @@ impl NetworkSessionDiagnostics {
547623 audio: None,548624 input_ack: InputAckDiagnostics::new(input_ack_negotiated),549625 media_authentication: MediaAuthenticationDiagnostics::default(),626+ adaptive: AdaptiveViewerSnapshot::default(),550627 }551628 }552629 @@ -594,6 +671,18 @@ impl NetworkSessionDiagnostics {
594671 }595672 }596673 674+ pub(crate) fn observe_adaptive_report(&mut self, report: &MediaFeedbackReportV1) {675+ self.adaptive.observe_report(report);676+ }677+678+ pub(crate) fn observe_adaptive_decision(&mut self, decision: &AdaptiveBitrateDecisionV1) {679+ self.adaptive.observe_decision(decision);680+ }681+682+ pub(crate) fn mark_adaptive_detached(&mut self) {683+ self.adaptive.detach();684+ }685+597686 pub(crate) fn snapshot(&self, now: Instant) -> NetworkDiagnosticsSnapshot {598687 NetworkDiagnosticsSnapshot {599688 version: 1,@@ -606,6 +695,7 @@ impl NetworkSessionDiagnostics {
606695 audio: self.audio.as_ref().map(LegDiagnostics::snapshot),607696 input_ack: self.input_ack.snapshot(),608697 media_authentication: self.media_authentication.snapshot(),698+ adaptive: self.adaptive.clone(),609699 }610700 }611701 }@@ -619,6 +709,7 @@ pub(crate) struct NetworkDiagnosticsSnapshot {
619709 audio: Option<NetworkLegSnapshot>,620710 input_ack: InputAckSnapshot,621711 media_authentication: MediaAuthenticationSnapshot,712+ adaptive: AdaptiveViewerSnapshot,622713 }623714 624715 @@ -822,4 +913,52 @@ mod tests {
822913 assert!(!serialized.contains("capability"));823914 assert!(!serialized.contains("public_key"));824915 }916+917+ 918+ fn adaptive_diagnostics_separate_local_pressure_from_the_aggregate_decision() {919+ let now = Instant::now();920+ let mut diagnostics = NetworkSessionDiagnostics::new(now, false);921+ let mut report = MediaFeedbackReportV1 {922+ report_id: 1,923+ interval_ms: 1_000,924+ flags: MediaFeedbackFlags::RESYNC_ACTIVE,925+ last_sequence: Some(10),926+ transport_dropped_delta: 1,927+ frontend_dropped_delta: 0,928+ decoder_dropped_delta: 0,929+ presenter_dropped_delta: 0,930+ frontend_queue_depth: 1,931+ frontend_queue_capacity: 4,932+ decode_queue_depth: 0,933+ decode_queue_capacity: 4,934+ presenter_queue_depth: 0,935+ presenter_queue_capacity: 2,936+ transport_delivery_p95_ms: Some(10),937+ decode_p95_ms: Some(3),938+ presentation_p95_ms: Some(5),939+ };940+ diagnostics.observe_adaptive_report(&report);941+ let decision = AdaptiveBitrateDecisionV1 {942+ decision_id: 1,943+ report_id: 1,944+ target_kbps: 1_000,945+ floor_kbps: 1_000,946+ ceiling_kbps: 12_000,947+ state: AdaptiveBitrateStateV1::Decrease,948+ reasons: AdaptiveBitrateReasonFlagsV1::RECEIVER_QUEUE,949+ applied: false,950+ };951+ diagnostics.observe_adaptive_decision(&decision);952+ let value = serde_json::to_value(diagnostics.snapshot(now)).unwrap();953+ assert_eq!(value["adaptive"]["scope"], "local_viewer");954+ assert_eq!(value["adaptive"]["local_pressure"], "recovering");955+ assert_eq!(value["adaptive"]["aggregate_target_kbps"], 1_000);956+ assert_eq!(value["adaptive"]["recovery_state"], "floor_limited");957+ report.flags = MediaFeedbackFlags::NONE;958+ diagnostics.observe_adaptive_report(&report);959+ assert_eq!(960+ serde_json::to_value(diagnostics.snapshot(now)).unwrap()["adaptive"]["local_pressure"],961+ "pressured"962+ );963+ }825964 }