1use std::collections::HashMap;2use std::sync::atomic::{AtomicUsize, Ordering};3use std::sync::{Arc, Mutex};45use super::session::{6 AdaptiveEncoderProposal, FeedbackLease, FeedbackScope, MediaV3TelemetrySnapshot,7 ResolutionEncoderProposal, SessionRegistry,8};9use super::*;1011const ADAPTIVE_BITRATE_FLOOR_KBPS: u32 = 1_000;12const SHADOW_BITRATE_CEILING_KBPS: u32 = 12_000;13const ADAPTIVE_BITRATE_QUANTUM_KBPS: u32 = 250;14const ADAPTIVE_BITRATE_CLEAN_WINDOWS: u8 = 10;15const ADAPTIVE_BITRATE_MODERATE_WINDOWS: u8 = 2;16const ADAPTIVE_BITRATE_COOLDOWN_WINDOWS: u8 = 10;17const MOTION_RESOLUTION_MODERATE_WINDOWS: u8 = 2;18const MOTION_RESOLUTION_CLEAN_WINDOWS: u8 = 3;19const MOTION_RESOLUTION_ACTIVITY_FPS: u64 = 10;20const MOTION_RESOLUTION_STILL_SETTLE: Duration = Duration::from_secs(2);21const MOTION_RESOLUTION_UPSCALE_COOLDOWN: Duration = Duration::from_secs(3);22const FEEDBACK_FRESH_INTERVAL_MIN_MS: u16 = 750;23const FEEDBACK_FRESH_INTERVAL_MAX_MS: u16 = 1_500;24const FEEDBACK_FRESH_ARRIVAL_GAP_MIN: Duration = Duration::from_millis(750);25const FEEDBACK_STALE_ARRIVAL_GAP: Duration = Duration::from_millis(2_500);26const FEEDBACK_MIN_READ_INTERVAL: Duration = Duration::from_millis(250);27const FEEDBACK_WRITE_TIMEOUT: Duration = Duration::from_secs(1);28// Mirrors the fixed v1 wire bound. Aggregates clamp here so a stalled29// consumer observes a stale 5-second window instead of a deceptively fresh30// latest report.31const MEDIA_FEEDBACK_INTERVAL_MAX_MS: u64 = 5_000;32const AGGREGATE_TICK_INTERVAL: Duration = Duration::from_millis(250);33const FLOOR_BREACH_WINDOWS: u8 = 3;34const ISOLATED_RECOVERY_CLEAN_WINDOWS: u8 = 3;35const ISOLATED_RECOVERY_PRESSURE_WINDOWS: u8 = 3;36const ISOLATED_RECOVERY_MAX_ATTEMPTS: u8 = 2;37const ISOLATED_RECOVERY_BACKOFF: Duration = Duration::from_secs(2);38// All policy state is one fixed record per hard-bounded viewer. Reports use a39// latest cumulative ingress slot and the generation owns one actuator/queued40// plan, so eight viewers cannot multiply encoder work or queue history.414243enum FeedbackSeverity {44 Clean,45 InsufficientEvidence,46 Moderate,47 Severe,48 Stale,49}505152pub(crate) struct VideoDimensions {53 pub width: u16,54 pub height: u16,55}5657impl VideoDimensions {58 fn three_quarter(self) -> Result<Self> {59 ensure!(60 self.width.is_multiple_of(4) && self.height.is_multiple_of(4),61 "native dimensions must be divisible by four for an exact three-quarter tier"62 );63 let reduced = Self {64 width: self.width / 4 * 3,65 height: self.height / 4 * 3,66 };67 ensure!(68 reduced.width >= 64 && reduced.height >= 64,69 "three-quarter dimensions are below the H.264 minimum"70 );71 ensure!(72 reduced.width.is_multiple_of(2) && reduced.height.is_multiple_of(2),73 "three-quarter H.264 dimensions must be even"74 );75 ensure!(76 u32::from(self.width) * u32::from(reduced.height)77 == u32::from(self.height) * u32::from(reduced.width),78 "three-quarter dimensions must preserve the native aspect ratio"79 );80 Ok(reduced)81 }82}838485enum ResolutionTier {86 Native,87 Reduced,88}899091enum ResolutionTrigger {92 Hold,93 Motion,94 Pressure,95 Recovery,96}979899struct MotionResolutionDecision {100 tier: ResolutionTier,101 target: VideoDimensions,102 trigger: ResolutionTrigger,103 changed: bool,104}105106/// Pure two-tier resolution policy. Fresh sequence progress from Gamescope's107/// damage-driven stream is the bounded motion proxy while the authenticated108/// feedback path supplies pressure. A downshift is safety-biased and109/// immediate; an upscale requires both settled motion and fresh clean110/// feedback, so stale reports can never restore detail.111112pub(crate) struct MotionResolutionPolicy {113 native: VideoDimensions,114 reduced: VideoDimensions,115 tier: ResolutionTier,116 motion_active: bool,117 still_since: Option<Instant>,118 pressure_active: bool,119 moderate_windows: u8,120 clean_windows: u8,121 last_transition_at: Option<Instant>,122 last_motion_sequence: Option<u64>,123}124125impl MotionResolutionPolicy {126 pub(crate) fn new(native: VideoDimensions) -> Result<Self> {127 ensure!(128 native.width >= 64 && native.height >= 64,129 "native dimensions are too small"130 );131 ensure!(132 native.width.is_multiple_of(2) && native.height.is_multiple_of(2),133 "native H.264 dimensions must be even"134 );135 Ok(Self {136 native,137 reduced: native.three_quarter()?,138 tier: ResolutionTier::Native,139 motion_active: false,140 still_since: None,141 pressure_active: false,142 moderate_windows: 0,143 clean_windows: 0,144 last_transition_at: None,145 last_motion_sequence: None,146 })147 }148149 pub(crate) fn target(&self) -> VideoDimensions {150 match self.tier {151 ResolutionTier::Native => self.native,152 ResolutionTier::Reduced => self.reduced,153 }154 }155156 /// Host motion hook. Repeated still observations preserve the original157 /// settle boundary instead of indefinitely postponing recovery.158 159 fn observe_motion(160 &mut self,161 motion_active: bool,162 observed_at: Instant,163 ) -> MotionResolutionDecision {164 self.update_motion(motion_active, observed_at);165 self.evaluate_target(observed_at)166 }167168 fn update_motion(&mut self, motion_active: bool, observed_at: Instant) {169 if motion_active {170 self.motion_active = true;171 self.still_since = None;172 return;173 }174 if self.motion_active || self.still_since.is_none() {175 self.still_since = Some(observed_at);176 }177 self.motion_active = false;178 }179180 /// Observe one classified feedback window. Only fresh moderate/severe181 /// pressure can initiate a downshift, and only fresh clean windows count182 /// toward restoring native detail.183 184 fn observe_feedback(185 &mut self,186 severity: FeedbackSeverity,187 fresh: bool,188 observed_at: Instant,189 ) -> MotionResolutionDecision {190 self.update_feedback(severity, fresh);191 self.evaluate_target(observed_at)192 }193194 /// Gamescope output is damage-driven, so fresh media sequence progress is195 /// a conservative constant-space motion proxy for this first slice. A196 /// future raw-frame observer can call `observe_motion` without changing197 /// pressure hysteresis or encoder actuation.198 fn observe_window(199 &mut self,200 report: &MediaFeedbackReportV1,201 evaluation: &AdaptiveBitrateEvaluation,202 observed_at: Instant,203 ) -> MotionResolutionDecision {204 let previous_sequence = self.last_motion_sequence;205 if let Some(sequence) = report.last_sequence {206 self.last_motion_sequence = Some(207 self.last_motion_sequence208 .map_or(sequence, |last| last.max(sequence)),209 );210 }211 if evaluation.fresh {212 let motion_active = report.last_sequence.is_some_and(|sequence| {213 let delta = previous_sequence.map_or(0, |last| sequence.saturating_sub(last));214 delta.saturating_mul(1_000)215 >= MOTION_RESOLUTION_ACTIVITY_FPS.saturating_mul(u64::from(report.interval_ms))216 });217 self.update_motion(motion_active, observed_at);218 }219 let resolution_severity = if evaluation.pressure_severity == FeedbackSeverity::Clean220 && !evaluation.clean_recovery_evidence221 {222 FeedbackSeverity::InsufficientEvidence223 } else {224 evaluation.pressure_severity225 };226 self.update_feedback(resolution_severity, evaluation.fresh);227 self.evaluate_target(observed_at)228 }229230 fn update_feedback(&mut self, severity: FeedbackSeverity, fresh: bool) {231 if !fresh232 || matches!(233 severity,234 FeedbackSeverity::InsufficientEvidence | FeedbackSeverity::Stale235 )236 {237 self.moderate_windows = 0;238 self.clean_windows = 0;239 return;240 }241242 match severity {243 FeedbackSeverity::Severe => {244 self.pressure_active = true;245 self.moderate_windows = 0;246 self.clean_windows = 0;247 }248 FeedbackSeverity::Moderate => {249 self.clean_windows = 0;250 self.moderate_windows = self.moderate_windows.saturating_add(1);251 if self.moderate_windows >= MOTION_RESOLUTION_MODERATE_WINDOWS {252 self.moderate_windows = 0;253 self.pressure_active = true;254 }255 }256 FeedbackSeverity::Clean => {257 self.moderate_windows = 0;258 self.clean_windows = self.clean_windows.saturating_add(1);259 if self.clean_windows >= MOTION_RESOLUTION_CLEAN_WINDOWS {260 self.pressure_active = false;261 }262 }263 FeedbackSeverity::InsufficientEvidence | FeedbackSeverity::Stale => {}264 }265 }266267 fn evaluate_target(&mut self, observed_at: Instant) -> MotionResolutionDecision {268 if self.motion_active {269 self.downshift(observed_at, ResolutionTrigger::Motion)270 } else if self.pressure_active {271 self.downshift(observed_at, ResolutionTrigger::Pressure)272 } else {273 self.maybe_restore(observed_at)274 }275 }276277 fn downshift(278 &mut self,279 observed_at: Instant,280 trigger: ResolutionTrigger,281 ) -> MotionResolutionDecision {282 let changed = self.tier != ResolutionTier::Reduced;283 if changed {284 self.tier = ResolutionTier::Reduced;285 self.last_transition_at = Some(observed_at);286 }287 MotionResolutionDecision {288 tier: self.tier,289 target: self.target(),290 trigger,291 changed,292 }293 }294295 fn maybe_restore(&mut self, observed_at: Instant) -> MotionResolutionDecision {296 let still_settled = !self.motion_active297 && self.still_since.is_some_and(|still_since| {298 observed_at.saturating_duration_since(still_since) >= MOTION_RESOLUTION_STILL_SETTLE299 });300 let cooldown_complete = self.last_transition_at.is_none_or(|last| {301 observed_at.saturating_duration_since(last) >= MOTION_RESOLUTION_UPSCALE_COOLDOWN302 });303 let clean = self.clean_windows >= MOTION_RESOLUTION_CLEAN_WINDOWS;304 if self.tier == ResolutionTier::Reduced305 && still_settled306 && cooldown_complete307 && clean308 && !self.pressure_active309 {310 self.tier = ResolutionTier::Native;311 self.clean_windows = 0;312 self.last_transition_at = Some(observed_at);313 return MotionResolutionDecision {314 tier: self.tier,315 target: self.target(),316 trigger: ResolutionTrigger::Recovery,317 changed: true,318 };319 }320 self.hold()321 }322323 fn hold(&self) -> MotionResolutionDecision {324 MotionResolutionDecision {325 tier: self.tier,326 target: self.target(),327 trigger: ResolutionTrigger::Hold,328 changed: false,329 }330 }331}332333fn motion_resolution_policy(width: u32, height: u32) -> Result<Option<MotionResolutionPolicy>> {334 let native = VideoDimensions {335 width: u16::try_from(width).context("native width exceeds protocol")?,336 height: u16::try_from(height).context("native height exceeds protocol")?,337 };338 match MotionResolutionPolicy::new(native) {339 Ok(policy) => Ok(Some(policy)),340 Err(error) => {341 warn!(342 %error,343 width,344 height,345 "motion resolution disabled because this native mode has no exact reduced tier"346 );347 Ok(None)348 }349 }350}351352fn motion_resolution_policy_for_encoder(353 control: &EncoderControl,354) -> Result<Option<MotionResolutionPolicy>> {355 let (width, height) = control.initial_dimensions();356 motion_resolution_policy(u32::from(width), u32::from(height))357}358359360struct FeedbackIngressCursor {361 last_report_id: Option<u64>,362 interval_ms: u64,363 transport_dropped: u64,364 frontend_dropped: u64,365 decoder_dropped: u64,366 presenter_dropped: u64,367 resync_reports: u64,368}369370371struct CumulativeFeedbackIngress {372 latest: Option<MediaFeedbackReportV1>,373 interval_ms: u64,374 transport_dropped: u64,375 frontend_dropped: u64,376 decoder_dropped: u64,377 presenter_dropped: u64,378 resync_reports: u64,379}380381impl CumulativeFeedbackIngress {382 fn observe(&mut self, report: MediaFeedbackReportV1) {383 self.interval_ms = self.interval_ms.wrapping_add(u64::from(report.interval_ms));384 self.transport_dropped = self385 .transport_dropped386 .wrapping_add(u64::from(report.transport_dropped_delta));387 self.frontend_dropped = self388 .frontend_dropped389 .wrapping_add(u64::from(report.frontend_dropped_delta));390 self.decoder_dropped = self391 .decoder_dropped392 .wrapping_add(u64::from(report.decoder_dropped_delta));393 self.presenter_dropped = self394 .presenter_dropped395 .wrapping_add(u64::from(report.presenter_dropped_delta));396 if report.flags.contains(MediaFeedbackFlags::RESYNC_ACTIVE) {397 self.resync_reports = self.resync_reports.wrapping_add(1);398 }399 self.latest = Some(report);400 }401402 fn report_since(403 &self,404 cursor: FeedbackIngressCursor,405 ) -> Result<Option<(MediaFeedbackReportV1, FeedbackIngressCursor)>> {406 let Some(latest) = self.latest else {407 return Ok(None);408 };409 if cursor410 .last_report_id411 .is_some_and(|report_id| latest.report_id <= report_id)412 {413 return Ok(None);414 }415 let mut report = latest;416 report.interval_ms = u16::try_from(417 self.interval_ms418 .wrapping_sub(cursor.interval_ms)419 .min(MEDIA_FEEDBACK_INTERVAL_MAX_MS),420 )421 .expect("feedback interval is clamped to u16");422 report.transport_dropped_delta =423 cumulative_feedback_delta(self.transport_dropped, cursor.transport_dropped);424 report.frontend_dropped_delta =425 cumulative_feedback_delta(self.frontend_dropped, cursor.frontend_dropped);426 report.decoder_dropped_delta =427 cumulative_feedback_delta(self.decoder_dropped, cursor.decoder_dropped);428 report.presenter_dropped_delta =429 cumulative_feedback_delta(self.presenter_dropped, cursor.presenter_dropped);430 report.flags = if self.resync_reports.wrapping_sub(cursor.resync_reports) > 0 {431 MediaFeedbackFlags::RESYNC_ACTIVE432 } else {433 MediaFeedbackFlags::NONE434 };435 report.validate()?;436 Ok(Some((437 report,438 FeedbackIngressCursor {439 last_report_id: Some(latest.report_id),440 interval_ms: self.interval_ms,441 transport_dropped: self.transport_dropped,442 frontend_dropped: self.frontend_dropped,443 decoder_dropped: self.decoder_dropped,444 presenter_dropped: self.presenter_dropped,445 resync_reports: self.resync_reports,446 },447 )))448 }449}450451fn cumulative_feedback_delta(total: u64, baseline: u64) -> u32 {452 u32::try_from(total.wrapping_sub(baseline).min(u64::from(u32::MAX)))453 .expect("feedback delta is clamped to u32")454}455456457struct GenerationContributorKey {458 remote: EndpointId,459 session_id: u64,460}461462463enum ContributorRecoveryState {464 Active {465 floor_breach_windows: u8,466 },467 Recovering {468 attempt: u8,469 retry_at: Instant,470 clean_windows: u8,471 pressure_windows: u8,472 },473 Detaching,474}475476impl Default for ContributorRecoveryState {477 fn default() -> Self {478 Self::Active {479 floor_breach_windows: 0,480 }481 }482}483484impl ContributorRecoveryState {485 fn contributes(self) -> bool {486 matches!(self, Self::Active { .. })487 }488489 fn observe(490 &mut self,491 evaluation: &AdaptiveBitrateEvaluation,492 now: Instant,493 ) -> RecoveryMutation {494 let pressured = evaluation.fresh495 && matches!(496 evaluation.pressure_severity,497 FeedbackSeverity::Moderate | FeedbackSeverity::Severe498 );499 let severe = evaluation.fresh && evaluation.pressure_severity == FeedbackSeverity::Severe;500 let clean = evaluation.fresh501 && evaluation.pressure_severity == FeedbackSeverity::Clean502 && evaluation.clean_recovery_evidence;503 match *self {504 Self::Active {505 ref mut floor_breach_windows,506 } => {507 if severe && evaluation.decision.target_kbps == evaluation.decision.floor_kbps {508 *floor_breach_windows = floor_breach_windows.saturating_add(1);509 } else {510 *floor_breach_windows = 0;511 }512 if *floor_breach_windows < FLOOR_BREACH_WINDOWS {513 return RecoveryMutation::None;514 }515 *self = Self::Recovering {516 attempt: 1,517 retry_at: now + ISOLATED_RECOVERY_BACKOFF,518 clean_windows: 0,519 pressure_windows: 0,520 };521 RecoveryMutation::RequestRecovery522 }523 Self::Recovering {524 attempt,525 retry_at,526 ref mut clean_windows,527 ref mut pressure_windows,528 } => {529 if now < retry_at {530 return RecoveryMutation::None;531 }532 if clean {533 *clean_windows = clean_windows.saturating_add(1);534 *pressure_windows = 0;535 if *clean_windows >= ISOLATED_RECOVERY_CLEAN_WINDOWS {536 *self = Self::default();537 return RecoveryMutation::Recovered;538 }539 } else if pressured {540 *pressure_windows = pressure_windows.saturating_add(1);541 *clean_windows = 0;542 if *pressure_windows >= ISOLATED_RECOVERY_PRESSURE_WINDOWS {543 if attempt >= ISOLATED_RECOVERY_MAX_ATTEMPTS {544 *self = Self::Detaching;545 return RecoveryMutation::Detach;546 }547 let next_attempt = attempt.saturating_add(1);548 *self = Self::Recovering {549 attempt: next_attempt,550 retry_at: now551 + ISOLATED_RECOVERY_BACKOFF.saturating_mul(u32::from(next_attempt)),552 clean_windows: 0,553 pressure_windows: 0,554 };555 return RecoveryMutation::RequestRecovery;556 }557 } else {558 *clean_windows = 0;559 *pressure_windows = 0;560 }561 RecoveryMutation::None562 }563 Self::Detaching => RecoveryMutation::None,564 }565 }566}567568569enum RecoveryMutation {570 None,571 RequestRecovery,572 Recovered,573 Detach,574}575576struct GenerationContributor {577 authorization_revision: u64,578 ingress: CumulativeFeedbackIngress,579 cursor: FeedbackIngressCursor,580 controller: ShadowBitrateController,581 latest_report: Option<MediaFeedbackReportV1>,582 latest_evaluation: Option<AdaptiveBitrateEvaluation>,583 last_received_at: Option<Instant>,584 recovery: ContributorRecoveryState,585 telemetry: Arc<super::session::MediaV3Telemetry>,586 decisions: tokio::sync::watch::Sender<Option<AdaptiveBitrateDecisionV1>>,587 detach: tokio::sync::watch::Sender<bool>,588}589590struct GenerationAggregateState {591 contributors: HashMap<GenerationContributorKey, GenerationContributor>,592 next_decision_id: u64,593 target_kbps: u32,594}595596pub(super) struct GenerationAdaptiveCoordinator {597 generation_id: u64,598 ceiling_kbps: u32,599 state: Mutex<GenerationAggregateState>,600 changed: tokio::sync::Notify,601 recovering_viewers: Arc<AtomicUsize>,602 // Detaching is an action. In shadow mode the controller computes a lower603 // bitrate and never applies it, so a pressured viewer cannot recover and604 // would be detached for failing to do something it was never allowed to do.605 encoder_actuation_available: bool,606}607608impl std::fmt::Debug for GenerationAdaptiveCoordinator {609 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {610 formatter611 .debug_struct("GenerationAdaptiveCoordinator")612 .field("generation_id", &self.generation_id)613 .field("ceiling_kbps", &self.ceiling_kbps)614 .finish_non_exhaustive()615 }616}617618pub(super) struct GenerationFeedbackRegistration {619 coordinator: Arc<GenerationAdaptiveCoordinator>,620 key: GenerationContributorKey,621 authorization_revision: u64,622 pub(super) decisions: tokio::sync::watch::Receiver<Option<AdaptiveBitrateDecisionV1>>,623 pub(super) detach: tokio::sync::watch::Receiver<bool>,624}625626impl GenerationFeedbackRegistration {627 pub(super) fn observe(&self, report: MediaFeedbackReportV1) -> Result<()> {628 report.validate()?;629 let mut state = self630 .coordinator631 .state632 .lock()633 .map_err(|_| anyhow::anyhow!("generation adaptive state poisoned"))?;634 let contributor = state635 .contributors636 .get_mut(&self.key)637 .filter(|contributor| contributor.authorization_revision == self.authorization_revision)638 .context("feedback contributor no longer matches the active viewer generation")?;639 contributor.ingress.observe(report);640 drop(state);641 self.coordinator.changed.notify_one();642 Ok(())643 }644}645646impl Drop for GenerationFeedbackRegistration {647 fn drop(&mut self) {648 self.coordinator649 .unregister(self.key, self.authorization_revision);650 }651}652653impl GenerationAdaptiveCoordinator {654 pub(super) fn start(655 config: HostConfig,656 generation_id: u64,657 encoder_control: Option<EncoderControl>,658 producer_telemetry: Arc<super::session::MediaV3Telemetry>,659 keyframe_requests: tokio::sync::watch::Sender<Option<MediaControlRequestV3>>,660 shutdown: tokio::sync::watch::Receiver<bool>,661 recovering_viewers: Arc<AtomicUsize>,662 ) -> Result<(Arc<Self>, tokio::task::JoinHandle<Result<()>>)> {663 let ceiling_kbps = adaptive_bitrate_ceiling_kbps(&config)?;664 let encoder_actuation_available =665 encoder_actuation_available(&config, encoder_control.as_ref());666 let coordinator = Arc::new(Self {667 generation_id,668 ceiling_kbps,669 state: Mutex::new(GenerationAggregateState {670 contributors: HashMap::with_capacity(crate::config::MAX_VIEWERS.into()),671 next_decision_id: 1,672 target_kbps: ceiling_kbps,673 }),674 changed: tokio::sync::Notify::new(),675 recovering_viewers,676 encoder_actuation_available,677 });678 let task_coordinator = Arc::clone(&coordinator);679 let task = tokio::spawn(async move {680 task_coordinator681 .run(682 config,683 encoder_control,684 producer_telemetry,685 keyframe_requests,686 shutdown,687 )688 .await689 });690 Ok((coordinator, task))691 }692693 pub(super) fn register(694 self: &Arc<Self>,695 remote: EndpointId,696 session_id: u64,697 authorization_revision: u64,698 media_generation_id: u64,699 telemetry: Arc<super::session::MediaV3Telemetry>,700 ) -> Result<GenerationFeedbackRegistration> {701 ensure!(702 media_generation_id == self.generation_id,703 "feedback media generation does not match the shared generation"704 );705 ensure!(706 authorization_revision != 0,707 "feedback authorization revision must be non-zero"708 );709 let key = GenerationContributorKey { remote, session_id };710 let (decisions, decision_rx) = tokio::sync::watch::channel(None);711 let (detach, detach_rx) = tokio::sync::watch::channel(false);712 let mut state = self713 .state714 .lock()715 .map_err(|_| anyhow::anyhow!("generation adaptive state poisoned"))?;716 ensure!(717 state.contributors.len() < usize::from(crate::config::MAX_VIEWERS)718 || state.contributors.contains_key(&key),719 "generation feedback contributor capacity is full"720 );721 state.contributors.insert(722 key,723 GenerationContributor {724 authorization_revision,725 ingress: CumulativeFeedbackIngress::default(),726 cursor: FeedbackIngressCursor::default(),727 controller: ShadowBitrateController::new(self.ceiling_kbps, telemetry.snapshot()),728 latest_report: None,729 latest_evaluation: None,730 last_received_at: None,731 recovery: ContributorRecoveryState::default(),732 telemetry,733 decisions,734 detach,735 },736 );737 self.update_recovering_count(&state);738 drop(state);739 self.changed.notify_one();740 Ok(GenerationFeedbackRegistration {741 coordinator: Arc::clone(self),742 key,743 authorization_revision,744 decisions: decision_rx,745 detach: detach_rx,746 })747 }748749 fn unregister(&self, key: GenerationContributorKey, authorization_revision: u64) {750 let Ok(mut state) = self.state.lock() else {751 return;752 };753 if state754 .contributors755 .get(&key)756 .is_some_and(|contributor| contributor.authorization_revision == authorization_revision)757 {758 state.contributors.remove(&key);759 self.update_recovering_count(&state);760 drop(state);761 self.changed.notify_one();762 }763 }764765 async fn run(766 self: Arc<Self>,767 config: HostConfig,768 encoder_control: Option<EncoderControl>,769 producer_telemetry: Arc<super::session::MediaV3Telemetry>,770 keyframe_requests: tokio::sync::watch::Sender<Option<MediaControlRequestV3>>,771 mut shutdown: tokio::sync::watch::Receiver<bool>,772 ) -> Result<()> {773 let actuation_enabled =774 adaptive_bitrate_actuation_enabled(&config) && encoder_control.is_some();775 let mut resolution_controller = if actuation_enabled {776 encoder_control777 .as_ref()778 .map(motion_resolution_policy_for_encoder)779 .transpose()?780 .flatten()781 } else {782 None783 };784 let initial_resolution = resolution_controller785 .as_ref()786 .map(MotionResolutionPolicy::target);787 let mut actuator = GenerationAdaptiveActuator::new(788 encoder_control,789 actuation_enabled,790 self.ceiling_kbps,791 initial_resolution,792 );793 let mut tick = tokio::time::interval(AGGREGATE_TICK_INTERVAL);794 tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);795 loop {796 tokio::select! {797 biased;798 changed = shutdown.changed() => {799 if changed.is_err() || *shutdown.borrow_and_update() {800 break;801 }802 }803 acknowledgement = actuator.acknowledgements.join_next(), if actuator.pending => {804 actuator.complete(acknowledgement)?;805 actuator.start_queued();806 }807 _ = self.changed.notified() => {}808 _ = tick.tick() => {}809 }810 if let Some(update) = self.evaluate(811 Instant::now(),812 producer_telemetry.snapshot(),813 actuator.bitrate_is_applied(),814 resolution_controller.as_mut(),815 )? {816 if update.request_recovery {817 let request_id = update.decision_id;818 keyframe_requests.send_replace(Some(MediaControlRequestV3::request_keyframe(819 request_id,820 update.last_sequence,821 KeyframeRequestReasonV3::FrontendBackpressure,822 )));823 }824 actuator.submit(update.plan);825 }826 }827 actuator.abort_and_drain().await;828 Ok(())829 }830831 fn evaluate(832 &self,833 now: Instant,834 producer_telemetry: MediaV3TelemetrySnapshot,835 bitrate_applied: bool,836 resolution_controller: Option<&mut MotionResolutionPolicy>,837 ) -> Result<Option<GenerationAggregateUpdate>> {838 let mut state = self839 .state840 .lock()841 .map_err(|_| anyhow::anyhow!("generation adaptive state poisoned"))?;842 let mut changed_keys = Vec::new();843 let mut request_recovery = false;844 for (key, contributor) in &mut state.contributors {845 let Some((report, next_cursor)) =846 contributor.ingress.report_since(contributor.cursor)?847 else {848 continue;849 };850 contributor.cursor = next_cursor;851 // Configured GOP supersession and producer-wide send state are not852 // charged to every viewer as local transport pressure. The shared853 // values remain available to generation diagnostics and actuation.854 let telemetry = contributor.telemetry.snapshot();855 let evaluation = contributor.controller.evaluate(&report, telemetry, now)?;856 match contributor.recovery.observe(&evaluation, now) {857 RecoveryMutation::RequestRecovery => request_recovery = true,858 RecoveryMutation::Detach => {859 if self.encoder_actuation_available {860 contributor.detach.send_replace(true);861 } else {862 warn!(863 generation_id = self.generation_id,864 "viewer stayed below the bitrate floor, but shadow mode never applied \865 a reduction; keeping it attached instead of detaching it for failing \866 to recover"867 );868 }869 }870 RecoveryMutation::None | RecoveryMutation::Recovered => {}871 }872 contributor.latest_report = Some(report);873 contributor.latest_evaluation = Some(evaluation);874 contributor.last_received_at = Some(now);875 changed_keys.push(*key);876 }877 self.update_recovering_count(&state);878879 let active = state880 .contributors881 .values()882 .filter(|contributor| {883 contributor.recovery.contributes()884 && contributor.last_received_at.is_some_and(|received| {885 now.saturating_duration_since(received) <= FEEDBACK_STALE_ARRIVAL_GAP886 })887 && contributor888 .latest_evaluation889 .is_some_and(|evaluation| evaluation.fresh)890 })891 .filter_map(|contributor| contributor.latest_evaluation.zip(contributor.latest_report))892 .collect::<Vec<_>>();893 if changed_keys.is_empty() && active.is_empty() {894 return Ok(None);895 }896 let (target_kbps, all_clean) =897 aggregate_target_and_clean(&active, state.target_kbps, self.ceiling_kbps);898 let previous_target = state.target_kbps;899 if changed_keys.is_empty() && target_kbps == previous_target {900 return Ok(None);901 }902 state.target_kbps = target_kbps;903 let global_state = match target_kbps.cmp(&previous_target) {904 std::cmp::Ordering::Less => AdaptiveBitrateStateV1::Decrease,905 std::cmp::Ordering::Greater => AdaptiveBitrateStateV1::Increase,906 std::cmp::Ordering::Equal => AdaptiveBitrateStateV1::Hold,907 };908 let worst = active909 .iter()910 .max_by_key(|(evaluation, _)| feedback_severity_rank(evaluation.pressure_severity));911 let (aggregate_evaluation, aggregate_report, aggregate_reasons) = match worst {912 Some((evaluation, report)) => {913 let mut aggregate = *evaluation;914 aggregate.clean_recovery_evidence = all_clean;915 (Some(aggregate), Some(*report), evaluation.decision.reasons)916 }917 None => (None, None, AdaptiveBitrateReasonFlagsV1::FEEDBACK_STALE),918 };919 let resolution_decision = resolution_controller.and_then(|policy| {920 aggregate_evaluation921 .zip(aggregate_report)922 .map(|(evaluation, report)| policy.observe_window(&report, &evaluation, now))923 });924 let plan_decision_id = state.next_decision_id;925 let plan_report_id = aggregate_report.map_or(1, |report| report.report_id);926 let plan = AdaptiveCommitPlan {927 decision: AdaptiveBitrateDecisionV1 {928 decision_id: plan_decision_id,929 report_id: plan_report_id,930 target_kbps,931 floor_kbps: ADAPTIVE_BITRATE_FLOOR_KBPS,932 ceiling_kbps: self.ceiling_kbps,933 state: global_state,934 reasons: aggregate_reasons,935 applied: bitrate_applied,936 },937 telemetry: producer_telemetry,938 resolution_decision,939 force_keyframe: request_recovery940 || aggregate_report941 .is_some_and(|report| report.flags.contains(MediaFeedbackFlags::RESYNC_ACTIVE)),942 };943944 for key in changed_keys {945 let Some((evaluation, decisions)) =946 state.contributors.get(&key).and_then(|contributor| {947 contributor948 .latest_evaluation949 .map(|evaluation| (evaluation, contributor.decisions.clone()))950 })951 else {952 continue;953 };954 let decision_id = state.next_decision_id;955 state.next_decision_id = state956 .next_decision_id957 .checked_add(1)958 .context("generation adaptive decision ID exhausted")?;959 let decision = AdaptiveBitrateDecisionV1 {960 decision_id,961 report_id: evaluation.decision.report_id,962 target_kbps,963 floor_kbps: ADAPTIVE_BITRATE_FLOOR_KBPS,964 ceiling_kbps: self.ceiling_kbps,965 state: global_state,966 reasons: evaluation.decision.reasons,967 applied: bitrate_applied,968 };969 decision.validate()?;970 decisions.send_replace(Some(decision));971 }972 plan.decision.validate()?;973 Ok(Some(GenerationAggregateUpdate {974 plan,975 request_recovery,976 decision_id: plan_decision_id,977 last_sequence: aggregate_report.and_then(|report| report.last_sequence),978 }))979 }980981 fn update_recovering_count(&self, state: &GenerationAggregateState) {982 let count = state983 .contributors984 .values()985 .filter(|contributor| {986 !matches!(987 contributor.recovery,988 ContributorRecoveryState::Active { .. }989 )990 })991 .count();992 debug_assert!(count <= usize::from(crate::config::MAX_VIEWERS));993 self.recovering_viewers.store(count, Ordering::Relaxed);994 }995}996997fn feedback_severity_rank(severity: FeedbackSeverity) -> u8 {998 match severity {999 FeedbackSeverity::Clean => 0,1000 FeedbackSeverity::InsufficientEvidence => 1,1001 FeedbackSeverity::Moderate => 2,1002 FeedbackSeverity::Severe => 3,1003 FeedbackSeverity::Stale => 4,1004 }1005}10061007fn aggregate_target_and_clean(1008 active: &[(AdaptiveBitrateEvaluation, MediaFeedbackReportV1)],1009 current_target_kbps: u32,1010 ceiling_kbps: u32,1011) -> (u32, bool) {1012 let target = active1013 .iter()1014 .map(|(evaluation, _)| evaluation.decision.target_kbps)1015 .min()1016 .unwrap_or(current_target_kbps)1017 .clamp(ADAPTIVE_BITRATE_FLOOR_KBPS, ceiling_kbps);1018 let all_clean = !active.is_empty()1019 && active.iter().all(|(evaluation, _)| {1020 evaluation.pressure_severity == FeedbackSeverity::Clean1021 && evaluation.clean_recovery_evidence1022 });1023 (target, all_clean)1024}10251026struct GenerationAggregateUpdate {1027 plan: AdaptiveCommitPlan,1028 request_recovery: bool,1029 decision_id: u64,1030 last_sequence: Option<u64>,1031}103210331034struct ShadowBitrateController {1035 floor_kbps: u32,1036 ceiling_kbps: u32,1037 target_kbps: u32,1038 next_decision_id: u64,1039 last_report_id: Option<u64>,1040 last_report_received_at: Option<Instant>,1041 previous_telemetry: MediaV3TelemetrySnapshot,1042 baseline_rtt_micros: Option<u64>,1043 last_sequence: Option<u64>,1044 moderate_windows: u8,1045 clean_windows: u8,1046 cooldown_windows: u8,1047}104810491050struct AdaptiveBitrateEvaluation {1051 decision: AdaptiveBitrateDecisionV1,1052 pressure_severity: FeedbackSeverity,1053 clean_recovery_evidence: bool,1054 fresh: bool,1055}10561057impl ShadowBitrateController {1058 fn new(configured_ceiling_kbps: u32, telemetry: MediaV3TelemetrySnapshot) -> Self {1059 let ceiling_kbps = quantize_down(configured_ceiling_kbps).max(ADAPTIVE_BITRATE_FLOOR_KBPS);1060 Self {1061 floor_kbps: ADAPTIVE_BITRATE_FLOOR_KBPS,1062 ceiling_kbps,1063 target_kbps: ceiling_kbps,1064 next_decision_id: 1,1065 last_report_id: None,1066 last_report_received_at: None,1067 previous_telemetry: telemetry,1068 baseline_rtt_micros: None,1069 last_sequence: None,1070 moderate_windows: 0,1071 clean_windows: 0,1072 cooldown_windows: 0,1073 }1074 }10751076 fn evaluate(1077 &mut self,1078 report: &MediaFeedbackReportV1,1079 telemetry: MediaV3TelemetrySnapshot,1080 received_at: Instant,1081 ) -> Result<AdaptiveBitrateEvaluation> {1082 ensure!(1083 self.last_report_id1084 .is_none_or(|last| report.report_id > last),1085 "feedback report IDs must increase monotonically"1086 );1087 let (severity, pressure_severity, reasons, stale, trusted_host_pressure) =1088 self.classify(report, telemetry, received_at);1089 let mut state = AdaptiveBitrateStateV1::Hold;10901091 if self.cooldown_windows > 0 {1092 self.cooldown_windows -= 1;1093 }1094 match severity {1095 FeedbackSeverity::Severe => {1096 self.moderate_windows = 0;1097 self.clean_windows = 0;1098 self.cooldown_windows = ADAPTIVE_BITRATE_COOLDOWN_WINDOWS;1099 let next =1100 quantize_down(self.target_kbps.saturating_mul(3) / 4).max(self.floor_kbps);1101 if next < self.target_kbps {1102 self.target_kbps = next;1103 state = AdaptiveBitrateStateV1::Decrease;1104 }1105 }1106 FeedbackSeverity::Moderate => {1107 self.moderate_windows = self.moderate_windows.saturating_add(1);1108 self.clean_windows = 0;1109 if self.moderate_windows >= ADAPTIVE_BITRATE_MODERATE_WINDOWS {1110 self.moderate_windows = 0;1111 self.cooldown_windows = ADAPTIVE_BITRATE_COOLDOWN_WINDOWS;1112 let next =1113 quantize_down(self.target_kbps.saturating_mul(4) / 5).max(self.floor_kbps);1114 if next < self.target_kbps {1115 self.target_kbps = next;1116 state = AdaptiveBitrateStateV1::Decrease;1117 }1118 }1119 }1120 FeedbackSeverity::Clean => {1121 self.moderate_windows = 0;1122 self.clean_windows = self.clean_windows.saturating_add(1);1123 if self.clean_windows >= ADAPTIVE_BITRATE_CLEAN_WINDOWS1124 && self.cooldown_windows == 01125 {1126 self.clean_windows = 0;1127 let maximum_step =1128 (self.target_kbps / 20).clamp(ADAPTIVE_BITRATE_QUANTUM_KBPS, 500);1129 let next = quantize_down(1130 self.target_kbps1131 .saturating_add(maximum_step)1132 .min(self.ceiling_kbps),1133 );1134 if next > self.target_kbps {1135 self.target_kbps = next;1136 state = AdaptiveBitrateStateV1::Increase;1137 }1138 }1139 }1140 FeedbackSeverity::InsufficientEvidence | FeedbackSeverity::Stale => {1141 self.moderate_windows = 0;1142 self.clean_windows = 0;1143 }1144 }11451146 self.target_kbps = self.target_kbps.clamp(self.floor_kbps, self.ceiling_kbps);1147 let decision = AdaptiveBitrateDecisionV1 {1148 decision_id: self.next_decision_id,1149 report_id: report.report_id,1150 target_kbps: self.target_kbps,1151 floor_kbps: self.floor_kbps,1152 ceiling_kbps: self.ceiling_kbps,1153 state,1154 reasons,1155 applied: false,1156 };1157 decision.validate()?;1158 self.next_decision_id = self1159 .next_decision_id1160 .checked_add(1)1161 .context("adaptive bitrate decision ID exhausted")?;1162 self.last_report_id = Some(report.report_id);1163 self.last_report_received_at = Some(received_at);1164 if !stale1165 || !trusted_host_pressure1166 || state == AdaptiveBitrateStateV1::Decrease1167 || self.target_kbps == self.floor_kbps1168 {1169 self.previous_telemetry = telemetry;1170 }1171 if !stale && let Some(sequence) = report.last_sequence {1172 self.last_sequence = Some(1173 self.last_sequence1174 .map_or(sequence, |last| last.max(sequence)),1175 );1176 }1177 Ok(AdaptiveBitrateEvaluation {1178 decision,1179 pressure_severity,1180 clean_recovery_evidence: severity == FeedbackSeverity::Clean,1181 fresh: !stale,1182 })1183 }11841185 1186 fn decide(1187 &mut self,1188 report: &MediaFeedbackReportV1,1189 telemetry: MediaV3TelemetrySnapshot,1190 received_at: Instant,1191 ) -> Result<AdaptiveBitrateDecisionV1> {1192 self.evaluate(report, telemetry, received_at)1193 .map(|evaluation| evaluation.decision)1194 }11951196 fn classify(1197 &mut self,1198 report: &MediaFeedbackReportV1,1199 telemetry: MediaV3TelemetrySnapshot,1200 received_at: Instant,1201 ) -> (1202 FeedbackSeverity,1203 FeedbackSeverity,1204 AdaptiveBitrateReasonFlagsV1,1205 bool,1206 bool,1207 ) {1208 let stale = !(FEEDBACK_FRESH_INTERVAL_MIN_MS..=FEEDBACK_FRESH_INTERVAL_MAX_MS)1209 .contains(&report.interval_ms)1210 || self.last_report_received_at.is_none()1211 || self.last_report_received_at.is_some_and(|last| {1212 let gap = received_at.saturating_duration_since(last);1213 !(FEEDBACK_FRESH_ARRIVAL_GAP_MIN..=FEEDBACK_STALE_ARRIVAL_GAP).contains(&gap)1214 });1215 let mut severity = FeedbackSeverity::Clean;1216 let mut reasons = AdaptiveBitrateReasonFlagsV1::NONE;12171218 let lost_delta = telemetry1219 .selected_path_lost_packets1220 .saturating_sub(self.previous_telemetry.selected_path_lost_packets);1221 let congestion_delta = telemetry1222 .selected_path_congestion_events1223 .saturating_sub(self.previous_telemetry.selected_path_congestion_events);1224 let cancellation_delta = telemetry1225 .scheduler_cancellations1226 .saturating_sub(self.previous_telemetry.scheduler_cancellations);1227 let failure_delta = telemetry1228 .send_failures1229 .saturating_sub(self.previous_telemetry.send_failures);12301231 if failure_delta > 0 || lost_delta > 0 || cancellation_delta >= 2 {1232 add_feedback_signal(1233 &mut severity,1234 &mut reasons,1235 FeedbackSeverity::Severe,1236 AdaptiveBitrateReasonFlagsV1::LOSS_OR_CANCELLATION,1237 );1238 } else if cancellation_delta > 0 || congestion_delta > 0 {1239 add_feedback_signal(1240 &mut severity,1241 &mut reasons,1242 FeedbackSeverity::Moderate,1243 AdaptiveBitrateReasonFlagsV1::LOSS_OR_CANCELLATION,1244 );1245 }1246 if cancellation_delta >= 2 || failure_delta > 0 {1247 add_feedback_signal(1248 &mut severity,1249 &mut reasons,1250 FeedbackSeverity::Severe,1251 AdaptiveBitrateReasonFlagsV1::SENDER_BACKPRESSURE,1252 );1253 } else if cancellation_delta > 0 {1254 add_feedback_signal(1255 &mut severity,1256 &mut reasons,1257 FeedbackSeverity::Moderate,1258 AdaptiveBitrateReasonFlagsV1::SENDER_BACKPRESSURE,1259 );1260 }12611262 let rtt = telemetry.selected_path_rtt_micros;1263 if rtt > 0 {1264 let baseline = self.baseline_rtt_micros.get_or_insert(rtt);1265 if rtt < *baseline {1266 *baseline = rtt;1267 } else if rtt >= baseline.saturating_mul(2).max(80_000) {1268 add_feedback_signal(1269 &mut severity,1270 &mut reasons,1271 FeedbackSeverity::Severe,1272 AdaptiveBitrateReasonFlagsV1::RTT_INFLATION,1273 );1274 } else if rtt >= (baseline.saturating_mul(3) / 2).max(40_000) {1275 add_feedback_signal(1276 &mut severity,1277 &mut reasons,1278 FeedbackSeverity::Moderate,1279 AdaptiveBitrateReasonFlagsV1::RTT_INFLATION,1280 );1281 }1282 }12831284 let trusted_host_pressure = severity != FeedbackSeverity::Clean;1285 if stale {1286 reasons = reasons.union(AdaptiveBitrateReasonFlagsV1::FEEDBACK_STALE);1287 if !trusted_host_pressure {1288 severity = FeedbackSeverity::Stale;1289 }1290 return (severity, severity, reasons, true, trusted_host_pressure);1291 }12921293 let receiver_drops = report1294 .transport_dropped_delta1295 .saturating_add(report.frontend_dropped_delta)1296 .saturating_add(report.decoder_dropped_delta)1297 .saturating_add(report.presenter_dropped_delta);1298 let queue_severe = queue_ratio_at_least(1299 report.frontend_queue_depth,1300 report.frontend_queue_capacity,1301 3,1302 4,1303 ) || queue_ratio_at_least(1304 report.decode_queue_depth,1305 report.decode_queue_capacity,1306 3,1307 4,1308 ) || queue_ratio_at_least(1309 report.presenter_queue_depth,1310 report.presenter_queue_capacity,1311 3,1312 4,1313 );1314 let queue_moderate = queue_ratio_at_least(1315 report.frontend_queue_depth,1316 report.frontend_queue_capacity,1317 1,1318 2,1319 ) || queue_ratio_at_least(1320 report.decode_queue_depth,1321 report.decode_queue_capacity,1322 1,1323 2,1324 ) || queue_ratio_at_least(1325 report.presenter_queue_depth,1326 report.presenter_queue_capacity,1327 1,1328 2,1329 );1330 if report.flags.contains(MediaFeedbackFlags::RESYNC_ACTIVE)1331 || queue_severe1332 || receiver_drops >= 41333 {1334 add_feedback_signal(1335 &mut severity,1336 &mut reasons,1337 FeedbackSeverity::Severe,1338 AdaptiveBitrateReasonFlagsV1::RECEIVER_QUEUE,1339 );1340 } else if queue_moderate || receiver_drops > 0 {1341 add_feedback_signal(1342 &mut severity,1343 &mut reasons,1344 FeedbackSeverity::Moderate,1345 AdaptiveBitrateReasonFlagsV1::RECEIVER_QUEUE,1346 );1347 }1348 if report.decode_queue_depth == report.decode_queue_capacity1349 || report.decoder_dropped_delta > 01350 {1351 add_feedback_signal(1352 &mut severity,1353 &mut reasons,1354 FeedbackSeverity::Severe,1355 AdaptiveBitrateReasonFlagsV1::DECODE_BACKLOG,1356 );1357 } else if report.decode_queue_depth >= 2 {1358 add_feedback_signal(1359 &mut severity,1360 &mut reasons,1361 FeedbackSeverity::Moderate,1362 AdaptiveBitrateReasonFlagsV1::DECODE_BACKLOG,1363 );1364 }13651366 let delivery_p95 = report.transport_delivery_p95_ms.unwrap_or(0);1367 let decode_p95 = report.decode_p95_ms.unwrap_or(0);1368 let presentation_p95 = report.presentation_p95_ms.unwrap_or(0);1369 if delivery_p95 >= 200 || decode_p95 >= 100 || presentation_p95 >= 100 {1370 add_feedback_signal(1371 &mut severity,1372 &mut reasons,1373 FeedbackSeverity::Severe,1374 AdaptiveBitrateReasonFlagsV1::DELIVERY_LATENCY,1375 );1376 } else if delivery_p95 >= 100 || decode_p95 >= 50 || presentation_p95 >= 50 {1377 add_feedback_signal(1378 &mut severity,1379 &mut reasons,1380 FeedbackSeverity::Moderate,1381 AdaptiveBitrateReasonFlagsV1::DELIVERY_LATENCY,1382 );1383 }13841385 let pressure_severity = severity;1386 if severity == FeedbackSeverity::Clean {1387 let sequence_progressed = self1388 .last_sequence1389 .zip(report.last_sequence)1390 .is_some_and(|(last, current)| current > last);1391 let complete_latency = report.transport_delivery_p95_ms.is_some()1392 && report.decode_p95_ms.is_some()1393 && report.presentation_p95_ms.is_some();1394 if sequence_progressed && complete_latency {1395 if self.target_kbps < self.ceiling_kbps {1396 reasons = reasons.union(AdaptiveBitrateReasonFlagsV1::CLEAN_RECOVERY);1397 }1398 } else {1399 severity = FeedbackSeverity::InsufficientEvidence;1400 }1401 }1402 (1403 severity,1404 pressure_severity,1405 reasons,1406 false,1407 trusted_host_pressure,1408 )1409 }1410}14111412fn add_feedback_signal(1413 severity: &mut FeedbackSeverity,1414 reasons: &mut AdaptiveBitrateReasonFlagsV1,1415 signal: FeedbackSeverity,1416 reason: AdaptiveBitrateReasonFlagsV1,1417) {1418 *reasons = reasons.union(reason);1419 if signal == FeedbackSeverity::Severe1420 || (*severity == FeedbackSeverity::Clean && signal == FeedbackSeverity::Moderate)1421 {1422 *severity = signal;1423 }1424}14251426fn quantize_down(kbps: u32) -> u32 {1427 kbps / ADAPTIVE_BITRATE_QUANTUM_KBPS * ADAPTIVE_BITRATE_QUANTUM_KBPS1428}14291430fn queue_ratio_at_least(depth: u8, capacity: u8, numerator: u16, denominator: u16) -> bool {1431 capacity > 01432 && u16::from(depth).saturating_mul(denominator)1433 >= u16::from(capacity).saturating_mul(numerator)1434}143514361437struct AdaptiveCommitPlan {1438 decision: AdaptiveBitrateDecisionV1,1439 telemetry: MediaV3TelemetrySnapshot,1440 resolution_decision: Option<MotionResolutionDecision>,1441 force_keyframe: bool,1442}144314441445struct AdaptiveCommitAcknowledgement {1446 plan: AdaptiveCommitPlan,1447 generation_ended: bool,1448 recovery_acknowledged: Option<bool>,1449 resolution_applied: Option<bool>,1450 applied_bitrate_kbps: Option<u32>,1451 applied_resolution: Option<VideoDimensions>,1452}14531454impl AdaptiveCommitAcknowledgement {1455 fn immediate(plan: AdaptiveCommitPlan) -> Self {1456 Self {1457 plan,1458 generation_ended: false,1459 recovery_acknowledged: None,1460 resolution_applied: None,1461 applied_bitrate_kbps: None,1462 applied_resolution: None,1463 }1464 }1465}14661467/// Bounded adaptive-control actuator.1468///1469/// Feedback evaluation remains synchronous and deterministic, while this1470/// coordinator owns the slower encoder acknowledgements. At most one commit1471/// task and one latest coalesced plan exist. A queued plan carries the latest1472/// desired bitrate and resolution even when its feedback decision is a hold,1473/// so a slow or failed predecessor cannot strand an uncommitted target.1474struct AdaptiveCommitCoordinator {1475 sessions: Arc<SessionRegistry>,1476 remote: EndpointId,1477 session_id: u64,1478 actuation_enabled: bool,1479 desired_bitrate_kbps: u32,1480 requested_bitrate_kbps: u32,1481 committed_bitrate_kbps: u32,1482 desired_resolution: Option<VideoDimensions>,1483 requested_resolution: Option<VideoDimensions>,1484 committed_resolution: Option<VideoDimensions>,1485 pending: bool,1486 queued: Option<AdaptiveCommitPlan>,1487 acknowledgements: tokio::task::JoinSet<AdaptiveCommitAcknowledgement>,1488}148914901491struct GenerationAdaptiveAcknowledgement {1492 bitrate_kbps: Option<u32>,1493 resolution: Option<VideoDimensions>,1494}14951496struct GenerationAdaptiveActuator {1497 control: Option<EncoderControl>,1498 actuation_enabled: bool,1499 desired_bitrate_kbps: u32,1500 requested_bitrate_kbps: u32,1501 committed_bitrate_kbps: u32,1502 desired_resolution: Option<VideoDimensions>,1503 requested_resolution: Option<VideoDimensions>,1504 committed_resolution: Option<VideoDimensions>,1505 pending: bool,1506 queued: Option<AdaptiveCommitPlan>,1507 acknowledgements: tokio::task::JoinSet<Result<GenerationAdaptiveAcknowledgement>>,1508}15091510impl GenerationAdaptiveActuator {1511 fn new(1512 control: Option<EncoderControl>,1513 actuation_enabled: bool,1514 bitrate_kbps: u32,1515 resolution: Option<VideoDimensions>,1516 ) -> Self {1517 Self {1518 control,1519 actuation_enabled,1520 desired_bitrate_kbps: bitrate_kbps,1521 requested_bitrate_kbps: bitrate_kbps,1522 committed_bitrate_kbps: bitrate_kbps,1523 desired_resolution: resolution,1524 requested_resolution: resolution,1525 committed_resolution: resolution,1526 pending: false,1527 queued: None,1528 acknowledgements: tokio::task::JoinSet::new(),1529 }1530 }15311532 fn bitrate_is_applied(&self) -> bool {1533 self.actuation_enabled1534 && !self.pending1535 && self.desired_bitrate_kbps == self.committed_bitrate_kbps1536 }15371538 fn submit(&mut self, mut plan: AdaptiveCommitPlan) {1539 self.desired_bitrate_kbps = plan.decision.target_kbps;1540 if let Some(resolution) = plan.resolution_decision {1541 self.desired_resolution = Some(resolution.target);1542 }1543 if self.pending {1544 if let Some(queued) = self.queued {1545 plan.force_keyframe |= queued.force_keyframe;1546 }1547 self.queued = Some(plan);1548 return;1549 }1550 self.start(plan);1551 }15521553 fn start_queued(&mut self) {1554 if let Some(plan) = self.queued.take() {1555 self.start(plan);1556 }1557 }15581559 fn start(&mut self, plan: AdaptiveCommitPlan) {1560 if !self.actuation_enabled {1561 return;1562 }1563 let Some(control) = self.control.clone() else {1564 return;1565 };1566 let bitrate_target = plan.decision.target_kbps;1567 let bitrate_revision = if bitrate_target != self.requested_bitrate_kbps1568 || bitrate_target != self.committed_bitrate_kbps1569 {1570 match control.request_bitrate_kbps(bitrate_target) {1571 Ok(revision) => {1572 self.requested_bitrate_kbps = bitrate_target;1573 Some(revision)1574 }1575 Err(error) => {1576 warn!(%error, bitrate_target, "generation adaptive bitrate request failed");1577 None1578 }1579 }1580 } else {1581 None1582 };1583 let resolution_target = plan.resolution_decision.map(|decision| decision.target);1584 let resolution_revision = if let Some(target) = resolution_target.filter(|target| {1585 Some(*target) != self.requested_resolution || Some(*target) != self.committed_resolution1586 }) {1587 match control.request_resolution(target.width, target.height) {1588 Ok(revision) => {1589 self.requested_resolution = Some(target);1590 Some((revision, target))1591 }1592 Err(error) => {1593 warn!(%error, width = target.width, height = target.height, "generation adaptive resolution request failed");1594 None1595 }1596 }1597 } else {1598 None1599 };1600 if bitrate_revision.is_none() && resolution_revision.is_none() {1601 return;1602 }1603 self.pending = true;1604 self.acknowledgements.spawn(async move {1605 let bitrate_kbps = if let Some(revision) = bitrate_revision {1606 let status = control1607 .wait_for_bitrate_applied(revision, ENCODER_CONTROL_COMMIT_TIMEOUT)1608 .await?;1609 ensure!(1610 status.applied_bitrate_revision == Some(revision)1611 && status.applied_bitrate_kbps == Some(bitrate_target),1612 "generation bitrate acknowledgement did not match revision {revision} at {bitrate_target} kbps"1613 );1614 Some(bitrate_target)1615 } else {1616 None1617 };1618 let resolution = if let Some((revision, target)) = resolution_revision {1619 let status = control1620 .wait_for_resolution_applied(1621 revision,1622 target.width,1623 target.height,1624 ENCODER_CONTROL_COMMIT_TIMEOUT,1625 )1626 .await?;1627 ensure!(1628 status.applied_resolution_revision == Some(revision)1629 && status.applied_width == Some(target.width)1630 && status.applied_height == Some(target.height),1631 "generation resolution acknowledgement did not match revision {revision} at {}x{}",1632 target.width,1633 target.height1634 );1635 Some(target)1636 } else {1637 None1638 };1639 Ok(GenerationAdaptiveAcknowledgement {1640 bitrate_kbps,1641 resolution,1642 })1643 });1644 }16451646 fn complete(1647 &mut self,1648 result: Option<Result<Result<GenerationAdaptiveAcknowledgement>, tokio::task::JoinError>>,1649 ) -> Result<()> {1650 ensure!(1651 self.pending,1652 "generation adaptive acknowledgement was not pending"1653 );1654 self.pending = false;1655 match result1656 .context("generation adaptive task ended without a result")?1657 .context("generation adaptive task failed")?1658 {1659 Ok(acknowledgement) => {1660 if let Some(bitrate_kbps) = acknowledgement.bitrate_kbps {1661 self.committed_bitrate_kbps = bitrate_kbps;1662 }1663 if let Some(resolution) = acknowledgement.resolution {1664 self.committed_resolution = Some(resolution);1665 }1666 }1667 Err(error) => {1668 warn!(%error, "generation adaptive commit was not applied");1669 }1670 }1671 Ok(())1672 }16731674 async fn abort_and_drain(&mut self) {1675 self.pending = false;1676 self.queued = None;1677 self.acknowledgements.abort_all();1678 while self.acknowledgements.join_next().await.is_some() {}1679 }1680}16811682impl AdaptiveCommitCoordinator {1683 fn new(1684 sessions: Arc<SessionRegistry>,1685 remote: EndpointId,1686 session_id: u64,1687 actuation_enabled: bool,1688 committed_bitrate_kbps: u32,1689 committed_resolution: Option<VideoDimensions>,1690 ) -> Self {1691 Self {1692 sessions,1693 remote,1694 session_id,1695 actuation_enabled,1696 desired_bitrate_kbps: committed_bitrate_kbps,1697 requested_bitrate_kbps: committed_bitrate_kbps,1698 committed_bitrate_kbps,1699 desired_resolution: committed_resolution,1700 requested_resolution: committed_resolution,1701 committed_resolution,1702 pending: false,1703 queued: None,1704 acknowledgements: tokio::task::JoinSet::new(),1705 }1706 }17071708 fn has_pending(&self) -> bool {1709 self.pending1710 }17111712 fn bitrate_is_applied(&self, target_kbps: u32) -> bool {1713 self.actuation_enabled1714 && !self.pending1715 && target_kbps == self.requested_bitrate_kbps1716 && target_kbps == self.committed_bitrate_kbps1717 }17181719 fn submit(&mut self, mut plan: AdaptiveCommitPlan) -> Option<AdaptiveCommitAcknowledgement> {1720 self.desired_bitrate_kbps = plan.decision.target_kbps;1721 if let Some(resolution) = plan.resolution_decision {1722 self.desired_resolution = Some(resolution.target);1723 }1724 if self.pending {1725 if let Some(queued) = self.queued {1726 // The latest decision replaces older diagnostics, but a1727 // recovery barrier requested by any coalesced desired update1728 // must survive until that desired state is actuated.1729 plan.force_keyframe |= queued.force_keyframe;1730 }1731 self.queued = Some(plan);1732 return None;1733 }1734 if let Some(queued) = self.queued.take() {1735 // A completion can deliberately defer this slot while an unread1736 // feedback generation is waiting. The newly evaluated plan is the1737 // actual latest state, but it must inherit any recovery barrier1738 // carried by the superseded slot.1739 plan.force_keyframe |= queued.force_keyframe;1740 }1741 self.start(plan)1742 }17431744 fn start_queued(&mut self) -> Option<AdaptiveCommitAcknowledgement> {1745 let plan = self.queued.take()?;1746 self.start(plan)1747 }17481749 fn start(&mut self, plan: AdaptiveCommitPlan) -> Option<AdaptiveCommitAcknowledgement> {1750 if !self.actuation_enabled {1751 return Some(AdaptiveCommitAcknowledgement::immediate(plan));1752 }17531754 let desired_resolution = plan.resolution_decision.map(|change| change.target);1755 debug_assert_eq!(plan.decision.target_kbps, self.desired_bitrate_kbps);1756 if desired_resolution.is_some() {1757 debug_assert_eq!(desired_resolution, self.desired_resolution);1758 }17591760 // A proposal mutates EncoderControl's persistent desired watch before1761 // its acknowledgement arrives. Reconcile against both the last1762 // requested and last committed values: after a timeout, returning to1763 // the old committed target must overwrite the still-pending stale1764 // request rather than allowing it to apply later.1765 let reconciles_resolution = desired_resolution.is_some()1766 && (desired_resolution != self.requested_resolution1767 || desired_resolution != self.committed_resolution);1768 let resolution_proposal = if reconciles_resolution {1769 let target = desired_resolution.expect("resolution target was checked as present");1770 match self1771 .sessions1772 .propose_resolution_update(self.remote, self.session_id, target)1773 {1774 Ok(Some(proposal)) => {1775 self.requested_resolution = Some(target);1776 Some(proposal)1777 }1778 Ok(None) => {1779 warn!(1780 remote = %self.remote,1781 session_id = self.session_id,1782 "active adaptive session lost encoder control; retaining committed resolution"1783 );1784 None1785 }1786 Err(error) => {1787 warn!(1788 remote = %self.remote,1789 session_id = self.session_id,1790 %error,1791 "adaptive resolution proposal failed; retaining committed resolution"1792 );1793 None1794 }1795 }1796 } else {1797 None1798 };17991800 let reconciles_bitrate = plan.decision.target_kbps != self.requested_bitrate_kbps1801 || plan.decision.target_kbps != self.committed_bitrate_kbps;1802 let force_keyframe = plan.force_keyframe && resolution_proposal.is_none();1803 let bitrate_proposal = if reconciles_bitrate || force_keyframe {1804 match self.sessions.propose_adaptive_encoder_update(1805 self.remote,1806 self.session_id,1807 plan.decision.target_kbps,1808 force_keyframe,1809 ) {1810 Ok(Some(proposal)) => {1811 self.requested_bitrate_kbps = plan.decision.target_kbps;1812 Some(proposal)1813 }1814 Ok(None) => {1815 warn!(1816 remote = %self.remote,1817 session_id = self.session_id,1818 "active adaptive session lost encoder control; retaining committed bitrate"1819 );1820 None1821 }1822 Err(error) => {1823 warn!(1824 remote = %self.remote,1825 session_id = self.session_id,1826 %error,1827 "adaptive encoder proposal failed; retaining committed bitrate"1828 );1829 None1830 }1831 }1832 } else {1833 None1834 };18351836 if bitrate_proposal.is_none() && resolution_proposal.is_none() {1837 return Some(AdaptiveCommitAcknowledgement::immediate(plan));1838 }18391840 self.pending = true;1841 let sessions = Arc::clone(&self.sessions);1842 let remote = self.remote;1843 let session_id = self.session_id;1844 self.acknowledgements.spawn(async move {1845 run_adaptive_commit(1846 sessions,1847 remote,1848 session_id,1849 plan,1850 bitrate_proposal,1851 resolution_proposal,1852 )1853 .await1854 });1855 None1856 }18571858 fn complete(1859 &mut self,1860 result: Option<std::result::Result<AdaptiveCommitAcknowledgement, tokio::task::JoinError>>,1861 ) -> Result<AdaptiveCommitAcknowledgement> {1862 ensure!(1863 self.pending,1864 "adaptive commit completed without pending work"1865 );1866 self.pending = false;1867 let acknowledgement = result1868 .context("adaptive commit task ended without a result")?1869 .context("adaptive commit task failed")?;1870 if let Some(applied) = acknowledgement.applied_bitrate_kbps {1871 self.committed_bitrate_kbps = applied;1872 }1873 if let Some(applied) = acknowledgement.applied_resolution {1874 self.committed_resolution = Some(applied);1875 }1876 Ok(acknowledgement)1877 }18781879 async fn abort_and_drain(&mut self) {1880 self.pending = false;1881 self.queued = None;1882 self.acknowledgements.abort_all();1883 while self.acknowledgements.join_next().await.is_some() {}1884 }1885}18861887pub(super) async fn serve_media_feedback(1888 connection: Connection,1889 config: &HostConfig,1890 sessions: &Arc<SessionRegistry>,1891 generations: &Arc<MediaGenerationManager>,1892 authorization: &AuthorizationPolicy,1893 input_operations: &Arc<InputOperations>,1894) -> Result<()> {1895 let remote = connection.remote_id();1896 let handshake_permit = sessions1897 .pending_handshakes1898 .try_acquire()1899 .context("too many pending handshakes")?;1900 let (mut send, mut recv) = tokio::time::timeout(HANDSHAKE_TIMEOUT, connection.accept_bi())1901 .await1902 .context("timed out accepting media feedback stream")?1903 .context("accepting media feedback stream")?;1904 let hello = receive_hello(&mut recv, Capability::VideoH264).await?;1905 drop(handshake_permit);1906 debug!(%remote, agent = %hello.agent, "media feedback hello received");1907 ensure!(1908 hello.invitation.is_none(),1909 "invitations are accepted only on the first media handshake"1910 );19111912 let grants = match authorization.authorize_or_redeem(remote, None, unix_timestamp_now()?) {1913 Ok(grants) => grants,1914 Err(error) => {1915 send_rejection(&mut send, "Portal peer is not authorized").await?;1916 return Err(error.context("authorizing media feedback peer"));1917 }1918 };1919 ensure!(1920 grants.contains(InvitationGrants::VIEW),1921 "authorized feedback peer lacks view permission"1922 );1923 let ceiling_kbps = adaptive_bitrate_ceiling_kbps(config)?;1924 let lease = match sessions.claim_feedback(remote, hello.nonce) {1925 Ok(lease) => lease,1926 Err(error) => {1927 send_rejection(&mut send, error.to_string()).await?;1928 return Err(error);1929 }1930 };1931 let encoder_actuation_available =1932 encoder_actuation_available(config, lease.encoder_control.as_ref());1933 // Shadow mode computes every decision and applies none. Say why, or a host1934 // silently never adapts and nothing in the logs explains it.1935 if !encoder_actuation_available {1936 let reason = if !adaptive_bitrate_actuation_enabled(config) {1937 match config.gamescope_pipewire.as_ref() {1938 Some(gamescope)1939 if gamescope.encoder_backend != GamescopeEncoderBackend::InProcessGstreamer =>1940 {1941 "gamescope_pipewire.encoder_backend is not in-process-gstreamer"1942 }1943 Some(gamescope) if gamescope.rate_control != VaapiRateControl::Cbr => {1944 "gamescope_pipewire.rate_control is not cbr"1945 }1946 Some(_) => "adaptive bitrate actuation is unavailable for this configuration",1947 None => "the configured source is not gamescope-pipewire",1948 }1949 } else {1950 "this session has no encoder control handle"1951 };1952 warn!(1953 %remote,1954 session_id = lease.session_id,1955 reason,1956 "adaptive bitrate and resolution run in shadow: decisions are computed but never applied"1957 );1958 }19591960 write_host_hello(1961 &mut send,1962 &HostHello::accepted(1963 lease.session_id,1964 negotiated_capabilities(&hello, MEDIA_CAPABILITIES),1965 ),1966 )1967 .await?;1968 if lease.scope == FeedbackScope::SharedGeneration {1969 let adaptive = generations.adaptive(lease.media_generation_id).await?;1970 let registration = adaptive.register(1971 remote,1972 lease.session_id,1973 lease.authorization_revision,1974 lease.media_generation_id,1975 Arc::clone(&lease.telemetry),1976 )?;1977 info!(1978 %remote,1979 session_id = lease.session_id,1980 media_generation_id = lease.media_generation_id,1981 authorization_revision = lease.authorization_revision,1982 ceiling_kbps,1983 mode = if encoder_actuation_available { "aggregate-active" } else { "aggregate-shadow" },1984 "media feedback contributor attached to shared generation"1985 );1986 let result = run_generation_feedback_session(1987 &connection,1988 &mut send,1989 recv,1990 remote,1991 sessions,1992 input_operations,1993 &lease,1994 registration,1995 )1996 .await;1997 let _ = send.finish();1998 drop(lease);1999 info!(%remote, "shared-generation media feedback contributor released");2000 return result;2001 }2002 info!(2003 %remote,2004 session_id = lease.session_id,2005 authorization_revision = lease.authorization_revision,2006 ceiling_kbps,2007 applied = false,2008 mode = if encoder_actuation_available { "active" } else { "shadow" },2009 "media feedback client accepted for bitrate control"2010 );20112012 let (feedback_sender, mut feedback_receiver) =2013 tokio::sync::watch::channel(CumulativeFeedbackIngress::default());2014 let reader_task = tokio::spawn(forward_media_feedback_reports(recv, feedback_sender));2015 let connection_closed = async {2016 let closed = connection.closed().await;2017 debug!(%remote, ?closed, "media feedback connection closed");2018 };2019 let result = run_media_feedback_loop(2020 &mut send,2021 connection_closed,2022 remote,2023 sessions,2024 &lease,2025 encoder_actuation_available,2026 ceiling_kbps,2027 &mut feedback_receiver,2028 reader_task,2029 )2030 .await;2031 let _ = send.finish();2032 drop(lease);2033 info!(%remote, "media feedback client released");2034 result2035}203620372038async fn run_generation_feedback_session<W>(2039 connection: &Connection,2040 send: &mut W,2041 recv: iroh::endpoint::RecvStream,2042 remote: EndpointId,2043 sessions: &Arc<SessionRegistry>,2044 input_operations: &Arc<InputOperations>,2045 lease: &FeedbackLease,2046 mut registration: GenerationFeedbackRegistration,2047) -> Result<()>2048where2049 W: tokio::io::AsyncWrite + Unpin,2050{2051 let (feedback_sender, mut feedback_receiver) =2052 tokio::sync::watch::channel(CumulativeFeedbackIngress::default());2053 let mut reader_task = tokio::spawn(forward_media_feedback_reports(recv, feedback_sender));2054 let mut feedback_cursor = FeedbackIngressCursor::default();2055 let result = loop {2056 let session_changed = sessions.session_changed.notified();2057 tokio::pin!(session_changed);2058 session_changed.as_mut().enable();2059 if !sessions.is_active(remote, lease.session_id) {2060 break Ok(());2061 }2062 tokio::select! {2063 reader_result = &mut reader_task => {2064 break reader_result2065 .context("shared-generation feedback reader task failed")?2066 .context("reading shared-generation feedback reports");2067 }2068 changed = feedback_receiver.changed() => {2069 if changed.is_err() {2070 break Ok(());2071 }2072 let ingress = *feedback_receiver.borrow_and_update();2073 let Some((report, next_cursor)) = ingress.report_since(feedback_cursor)? else {2074 continue;2075 };2076 feedback_cursor = next_cursor;2077 registration.observe(report)?;2078 }2079 changed = registration.decisions.changed() => {2080 changed.context("shared-generation adaptive decision publisher stopped")?;2081 let Some(decision) = *registration.decisions.borrow_and_update() else {2082 continue;2083 };2084 tokio::time::timeout(2085 FEEDBACK_WRITE_TIMEOUT,2086 write_adaptive_bitrate_decision_v1(send, &decision),2087 )2088 .await2089 .context("timed out writing shared-generation adaptive decision")??;2090 }2091 changed = registration.detach.changed() => {2092 changed.context("shared-generation recovery state publisher stopped")?;2093 if !*registration.detach.borrow_and_update() {2094 continue;2095 }2096 if let Some(transition) = sessions.invalidate_v2_focus(2097 remote,2098 lease.session_id,2099 FocusTransitionReasonV2::Disconnected,2100 )? {2101 input_operations.neutralize_focus_transition(2102 sessions,2103 transition.transition_id,2104 FocusTransitionReasonV2::Disconnected,2105 )?;2106 }2107 sessions.disconnect_v2_viewer(remote, lease.session_id)?;2108 info!(2109 %remote,2110 session_id = lease.session_id,2111 media_generation_id = lease.media_generation_id,2112 "detached persistently below-floor viewer after bounded isolated recovery"2113 );2114 break Ok(());2115 }2116 _ = &mut session_changed => {}2117 reason = connection.closed() => {2118 debug!(%remote, ?reason, "shared-generation feedback connection closed");2119 break Ok(());2120 }2121 }2122 };2123 if !reader_task.is_finished() {2124 reader_task.abort();2125 let _ = reader_task.await;2126 }2127 result2128}212921302131async fn run_media_feedback_loop<W, C>(2132 send: &mut W,2133 connection_closed: C,2134 remote: EndpointId,2135 sessions: &Arc<SessionRegistry>,2136 lease: &FeedbackLease,2137 encoder_actuation_available: bool,2138 ceiling_kbps: u32,2139 feedback_receiver: &mut tokio::sync::watch::Receiver<CumulativeFeedbackIngress>,2140 mut reader_task: tokio::task::JoinHandle<Result<()>>,2141) -> Result<()>2142where2143 W: tokio::io::AsyncWrite + Unpin,2144 C: Future<Output = ()>,2145{2146 let mut reader_finished = false;2147 let mut receiver_open = true;2148 let mut controller = ShadowBitrateController::new(ceiling_kbps, lease.telemetry.snapshot());2149 let mut resolution_controller = if encoder_actuation_available {2150 lease2151 .encoder_control2152 .as_ref()2153 .map(motion_resolution_policy_for_encoder)2154 .transpose()?2155 .flatten()2156 } else {2157 None2158 };2159 let initial_resolution = resolution_controller2160 .as_ref()2161 .map(MotionResolutionPolicy::target);2162 let mut commit_coordinator = AdaptiveCommitCoordinator::new(2163 Arc::clone(sessions),2164 remote,2165 lease.session_id,2166 encoder_actuation_available,2167 ceiling_kbps,2168 initial_resolution,2169 );2170 let mut feedback_cursor = FeedbackIngressCursor::default();2171 tokio::pin!(connection_closed);21722173 // Keep cleanup on every error path: `?` below exits this async block, not2174 // the outer function that owns the reader and actuator tasks.2175 let result: Result<()> = async {2176 loop {2177 let session_changed = sessions.session_changed.notified();2178 tokio::pin!(session_changed);2179 session_changed.as_mut().enable();2180 if !sessions.is_active(remote, lease.session_id) {2181 break Ok(());2182 }2183 tokio::select! {2184 biased;2185 reader_result = &mut reader_task, if !reader_finished => {2186 reader_finished = true;2187 match reader_result {2188 Ok(Ok(())) => {2189 debug!(%remote, "media feedback report stream closed");2190 }2191 Ok(Err(error)) => {2192 break Err(error).context("reading media feedback reports");2193 }2194 Err(error) => {2195 break Err(error).context("media feedback reader task failed");2196 }2197 }2198 }2199 commit_result = commit_coordinator.acknowledgements.join_next(),2200 if commit_coordinator.has_pending() =>2201 {2202 let acknowledgement = commit_coordinator.complete(commit_result)?;2203 if acknowledgement.generation_ended {2204 break Ok(());2205 }2206 log_adaptive_acknowledgement(2207 remote,2208 lease.session_id,2209 encoder_actuation_available,2210 &commit_coordinator,2211 &acknowledgement,2212 );2213 // If a newer report is already waiting, let its evaluation2214 // replace the queued actuator slot before starting more2215 // work. This avoids spending another full timeout on a2216 // plan that was stale at completion time.2217 if !matches!(feedback_receiver.has_changed(), Ok(true))2218 && let Some(immediate) = commit_coordinator.start_queued()2219 {2220 log_adaptive_acknowledgement(2221 remote,2222 lease.session_id,2223 encoder_actuation_available,2224 &commit_coordinator,2225 &immediate,2226 );2227 }2228 }2229 changed = feedback_receiver.changed(), if receiver_open => {2230 if changed.is_err() {2231 receiver_open = false;2232 if reader_finished {2233 break Ok(());2234 }2235 continue;2236 }2237 let ingress = *feedback_receiver.borrow_and_update();2238 let Some((report, next_cursor)) = ingress.report_since(feedback_cursor)? else {2239 continue;2240 };2241 // This branch now owns the complete aggregate. Advance the2242 // ingress baseline exactly once before evaluation; reports2243 // arriving during an encoder wait accumulate beyond it.2244 feedback_cursor = next_cursor;2245 let telemetry = lease.telemetry.snapshot();2246 let observed_at = Instant::now();2247 let mut candidate = controller.clone();2248 let evaluation = candidate.evaluate(&report, telemetry, observed_at)?;2249 let mut decision = evaluation.decision;2250 decision.applied =2251 commit_coordinator.bitrate_is_applied(decision.target_kbps);2252 let mut resolution_candidate = resolution_controller.clone();2253 let resolution_decision = resolution_candidate2254 .as_mut()2255 .map(|policy| policy.observe_window(&report, &evaluation, observed_at));2256 let force_keyframe =2257 adaptive_recovery_keyframe_required(&report, &decision);2258 let plan = AdaptiveCommitPlan {2259 decision,2260 telemetry,2261 resolution_decision,2262 force_keyframe,2263 };22642265 // One fixed-width decision is written for every evaluated2266 // report before actuator work is submitted. Encoder2267 // acknowledgements are correlated in bounded host2268 // diagnostics and never gate the Portal-facing cadence.2269 write_adaptive_decision(2270 send,2271 remote,2272 lease.session_id,2273 encoder_actuation_available,2274 &plan,2275 )2276 .await?;22772278 // Evaluation state advances at feedback cadence. The2279 // actuator independently retains the exact requested,2280 // committed, and latest desired targets.2281 controller = candidate;2282 resolution_controller = resolution_candidate;2283 if let Some(immediate) = commit_coordinator.submit(plan) {2284 log_adaptive_acknowledgement(2285 remote,2286 lease.session_id,2287 encoder_actuation_available,2288 &commit_coordinator,2289 &immediate,2290 );2291 }2292 }2293 _ = &mut session_changed => {2294 if !sessions.is_active(remote, lease.session_id) {2295 break Ok(());2296 }2297 }2298 _ = &mut connection_closed => {2299 break Ok(());2300 }2301 }2302 if reader_finished2303 && !receiver_open2304 && !commit_coordinator.has_pending()2305 && commit_coordinator.queued.is_none()2306 {2307 break Ok(());2308 }2309 }2310 }2311 .await;23122313 if !reader_finished {2314 reader_task.abort();2315 let _ = reader_task.await;2316 }2317 commit_coordinator.abort_and_drain().await;2318 result2319}23202321async fn write_adaptive_decision<W>(2322 send: &mut W,2323 remote: EndpointId,2324 session_id: u64,2325 encoder_actuation_available: bool,2326 plan: &AdaptiveCommitPlan,2327) -> Result<()>2328where2329 W: tokio::io::AsyncWrite + Unpin,2330{2331 let decision = plan.decision;2332 let telemetry = plan.telemetry;2333 let resolution_decision = plan.resolution_decision;2334 tokio::time::timeout(2335 FEEDBACK_WRITE_TIMEOUT,2336 write_adaptive_bitrate_decision_v1(send, &decision),2337 )2338 .await2339 .context("timed out writing adaptive bitrate decision")??;2340 debug!(2341 %remote,2342 session_id,2343 report_id = decision.report_id,2344 decision_id = decision.decision_id,2345 target_kbps = decision.target_kbps,2346 ?decision.state,2347 reasons = decision.reasons.bits(),2348 applied = decision.applied,2349 resolution_width = resolution_decision.map(|change| change.target.width),2350 resolution_height = resolution_decision.map(|change| change.target.height),2351 resolution_trigger = ?resolution_decision.map(|change| change.trigger),2352 path_rtt_micros = telemetry.selected_path_rtt_micros,2353 path_lost_packets = telemetry.selected_path_lost_packets,2354 path_congestion_events = telemetry.selected_path_congestion_events,2355 scheduler_cancellations = telemetry.scheduler_cancellations,2356 send_failures = telemetry.send_failures,2357 mode = if encoder_actuation_available { "active" } else { "shadow" },2358 "adaptive bitrate decision"2359 );2360 Ok(())2361}23622363fn log_adaptive_acknowledgement(2364 remote: EndpointId,2365 session_id: u64,2366 encoder_actuation_available: bool,2367 coordinator: &AdaptiveCommitCoordinator,2368 acknowledgement: &AdaptiveCommitAcknowledgement,2369) {2370 let decision = acknowledgement.plan.decision;2371 let resolution_decision = acknowledgement.plan.resolution_decision;2372 debug!(2373 %remote,2374 session_id,2375 report_id = decision.report_id,2376 decision_id = decision.decision_id,2377 target_kbps = decision.target_kbps,2378 bitrate_applied = acknowledgement.applied_bitrate_kbps.is_some(),2379 applied_bitrate_kbps = ?acknowledgement.applied_bitrate_kbps,2380 recovery_acknowledged = ?acknowledgement.recovery_acknowledged,2381 resolution_width = resolution_decision.map(|change| change.target.width),2382 resolution_height = resolution_decision.map(|change| change.target.height),2383 resolution_applied = ?acknowledgement.resolution_applied,2384 applied_resolution = ?acknowledgement.applied_resolution,2385 desired_bitrate_kbps = coordinator.desired_bitrate_kbps,2386 requested_bitrate_kbps = coordinator.requested_bitrate_kbps,2387 committed_bitrate_kbps = coordinator.committed_bitrate_kbps,2388 desired_resolution = ?coordinator.desired_resolution,2389 requested_resolution = ?coordinator.requested_resolution,2390 committed_resolution = ?coordinator.committed_resolution,2391 mode = if encoder_actuation_available { "active" } else { "shadow" },2392 "adaptive actuator acknowledgement"2393 );2394}23952396async fn run_adaptive_commit(2397 sessions: Arc<SessionRegistry>,2398 remote: EndpointId,2399 session_id: u64,2400 plan: AdaptiveCommitPlan,2401 bitrate_proposal: Option<AdaptiveEncoderProposal>,2402 resolution_proposal: Option<ResolutionEncoderProposal>,2403) -> AdaptiveCommitAcknowledgement {2404 let mut acknowledgement = AdaptiveCommitAcknowledgement::immediate(plan);2405 if let Some(proposal) = bitrate_proposal {2406 match commit_adaptive_encoder_proposal(&sessions, remote, session_id, &proposal).await {2407 AdaptiveEncoderCommit::GenerationEnded => {2408 acknowledgement.generation_ended = true;2409 return acknowledgement;2410 }2411 AdaptiveEncoderCommit::NotApplied(error) => {2412 warn!(2413 %remote,2414 session_id,2415 %error,2416 "adaptive encoder application failed; retaining committed bitrate"2417 );2418 }2419 AdaptiveEncoderCommit::Applied => {2420 acknowledgement.plan.decision.applied = true;2421 acknowledgement.applied_bitrate_kbps = Some(proposal.target_kbps);2422 if proposal.force_keyframe_revision.is_some() {2423 match await_adaptive_recovery_keyframe(&sessions, remote, session_id, &proposal)2424 .await2425 {2426 None => {2427 acknowledgement.generation_ended = true;2428 return acknowledgement;2429 }2430 Some(Ok(())) => acknowledgement.recovery_acknowledged = Some(true),2431 Some(Err(error)) => {2432 acknowledgement.recovery_acknowledged = Some(false);2433 warn!(2434 %remote,2435 session_id,2436 %error,2437 "adaptive bitrate applied but forced-IDR recovery was not acknowledged"2438 );2439 }2440 }2441 }2442 }2443 }2444 }24452446 if let Some(proposal) = resolution_proposal {2447 match commit_resolution_encoder_proposal(&sessions, remote, session_id, &proposal).await {2448 AdaptiveEncoderCommit::GenerationEnded => {2449 acknowledgement.generation_ended = true;2450 return acknowledgement;2451 }2452 AdaptiveEncoderCommit::NotApplied(error) => {2453 acknowledgement.resolution_applied = Some(false);2454 warn!(2455 %remote,2456 session_id,2457 target_width = proposal.target.width,2458 target_height = proposal.target.height,2459 %error,2460 "adaptive resolution application failed; retaining committed resolution"2461 );2462 }2463 AdaptiveEncoderCommit::Applied => {2464 acknowledgement.resolution_applied = Some(true);2465 acknowledgement.applied_resolution = Some(proposal.target);2466 }2467 }2468 }2469 acknowledgement2470}24712472async fn forward_media_feedback_reports<R>(2473 mut reader: R,2474 sender: tokio::sync::watch::Sender<CumulativeFeedbackIngress>,2475) -> Result<()>2476where2477 R: tokio::io::AsyncRead + Unpin,2478{2479 let mut last_report_id = None;2480 let mut next_read_at = tokio::time::Instant::now();2481 loop {2482 tokio::time::sleep_until(next_read_at).await;2483 let Some(report) = read_media_feedback_report_v1(&mut reader).await? else {2484 return Ok(());2485 };2486 ensure!(2487 last_report_id.is_none_or(|last| report.report_id > last),2488 "feedback report IDs must increase monotonically"2489 );2490 last_report_id = Some(report.report_id);2491 sender.send_modify(|ingress| ingress.observe(report));2492 if sender.is_closed() {2493 return Ok(());2494 }2495 next_read_at = tokio::time::Instant::now() + FEEDBACK_MIN_READ_INTERVAL;2496 }2497}24982499fn adaptive_bitrate_ceiling_kbps(config: &HostConfig) -> Result<u32> {2500 let configured = match config.source {2501 VideoSource::TestPattern => SHADOW_BITRATE_CEILING_KBPS,2502 VideoSource::GamescopePipewire => {2503 let gamescope = config2504 .gamescope_pipewire2505 .as_ref()2506 .context("Gamescope feedback requires gamescope_pipewire configuration")?;2507 match gamescope.rate_control {2508 VaapiRateControl::Cbr => gamescope2509 .bitrate_kbps2510 .context("adaptive bitrate shadow mode requires a Gamescope CBR bitrate")?,2511 VaapiRateControl::Cqp => SHADOW_BITRATE_CEILING_KBPS,2512 }2513 }2514 };2515 let ceiling = quantize_down(configured);2516 ensure!(2517 ceiling >= ADAPTIVE_BITRATE_FLOOR_KBPS,2518 "adaptive bitrate ceiling is below the 1000 kbps floor"2519 );2520 Ok(ceiling)2521}25222523/// Whether adaptive decisions can actually be applied to the encoder. When this2524/// is false the controller runs in shadow: it computes every decision and2525/// applies none, so a pressured viewer has no path back and must not be2526/// detached for failing to take it.2527fn encoder_actuation_available(2528 config: &HostConfig,2529 encoder_control: Option<&EncoderControl>,2530) -> bool {2531 adaptive_bitrate_actuation_enabled(config) && encoder_control.is_some()2532}25332534fn adaptive_bitrate_actuation_enabled(config: &HostConfig) -> bool {2535 config.source == VideoSource::GamescopePipewire2536 && config.gamescope_pipewire.as_ref().is_some_and(|gamescope| {2537 gamescope.encoder_backend == GamescopeEncoderBackend::InProcessGstreamer2538 && gamescope.rate_control == VaapiRateControl::Cbr2539 })2540}25412542fn adaptive_recovery_keyframe_required(2543 report: &MediaFeedbackReportV1,2544 decision: &AdaptiveBitrateDecisionV1,2545) -> bool {2546 decision.state != AdaptiveBitrateStateV1::Hold2547 && report.flags.contains(MediaFeedbackFlags::RESYNC_ACTIVE)2548}254925502551enum AdaptiveEncoderCommit {2552 GenerationEnded,2553 NotApplied(anyhow::Error),2554 Applied,2555}25562557async fn await_while_session_active<F, T>(2558 sessions: &SessionRegistry,2559 remote: EndpointId,2560 session_id: u64,2561 future: F,2562) -> Option<T>2563where2564 F: Future<Output = T>,2565{2566 tokio::pin!(future);2567 loop {2568 let session_changed = sessions.session_changed.notified();2569 tokio::pin!(session_changed);2570 session_changed.as_mut().enable();2571 if !sessions.is_active(remote, session_id) {2572 return None;2573 }2574 tokio::select! {2575 result = &mut future => return sessions.is_active(remote, session_id).then_some(result),2576 _ = &mut session_changed => {}2577 }2578 }2579}25802581async fn commit_adaptive_encoder_proposal(2582 sessions: &SessionRegistry,2583 remote: EndpointId,2584 session_id: u64,2585 proposal: &AdaptiveEncoderProposal,2586) -> AdaptiveEncoderCommit {2587 let bitrate_result = await_while_session_active(2588 sessions,2589 remote,2590 session_id,2591 proposal2592 .control2593 .wait_for_bitrate_applied(proposal.bitrate_revision, ENCODER_CONTROL_COMMIT_TIMEOUT),2594 )2595 .await;2596 let Some(bitrate_result) = bitrate_result else {2597 return AdaptiveEncoderCommit::GenerationEnded;2598 };2599 let status = match bitrate_result {2600 Ok(status) => status,2601 Err(error) => return AdaptiveEncoderCommit::NotApplied(error),2602 };2603 if status.applied_bitrate_revision != Some(proposal.bitrate_revision)2604 || status.applied_bitrate_kbps != Some(proposal.target_kbps)2605 {2606 return AdaptiveEncoderCommit::NotApplied(anyhow::anyhow!(2607 "encoder bitrate commit mismatch: revision {:?}, readback {:?}, expected revision {} at {} kbps",2608 status.applied_bitrate_revision,2609 status.applied_bitrate_kbps,2610 proposal.bitrate_revision,2611 proposal.target_kbps2612 ));2613 }26142615 AdaptiveEncoderCommit::Applied2616}26172618async fn commit_resolution_encoder_proposal(2619 sessions: &SessionRegistry,2620 remote: EndpointId,2621 session_id: u64,2622 proposal: &ResolutionEncoderProposal,2623) -> AdaptiveEncoderCommit {2624 let result = await_while_session_active(2625 sessions,2626 remote,2627 session_id,2628 proposal.control.wait_for_resolution_applied(2629 proposal.revision,2630 proposal.target.width,2631 proposal.target.height,2632 ENCODER_CONTROL_COMMIT_TIMEOUT,2633 ),2634 )2635 .await;2636 let Some(result) = result else {2637 return AdaptiveEncoderCommit::GenerationEnded;2638 };2639 match result {2640 Ok(status)2641 if status.applied_resolution_revision == Some(proposal.revision)2642 && status.applied_width == Some(proposal.target.width)2643 && status.applied_height == Some(proposal.target.height) =>2644 {2645 AdaptiveEncoderCommit::Applied2646 }2647 Ok(status) => AdaptiveEncoderCommit::NotApplied(anyhow::anyhow!(2648 "encoder resolution commit mismatch: revision {:?}, dimensions {:?}x{:?}, expected revision {} at {}x{}",2649 status.applied_resolution_revision,2650 status.applied_width,2651 status.applied_height,2652 proposal.revision,2653 proposal.target.width,2654 proposal.target.height2655 )),2656 Err(error) => AdaptiveEncoderCommit::NotApplied(error),2657 }2658}26592660async fn await_adaptive_recovery_keyframe(2661 sessions: &SessionRegistry,2662 remote: EndpointId,2663 session_id: u64,2664 proposal: &AdaptiveEncoderProposal,2665) -> Option<Result<()>> {2666 let force_keyframe_revision = proposal.force_keyframe_revision?;2667 await_while_session_active(2668 sessions,2669 remote,2670 session_id,2671 proposal.control.wait_for_force_keyframe_acknowledged(2672 force_keyframe_revision,2673 ENCODER_CONTROL_COMMIT_TIMEOUT,2674 ),2675 )2676 .await2677 .map(|result| result.map(|_| ()))2678}267926802681mod tests {2682 use super::super::{endpoint, moq_test_config};2683 use super::*;26842685 fn clean_feedback(report_id: u64) -> MediaFeedbackReportV1 {2686 MediaFeedbackReportV1 {2687 report_id,2688 interval_ms: 1_000,2689 flags: MediaFeedbackFlags::NONE,2690 last_sequence: Some(report_id),2691 transport_dropped_delta: 0,2692 frontend_dropped_delta: 0,2693 decoder_dropped_delta: 0,2694 presenter_dropped_delta: 0,2695 frontend_queue_depth: 0,2696 frontend_queue_capacity: 4,2697 decode_queue_depth: 0,2698 decode_queue_capacity: 4,2699 presenter_queue_depth: 0,2700 presenter_queue_capacity: 2,2701 transport_delivery_p95_ms: Some(10),2702 decode_p95_ms: Some(3),2703 presentation_p95_ms: Some(5),2704 }2705 }27062707 fn aggregate_evaluation(2708 report_id: u64,2709 target_kbps: u32,2710 severity: FeedbackSeverity,2711 clean: bool,2712 ) -> AdaptiveBitrateEvaluation {2713 AdaptiveBitrateEvaluation {2714 decision: AdaptiveBitrateDecisionV1 {2715 decision_id: report_id,2716 report_id,2717 target_kbps,2718 floor_kbps: ADAPTIVE_BITRATE_FLOOR_KBPS,2719 ceiling_kbps: 12_000,2720 state: AdaptiveBitrateStateV1::Hold,2721 reasons: if severity == FeedbackSeverity::Clean {2722 AdaptiveBitrateReasonFlagsV1::CLEAN_RECOVERY2723 } else {2724 AdaptiveBitrateReasonFlagsV1::RECEIVER_QUEUE2725 },2726 applied: false,2727 },2728 pressure_severity: severity,2729 clean_recovery_evidence: clean,2730 fresh: severity != FeedbackSeverity::Stale,2731 }2732 }27332734 2735 fn aggregate_uses_worst_fresh_target_and_requires_every_contributor_clean() {2736 let clean = aggregate_evaluation(1, 10_000, FeedbackSeverity::Clean, true);2737 let pressured = aggregate_evaluation(2, 6_000, FeedbackSeverity::Moderate, false);2738 let active = vec![(clean, clean_feedback(1)), (pressured, clean_feedback(2))];2739 assert_eq!(2740 aggregate_target_and_clean(&active, 12_000, 12_000),2741 (6_000, false)2742 );27432744 let recovered = vec![2745 (clean, clean_feedback(3)),2746 (2747 aggregate_evaluation(4, 8_000, FeedbackSeverity::Clean, true),2748 clean_feedback(4),2749 ),2750 ];2751 assert_eq!(2752 aggregate_target_and_clean(&recovered, 6_000, 12_000),2753 (8_000, true)2754 );2755 }27562757 2758 fn sub_millisecond_rtt_jitter_does_not_become_permanent_moderate_pressure() {2759 let now = Instant::now();2760 let baseline = MediaV3TelemetrySnapshot {2761 selected_path_rtt_micros: 500,2762 ..Default::default()2763 };2764 let mut controller = ShadowBitrateController::new(12_000, baseline);2765 controller2766 .evaluate(&clean_feedback(1), baseline, now)2767 .unwrap();2768 let jitter = MediaV3TelemetrySnapshot {2769 selected_path_rtt_micros: 1_000,2770 ..Default::default()2771 };2772 let evaluation = controller2773 .evaluate(&clean_feedback(2), jitter, now + Duration::from_secs(1))2774 .unwrap();2775 assert!(2776 !evaluation2777 .decision2778 .reasons2779 .contains(AdaptiveBitrateReasonFlagsV1::RTT_INFLATION)2780 );2781 assert_eq!(evaluation.pressure_severity, FeedbackSeverity::Clean);2782 }27832784 // Regression: the shipped activation guide writes2785 // encoder_backend = "external-gst-launch", which puts adaptive in shadow.2786 // Detaching is an action, so shadow mode must not reach for it: the viewer2787 // is below floor precisely because no reduction was ever applied.2788 2789 fn shadow_mode_never_gains_permission_to_detach_a_viewer() {2790 let external = gamescope_feedback_test_config(VaapiRateControl::Cbr);2791 assert_eq!(2792 external2793 .gamescope_pipewire2794 .as_ref()2795 .unwrap()2796 .encoder_backend,2797 GamescopeEncoderBackend::ExternalGstLaunch,2798 "the activation guide's backend is the one that selects shadow mode"2799 );2800 assert!(!adaptive_bitrate_actuation_enabled(&external));2801 assert!(!encoder_actuation_available(&external, None));28022803 let mut in_process = external.clone();2804 in_process2805 .gamescope_pipewire2806 .as_mut()2807 .unwrap()2808 .encoder_backend = GamescopeEncoderBackend::InProcessGstreamer;2809 assert!(adaptive_bitrate_actuation_enabled(&in_process));2810 // Still shadow without a live encoder handle: nothing to actuate.2811 assert!(!encoder_actuation_available(&in_process, None));2812 }28132814 2815 fn persistent_floor_breach_retries_then_detaches_without_holding_survivors_down() {2816 let now = Instant::now();2817 let severe = aggregate_evaluation(2818 1,2819 ADAPTIVE_BITRATE_FLOOR_KBPS,2820 FeedbackSeverity::Severe,2821 false,2822 );2823 let mut recovery = ContributorRecoveryState::default();2824 assert_eq!(recovery.observe(&severe, now), RecoveryMutation::None);2825 assert_eq!(recovery.observe(&severe, now), RecoveryMutation::None);2826 assert_eq!(2827 recovery.observe(&severe, now),2828 RecoveryMutation::RequestRecovery2829 );2830 assert!(!recovery.contributes());28312832 let first_retry = now + ISOLATED_RECOVERY_BACKOFF;2833 assert_eq!(2834 recovery.observe(&severe, first_retry),2835 RecoveryMutation::None2836 );2837 assert_eq!(2838 recovery.observe(&severe, first_retry),2839 RecoveryMutation::None2840 );2841 assert_eq!(2842 recovery.observe(&severe, first_retry),2843 RecoveryMutation::RequestRecovery2844 );2845 let second_retry = first_retry + ISOLATED_RECOVERY_BACKOFF.saturating_mul(2);2846 assert_eq!(2847 recovery.observe(&severe, second_retry),2848 RecoveryMutation::None2849 );2850 assert_eq!(2851 recovery.observe(&severe, second_retry),2852 RecoveryMutation::None2853 );2854 assert_eq!(2855 recovery.observe(&severe, second_retry),2856 RecoveryMutation::Detach2857 );2858 assert_eq!(recovery, ContributorRecoveryState::Detaching);2859 }28602861 2862 fn isolated_viewer_rejoins_only_after_bounded_clean_recovery() {2863 let now = Instant::now();2864 let severe = aggregate_evaluation(2865 1,2866 ADAPTIVE_BITRATE_FLOOR_KBPS,2867 FeedbackSeverity::Severe,2868 false,2869 );2870 let clean = aggregate_evaluation(2, 2_000, FeedbackSeverity::Clean, true);2871 let mut recovery = ContributorRecoveryState::default();2872 for _ in 0..FLOOR_BREACH_WINDOWS {2873 let _ = recovery.observe(&severe, now);2874 }2875 let retry = now + ISOLATED_RECOVERY_BACKOFF;2876 assert_eq!(recovery.observe(&clean, retry), RecoveryMutation::None);2877 assert_eq!(recovery.observe(&clean, retry), RecoveryMutation::None);2878 assert_eq!(recovery.observe(&clean, retry), RecoveryMutation::Recovered);2879 assert!(recovery.contributes());2880 }28812882 fn controller_decide(2883 controller: &mut ShadowBitrateController,2884 report: &MediaFeedbackReportV1,2885 second: u64,2886 ) -> AdaptiveBitrateDecisionV1 {2887 controller2888 .decide(2889 report,2890 MediaV3TelemetrySnapshot::default(),2891 Instant::now() + Duration::from_secs(second),2892 )2893 .unwrap()2894 }28952896 fn evaluate_commit_plan(2897 controller: &mut ShadowBitrateController,2898 resolution_controller: &mut Option<MotionResolutionPolicy>,2899 report: &MediaFeedbackReportV1,2900 observed_at: Instant,2901 ) -> AdaptiveCommitPlan {2902 let telemetry = MediaV3TelemetrySnapshot::default();2903 let evaluation = controller.evaluate(report, telemetry, observed_at).unwrap();2904 let resolution_decision = resolution_controller2905 .as_mut()2906 .map(|policy| policy.observe_window(report, &evaluation, observed_at));2907 AdaptiveCommitPlan {2908 decision: evaluation.decision,2909 telemetry,2910 resolution_decision,2911 force_keyframe: adaptive_recovery_keyframe_required(report, &evaluation.decision),2912 }2913 }29142915 fn native_dimensions() -> VideoDimensions {2916 VideoDimensions {2917 width: 1_280,2918 height: 800,2919 }2920 }29212922 2923 fn motion_resolution_tier_is_exact_even_and_same_aspect() {2924 let native = native_dimensions();2925 let policy = MotionResolutionPolicy::new(native).unwrap();2926 assert_eq!(policy.target(), native);2927 assert_eq!(2928 policy.reduced,2929 VideoDimensions {2930 width: 960,2931 height: 600,2932 }2933 );2934 assert_eq!(2935 u32::from(native.width) * u32::from(policy.reduced.height),2936 u32::from(native.height) * u32::from(policy.reduced.width)2937 );2938 assert!(policy.reduced.width.is_multiple_of(2));2939 assert!(policy.reduced.height.is_multiple_of(2));29402941 assert!(2942 MotionResolutionPolicy::new(VideoDimensions {2943 width: 1_278,2944 height: 800,2945 })2946 .is_err()2947 );2948 assert!(2949 MotionResolutionPolicy::new(VideoDimensions {2950 width: 80,2951 height: 64,2952 })2953 .is_err()2954 );2955 assert!(motion_resolution_policy(1_920, 1_080).unwrap().is_some());2956 assert!(motion_resolution_policy(1_366, 768).unwrap().is_none());2957 }29582959 2960 fn feedback_resolution_policy_uses_encoder_generation_dimensions() {2961 let full_hd = crate::source::EncoderControlTestHarness::with_dimensions(1_920, 1_080);2962 let policy = motion_resolution_policy_for_encoder(&full_hd.control)2963 .unwrap()2964 .unwrap();2965 assert_eq!(2966 policy.native,2967 VideoDimensions {2968 width: 1_920,2969 height: 1_080,2970 }2971 );29722973 let no_exact_tier = crate::source::EncoderControlTestHarness::with_dimensions(1_366, 768);2974 assert!(2975 motion_resolution_policy_for_encoder(&no_exact_tier.control)2976 .unwrap()2977 .is_none()2978 );2979 }29802981 2982 fn motion_downshifts_immediately_and_recovers_only_after_settle_and_clean_windows() {2983 let now = Instant::now();2984 let mut policy = MotionResolutionPolicy::new(native_dimensions()).unwrap();2985 let down = policy.observe_motion(true, now);2986 assert_eq!(down.tier, ResolutionTier::Reduced);2987 assert_eq!(down.trigger, ResolutionTrigger::Motion);2988 assert!(down.changed);29892990 let still = policy.observe_motion(false, now + Duration::from_secs(1));2991 assert!(!still.changed);2992 for second in [2, 3] {2993 assert!(2994 !policy2995 .observe_feedback(2996 FeedbackSeverity::Clean,2997 true,2998 now + Duration::from_secs(second),2999 )3000 .changed3001 );3002 }3003 let recovered =3004 policy.observe_feedback(FeedbackSeverity::Clean, true, now + Duration::from_secs(4));3005 assert_eq!(recovered.tier, ResolutionTier::Native);3006 assert_eq!(recovered.target, native_dimensions());3007 assert_eq!(recovered.trigger, ResolutionTrigger::Recovery);3008 assert!(recovered.changed);3009 }30103011 3012 fn repeated_still_observations_do_not_postpone_resolution_recovery() {3013 let now = Instant::now();3014 let mut policy = MotionResolutionPolicy::new(native_dimensions()).unwrap();3015 policy.observe_motion(true, now);3016 policy.observe_motion(false, now + Duration::from_secs(1));3017 policy.observe_feedback(FeedbackSeverity::Clean, true, now + Duration::from_secs(2));3018 policy.observe_motion(false, now + Duration::from_millis(2_500));3019 policy.observe_feedback(FeedbackSeverity::Clean, true, now + Duration::from_secs(3));3020 let recovered =3021 policy.observe_feedback(FeedbackSeverity::Clean, true, now + Duration::from_secs(4));3022 assert!(recovered.changed);3023 assert_eq!(recovered.tier, ResolutionTier::Native);3024 }30253026 3027 fn resolution_pressure_requires_fresh_hysteresis_but_severe_is_immediate() {3028 let now = Instant::now();3029 let mut moderate = MotionResolutionPolicy::new(native_dimensions()).unwrap();3030 moderate.observe_motion(false, now);3031 assert!(3032 !moderate3033 .observe_feedback(3034 FeedbackSeverity::Moderate,3035 true,3036 now + Duration::from_secs(1),3037 )3038 .changed3039 );3040 assert!(3041 !moderate3042 .observe_feedback(3043 FeedbackSeverity::Moderate,3044 false,3045 now + Duration::from_secs(2),3046 )3047 .changed3048 );3049 assert!(3050 !moderate3051 .observe_feedback(3052 FeedbackSeverity::Moderate,3053 true,3054 now + Duration::from_secs(3),3055 )3056 .changed3057 );3058 let down = moderate.observe_feedback(3059 FeedbackSeverity::Moderate,3060 true,3061 now + Duration::from_secs(4),3062 );3063 assert!(down.changed);3064 assert_eq!(down.trigger, ResolutionTrigger::Pressure);30653066 let mut severe = MotionResolutionPolicy::new(native_dimensions()).unwrap();3067 let immediate =3068 severe.observe_feedback(FeedbackSeverity::Severe, true, now + Duration::from_secs(1));3069 assert!(immediate.changed);3070 assert_eq!(immediate.tier, ResolutionTier::Reduced);3071 }30723073 3074 fn stale_feedback_cannot_restore_native_resolution() {3075 let now = Instant::now();3076 let mut policy = MotionResolutionPolicy::new(native_dimensions()).unwrap();3077 policy.observe_motion(false, now);3078 policy.observe_feedback(FeedbackSeverity::Severe, true, now + Duration::from_secs(1));3079 for second in [2, 3, 4, 5] {3080 let decision = policy.observe_feedback(3081 FeedbackSeverity::Clean,3082 false,3083 now + Duration::from_secs(second),3084 );3085 assert_eq!(decision.tier, ResolutionTier::Reduced);3086 assert!(!decision.changed);3087 }3088 assert!(policy.pressure_active);30893090 for second in [6, 7] {3091 assert!(3092 !policy3093 .observe_feedback(3094 FeedbackSeverity::Clean,3095 true,3096 now + Duration::from_secs(second),3097 )3098 .changed3099 );3100 }3101 let recovered =3102 policy.observe_feedback(FeedbackSeverity::Clean, true, now + Duration::from_secs(8));3103 assert!(recovered.changed);3104 assert_eq!(recovered.tier, ResolutionTier::Native);3105 }31063107 3108 fn motion_keeps_resolution_reduced_after_pressure_clears() {3109 let now = Instant::now();3110 let mut policy = MotionResolutionPolicy::new(native_dimensions()).unwrap();3111 policy.observe_motion(true, now);3112 policy.observe_feedback(FeedbackSeverity::Severe, true, now + Duration::from_secs(1));3113 for second in 2..=5 {3114 let decision = policy.observe_feedback(3115 FeedbackSeverity::Clean,3116 true,3117 now + Duration::from_secs(second),3118 );3119 assert_eq!(decision.tier, ResolutionTier::Reduced);3120 assert!(!decision.changed);3121 }3122 assert!(!policy.pressure_active);3123 assert!(policy.motion_active);3124 }31253126 3127 fn damage_driven_motion_proxy_seeds_then_downshifts_and_stale_blocks_upscale() {3128 let now = Instant::now();3129 let telemetry = MediaV3TelemetrySnapshot::default();3130 let mut bitrate = ShadowBitrateController::new(12_000, telemetry);3131 let mut resolution = MotionResolutionPolicy::new(native_dimensions()).unwrap();31323133 let mut first = clean_feedback(1);3134 first.last_sequence = Some(100);3135 let first_evaluation = bitrate.evaluate(&first, telemetry, now).unwrap();3136 assert!(!first_evaluation.fresh);3137 let seeded = resolution.observe_window(&first, &first_evaluation, now);3138 assert_eq!(seeded.tier, ResolutionTier::Native);3139 assert!(!seeded.changed);31403141 let mut second = clean_feedback(2);3142 second.last_sequence = Some(110);3143 let second_evaluation = bitrate3144 .evaluate(&second, telemetry, now + Duration::from_secs(1))3145 .unwrap();3146 assert!(second_evaluation.fresh);3147 let down =3148 resolution.observe_window(&second, &second_evaluation, now + Duration::from_secs(1));3149 assert_eq!(down.trigger, ResolutionTrigger::Motion);3150 assert_eq!(down.tier, ResolutionTier::Reduced);3151 assert!(down.changed);31523153 let mut stale = clean_feedback(3);3154 stale.interval_ms = 5_000;3155 stale.last_sequence = Some(110);3156 let stale_evaluation = bitrate3157 .evaluate(&stale, telemetry, now + Duration::from_secs(5))3158 .unwrap();3159 assert!(!stale_evaluation.fresh);3160 let held =3161 resolution.observe_window(&stale, &stale_evaluation, now + Duration::from_secs(5));3162 assert_eq!(held.tier, ResolutionTier::Reduced);3163 assert!(!held.changed);3164 }31653166 3167 fn damage_driven_no_progress_cannot_restore_native_resolution() {3168 let now = Instant::now();3169 let telemetry = MediaV3TelemetrySnapshot::default();3170 let mut bitrate = ShadowBitrateController::new(12_000, telemetry);3171 let mut resolution = MotionResolutionPolicy::new(native_dimensions()).unwrap();31723173 let mut seed = clean_feedback(1);3174 seed.last_sequence = Some(100);3175 let seed_evaluation = bitrate.evaluate(&seed, telemetry, now).unwrap();3176 resolution.observe_window(&seed, &seed_evaluation, now);31773178 let mut pressure = clean_feedback(2);3179 pressure.last_sequence = Some(110);3180 pressure.flags = MediaFeedbackFlags::RESYNC_ACTIVE;3181 let pressure_evaluation = bitrate3182 .evaluate(&pressure, telemetry, now + Duration::from_secs(1))3183 .unwrap();3184 let downshift = resolution.observe_window(3185 &pressure,3186 &pressure_evaluation,3187 now + Duration::from_secs(1),3188 );3189 assert_eq!(downshift.tier, ResolutionTier::Reduced);3190 assert!(downshift.changed);31913192 for second in 2..=4 {3193 let mut no_progress = clean_feedback(second + 1);3194 no_progress.last_sequence = Some(110);3195 let evaluation = bitrate3196 .evaluate(&no_progress, telemetry, now + Duration::from_secs(second))3197 .unwrap();3198 assert_eq!(evaluation.pressure_severity, FeedbackSeverity::Clean);3199 assert!(!evaluation.clean_recovery_evidence);3200 let decision = resolution.observe_window(3201 &no_progress,3202 &evaluation,3203 now + Duration::from_secs(second),3204 );3205 assert_eq!(decision.tier, ResolutionTier::Reduced);3206 assert_eq!(decision.trigger, ResolutionTrigger::Pressure);3207 assert!(!decision.changed);3208 assert_eq!(resolution.clean_windows, 0);3209 }32103211 for (report_id, second, sequence) in [(6, 5, 111), (7, 6, 112), (8, 7, 113)] {3212 let mut low_motion = clean_feedback(report_id);3213 low_motion.last_sequence = Some(sequence);3214 let evaluation = bitrate3215 .evaluate(&low_motion, telemetry, now + Duration::from_secs(second))3216 .unwrap();3217 assert!(evaluation.clean_recovery_evidence);3218 let decision = resolution.observe_window(3219 &low_motion,3220 &evaluation,3221 now + Duration::from_secs(second),3222 );3223 if report_id < 8 {3224 assert_eq!(decision.tier, ResolutionTier::Reduced);3225 assert!(!decision.changed);3226 } else {3227 assert_eq!(decision.tier, ResolutionTier::Native);3228 assert_eq!(decision.trigger, ResolutionTrigger::Recovery);3229 assert!(decision.changed);3230 }3231 }3232 }32333234 3235 fn shadow_controller_immediately_decreases_for_severe_feedback_and_stays_bounded() {3236 let mut controller =3237 ShadowBitrateController::new(12_137, MediaV3TelemetrySnapshot::default());3238 assert_eq!(controller.ceiling_kbps, 12_000);3239 let first = controller_decide(&mut controller, &clean_feedback(1), 0);3240 assert_eq!(first.state, AdaptiveBitrateStateV1::Hold);3241 assert!(3242 first3243 .reasons3244 .contains(AdaptiveBitrateReasonFlagsV1::FEEDBACK_STALE)3245 );3246 for report_id in 2..=21 {3247 let mut report = clean_feedback(report_id);3248 report.flags = MediaFeedbackFlags::RESYNC_ACTIVE;3249 let decision = controller_decide(&mut controller, &report, report_id - 1);3250 assert!(!decision.applied);3251 assert!(decision.target_kbps >= ADAPTIVE_BITRATE_FLOOR_KBPS);3252 assert!(decision.target_kbps <= 12_000);3253 assert_eq!(decision.target_kbps % ADAPTIVE_BITRATE_QUANTUM_KBPS, 0);3254 assert!(3255 decision3256 .reasons3257 .contains(AdaptiveBitrateReasonFlagsV1::RECEIVER_QUEUE)3258 );3259 }3260 assert_eq!(controller.target_kbps, ADAPTIVE_BITRATE_FLOOR_KBPS);3261 }32623263 3264 fn shadow_controller_requires_two_moderate_windows() {3265 let mut controller =3266 ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());3267 controller_decide(&mut controller, &clean_feedback(1), 0);3268 let mut first = clean_feedback(2);3269 first.frontend_queue_depth = 2;3270 assert_eq!(3271 controller_decide(&mut controller, &first, 1).state,3272 AdaptiveBitrateStateV1::Hold3273 );3274 let mut second = clean_feedback(3);3275 second.frontend_queue_depth = 2;3276 let decision = controller_decide(&mut controller, &second, 2);3277 assert_eq!(decision.state, AdaptiveBitrateStateV1::Decrease);3278 assert_eq!(decision.target_kbps, 9_500);3279 }32803281 3282 fn shadow_controller_increases_only_after_clean_cooldown() {3283 let mut controller =3284 ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());3285 controller_decide(&mut controller, &clean_feedback(1), 0);3286 let mut severe = clean_feedback(2);3287 severe.decoder_dropped_delta = 1;3288 assert_eq!(3289 controller_decide(&mut controller, &severe, 1).target_kbps,3290 9_0003291 );3292 for report_id in 3..=11 {3293 let decision =3294 controller_decide(&mut controller, &clean_feedback(report_id), report_id - 1);3295 assert_eq!(decision.state, AdaptiveBitrateStateV1::Hold);3296 assert_eq!(decision.target_kbps, 9_000);3297 }3298 let decision = controller_decide(&mut controller, &clean_feedback(12), 11);3299 assert_eq!(decision.state, AdaptiveBitrateStateV1::Increase);3300 assert_eq!(decision.target_kbps, 9_250);3301 assert!(3302 decision3303 .reasons3304 .contains(AdaptiveBitrateReasonFlagsV1::CLEAN_RECOVERY)3305 );3306 }33073308 3309 fn stale_or_non_monotonic_feedback_cannot_drive_an_increase() {3310 let mut controller =3311 ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());3312 controller_decide(&mut controller, &clean_feedback(1), 0);3313 let mut severe = clean_feedback(2);3314 severe.flags = MediaFeedbackFlags::RESYNC_ACTIVE;3315 controller_decide(&mut controller, &severe, 1);3316 let stale = controller_decide(&mut controller, &clean_feedback(3), 4);3317 assert_eq!(stale.state, AdaptiveBitrateStateV1::Hold);3318 assert_eq!(stale.target_kbps, 9_000);3319 assert!(3320 stale3321 .reasons3322 .contains(AdaptiveBitrateReasonFlagsV1::FEEDBACK_STALE)3323 );3324 assert!(3325 controller3326 .decide(3327 &clean_feedback(2),3328 MediaV3TelemetrySnapshot::default(),3329 Instant::now() + Duration::from_secs(4),3330 )3331 .is_err()3332 );33333334 let now = Instant::now();3335 let mut rapid = ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());3336 rapid3337 .decide(&clean_feedback(1), MediaV3TelemetrySnapshot::default(), now)3338 .unwrap();3339 let too_fast = rapid3340 .decide(3341 &clean_feedback(2),3342 MediaV3TelemetrySnapshot::default(),3343 now + Duration::from_millis(500),3344 )3345 .unwrap();3346 assert_eq!(too_fast.state, AdaptiveBitrateStateV1::Hold);3347 assert!(3348 too_fast3349 .reasons3350 .contains(AdaptiveBitrateReasonFlagsV1::FEEDBACK_STALE)3351 );3352 }33533354 3355 fn full_decode_queue_is_severe_even_at_capacity_two() {3356 let mut controller =3357 ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());3358 controller_decide(&mut controller, &clean_feedback(1), 0);3359 let mut report = clean_feedback(2);3360 report.decode_queue_capacity = 2;3361 report.decode_queue_depth = 2;3362 let decision = controller_decide(&mut controller, &report, 1);3363 assert_eq!(decision.state, AdaptiveBitrateStateV1::Decrease);3364 assert!(3365 decision3366 .reasons3367 .contains(AdaptiveBitrateReasonFlagsV1::DECODE_BACKLOG)3368 );3369 }33703371 3372 fn decision_id_exhaustion_fails_instead_of_repeating() {3373 let mut controller =3374 ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());3375 controller.next_decision_id = u64::MAX;3376 assert!(3377 controller3378 .decide(3379 &clean_feedback(1),3380 MediaV3TelemetrySnapshot::default(),3381 Instant::now(),3382 )3383 .is_err()3384 );3385 assert_eq!(controller.next_decision_id, u64::MAX);3386 }33873388 3389 fn stale_feedback_applies_trusted_loss_and_send_failure_pressure_once() {3390 for telemetry in [3391 MediaV3TelemetrySnapshot {3392 selected_path_lost_packets: 1,3393 ..MediaV3TelemetrySnapshot::default()3394 },3395 MediaV3TelemetrySnapshot {3396 send_failures: 1,3397 ..MediaV3TelemetrySnapshot::default()3398 },3399 ] {3400 let now = Instant::now();3401 let mut controller =3402 ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());3403 let decision = controller3404 .decide(&clean_feedback(1), telemetry, now)3405 .unwrap();3406 assert_eq!(decision.state, AdaptiveBitrateStateV1::Decrease);3407 assert_eq!(decision.target_kbps, 9_000);3408 assert!(3409 decision3410 .reasons3411 .contains(AdaptiveBitrateReasonFlagsV1::FEEDBACK_STALE)3412 );3413 assert!(3414 decision3415 .reasons3416 .contains(AdaptiveBitrateReasonFlagsV1::LOSS_OR_CANCELLATION)3417 );34183419 let repeated = controller3420 .decide(3421 &clean_feedback(2),3422 telemetry,3423 now + Duration::from_millis(500),3424 )3425 .unwrap();3426 assert_eq!(repeated.state, AdaptiveBitrateStateV1::Hold);3427 assert_eq!(repeated.target_kbps, 9_000);3428 }3429 }34303431 3432 fn stale_moderate_cancellation_is_retained_until_hysteresis_reduces() {3433 let now = Instant::now();3434 let telemetry = MediaV3TelemetrySnapshot {3435 scheduler_cancellations: 1,3436 ..MediaV3TelemetrySnapshot::default()3437 };3438 let mut controller =3439 ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());3440 let first = controller3441 .decide(&clean_feedback(1), telemetry, now)3442 .unwrap();3443 assert_eq!(first.state, AdaptiveBitrateStateV1::Hold);3444 assert_eq!(controller.previous_telemetry.scheduler_cancellations, 0);34453446 let second = controller3447 .decide(3448 &clean_feedback(2),3449 telemetry,3450 now + Duration::from_millis(500),3451 )3452 .unwrap();3453 assert_eq!(second.state, AdaptiveBitrateStateV1::Decrease);3454 assert_eq!(second.target_kbps, 9_500);3455 assert_eq!(controller.previous_telemetry.scheduler_cancellations, 1);3456 }34573458 3459 fn clean_recovery_below_five_mbps_advances_by_one_quantum() {3460 let mut controller =3461 ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());3462 controller.target_kbps = 4_000;3463 controller_decide(&mut controller, &clean_feedback(1), 0);3464 controller_decide(&mut controller, &clean_feedback(2), 1);3465 for report_id in 3..=11 {3466 let decision =3467 controller_decide(&mut controller, &clean_feedback(report_id), report_id - 1);3468 assert_eq!(decision.state, AdaptiveBitrateStateV1::Hold);3469 }3470 let decision = controller_decide(&mut controller, &clean_feedback(12), 11);3471 assert_eq!(decision.state, AdaptiveBitrateStateV1::Increase);3472 assert_eq!(decision.target_kbps, 4_250);3473 }34743475 3476 fn recovery_requires_latency_samples_and_sequence_progress_but_pressure_still_reduces() {3477 let mut controller =3478 ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());3479 controller.target_kbps = 9_000;3480 controller_decide(&mut controller, &clean_feedback(1), 0);3481 controller_decide(&mut controller, &clean_feedback(2), 1);34823483 for report_id in 3..=14 {3484 let mut report = clean_feedback(report_id);3485 report.last_sequence = Some(2);3486 let decision = controller_decide(&mut controller, &report, report_id - 1);3487 assert_eq!(decision.state, AdaptiveBitrateStateV1::Hold);3488 assert_eq!(decision.target_kbps, 9_000);3489 }3490 assert_eq!(controller.clean_windows, 0);34913492 for report_id in 15..=26 {3493 let mut report = clean_feedback(report_id);3494 report.decode_p95_ms = None;3495 let decision = controller_decide(&mut controller, &report, report_id - 1);3496 assert_eq!(decision.state, AdaptiveBitrateStateV1::Hold);3497 assert_eq!(decision.target_kbps, 9_000);3498 }3499 assert_eq!(controller.clean_windows, 0);35003501 let mut pressure = clean_feedback(27);3502 pressure.decode_p95_ms = None;3503 pressure.decode_queue_capacity = 2;3504 pressure.decode_queue_depth = 2;3505 let decision = controller_decide(&mut controller, &pressure, 26);3506 assert_eq!(decision.state, AdaptiveBitrateStateV1::Decrease);3507 assert_eq!(decision.target_kbps, 6_750);3508 assert!(3509 decision3510 .reasons3511 .contains(AdaptiveBitrateReasonFlagsV1::DECODE_BACKLOG)3512 );3513 }35143515 fn gamescope_feedback_test_config(rate_control: VaapiRateControl) -> HostConfig {3516 let (bitrate_kbps, quantizer) = match rate_control {3517 VaapiRateControl::Cbr => (Some(12_123), None),3518 VaapiRateControl::Cqp => (None, Some(24)),3519 };3520 let mut config = moq_test_config();3521 config.source = VideoSource::GamescopePipewire;3522 config.width = None;3523 config.height = None;3524 config.gamescope_pipewire = Some(crate::config::GamescopePipewireConfig {3525 node_name: "gamescope".into(),3526 media_class: "Video/Source".into(),3527 match_properties: std::collections::BTreeMap::new(),3528 xwayland_display: None,3529 pw_dump_path: "/usr/bin/pw-dump".into(),3530 gst_launch_path: "/usr/bin/gst-launch-1.0".into(),3531 gst_inspect_path: "/usr/bin/gst-inspect-1.0".into(),3532 encoder_backend: GamescopeEncoderBackend::ExternalGstLaunch,3533 vaapi_encoder: "vah264enc".into(),3534 vaapi_render_node: "/dev/dri/renderD128".into(),3535 rate_control,3536 bitrate_kbps,3537 quantizer,3538 });3539 config.validate().unwrap();3540 config3541 }35423543 3544 fn external_cqp_uses_a_fixed_shadow_bitrate_ceiling() {3545 let config = gamescope_feedback_test_config(VaapiRateControl::Cqp);35463547 assert_eq!(3548 adaptive_bitrate_ceiling_kbps(&config).unwrap(),3549 SHADOW_BITRATE_CEILING_KBPS3550 );3551 assert!(!adaptive_bitrate_actuation_enabled(&config));3552 }35533554 3555 fn gamescope_cbr_keeps_its_quantized_configured_ceiling() {3556 let config = gamescope_feedback_test_config(VaapiRateControl::Cbr);35573558 assert_eq!(adaptive_bitrate_ceiling_kbps(&config).unwrap(), 12_000);3559 }35603561 3562 fn malformed_cbr_without_a_bitrate_still_fails_closed() {3563 let mut config = gamescope_feedback_test_config(VaapiRateControl::Cbr);3564 config.gamescope_pipewire.as_mut().unwrap().bitrate_kbps = None;35653566 assert_eq!(3567 adaptive_bitrate_ceiling_kbps(&config)3568 .unwrap_err()3569 .to_string(),3570 "adaptive bitrate shadow mode requires a Gamescope CBR bitrate"3571 );3572 }35733574 3575 async fn adaptive_encoder_commit_requires_exact_readback_and_tracks_recovery_separately() {3576 let sessions = Arc::new(SessionRegistry::default());3577 let remote = endpoint(1);3578 let media = sessions3579 .claim(remote, [4; 16], InvitationGrants::VIEW)3580 .unwrap();3581 let harness = crate::source::EncoderControlTestHarness::new();3582 sessions3583 .install_encoder_control(remote, media.session_id, Some(harness.control.clone()))3584 .unwrap();35853586 let proposal = sessions3587 .propose_adaptive_encoder_update(remote, media.session_id, 8_000, true)3588 .unwrap()3589 .unwrap();3590 harness.status.send_modify(|status| {3591 status.applied_bitrate_revision = Some(proposal.bitrate_revision);3592 status.applied_bitrate_kbps = Some(8_000);3593 status.requested_force_keyframe_revision = proposal.force_keyframe_revision;3594 status.acknowledged_force_keyframe_revision = proposal.force_keyframe_revision;3595 });3596 assert!(matches!(3597 commit_adaptive_encoder_proposal(&sessions, remote, media.session_id, &proposal).await,3598 AdaptiveEncoderCommit::Applied3599 ));3600 assert!(matches!(3601 await_adaptive_recovery_keyframe(&sessions, remote, media.session_id, &proposal).await,3602 Some(Ok(()))3603 ));36043605 let mismatch = sessions3606 .propose_adaptive_encoder_update(remote, media.session_id, 7_000, false)3607 .unwrap()3608 .unwrap();3609 harness.status.send_modify(|status| {3610 status.applied_bitrate_revision = Some(mismatch.bitrate_revision);3611 status.applied_bitrate_kbps = Some(7_250);3612 });3613 assert!(matches!(3614 commit_adaptive_encoder_proposal(&sessions, remote, media.session_id, &mismatch).await,3615 AdaptiveEncoderCommit::NotApplied(_)3616 ));36173618 let stale_generation = sessions3619 .propose_adaptive_encoder_update(remote, media.session_id, 6_000, false)3620 .unwrap()3621 .unwrap();3622 let session_id = media.session_id;3623 drop(media);3624 assert!(matches!(3625 commit_adaptive_encoder_proposal(&sessions, remote, session_id, &stale_generation)3626 .await,3627 AdaptiveEncoderCommit::GenerationEnded3628 ));3629 }36303631 3632 fn adaptive_recovery_idr_is_only_requested_for_a_bitrate_change_during_resync() {3633 let mut report = clean_feedback(1);3634 let mut decision = AdaptiveBitrateDecisionV1 {3635 decision_id: 1,3636 report_id: 1,3637 target_kbps: 8_000,3638 floor_kbps: 1_000,3639 ceiling_kbps: 12_000,3640 state: AdaptiveBitrateStateV1::Decrease,3641 reasons: AdaptiveBitrateReasonFlagsV1::RECEIVER_QUEUE,3642 applied: false,3643 };3644 assert!(!adaptive_recovery_keyframe_required(&report, &decision));3645 report.flags = MediaFeedbackFlags::RESYNC_ACTIVE;3646 assert!(adaptive_recovery_keyframe_required(&report, &decision));3647 decision.state = AdaptiveBitrateStateV1::Hold;3648 assert!(!adaptive_recovery_keyframe_required(&report, &decision));3649 }36503651 3652 async fn stalled_adaptive_apply_coalesces_feedback_without_losing_pressure() {3653 use tokio::io::AsyncWriteExt as _;36543655 let (sender, mut receiver) =3656 tokio::sync::watch::channel(CumulativeFeedbackIngress::default());3657 let (mut writer, reader) = tokio::io::duplex(512);3658 let reader_task = tokio::spawn(forward_media_feedback_reports(reader, sender.clone()));3659 let mut cursor = FeedbackIngressCursor::default();36603661 sigil_protocol::write_media_feedback_report_v1(&mut writer, &clean_feedback(1))3662 .await3663 .unwrap();3664 receiver.changed().await.unwrap();3665 let (first, next_cursor) = receiver3666 .borrow_and_update()3667 .report_since(cursor)3668 .unwrap()3669 .unwrap();3670 assert_eq!(first.report_id, 1);3671 cursor = next_cursor;36723673 // The consumer is now notionally stalled on an encoder commit. Two3674 // reports overwrite one watch notification, but their interval,3675 // deltas, and resync pressure must remain cumulative.3676 let mut pressured = clean_feedback(2);3677 pressured.flags = MediaFeedbackFlags::RESYNC_ACTIVE;3678 pressured.transport_dropped_delta = 2;3679 pressured.decoder_dropped_delta = 1;3680 pressured.decode_queue_depth = 3;3681 pressured.transport_delivery_p95_ms = Some(180);3682 sigil_protocol::write_media_feedback_report_v1(&mut writer, &pressured)3683 .await3684 .unwrap();36853686 let mut latest = clean_feedback(3);3687 latest.frontend_dropped_delta = 4;3688 latest.presenter_dropped_delta = 3;3689 latest.frontend_queue_depth = 3;3690 latest.decode_queue_depth = 1;3691 latest.transport_delivery_p95_ms = Some(40);3692 latest.decode_p95_ms = Some(9);3693 sigil_protocol::write_media_feedback_report_v1(&mut writer, &latest)3694 .await3695 .unwrap();3696 writer.shutdown().await.unwrap();3697 reader_task.await.unwrap().unwrap();36983699 receiver.changed().await.unwrap();3700 let (aggregate, next_cursor) = receiver3701 .borrow_and_update()3702 .report_since(cursor)3703 .unwrap()3704 .unwrap();3705 assert_eq!(aggregate.report_id, 3);3706 assert_eq!(aggregate.interval_ms, 2_000);3707 assert_eq!(aggregate.transport_dropped_delta, 2);3708 assert_eq!(aggregate.frontend_dropped_delta, 4);3709 assert_eq!(aggregate.decoder_dropped_delta, 1);3710 assert_eq!(aggregate.presenter_dropped_delta, 3);3711 assert!(aggregate.flags.contains(MediaFeedbackFlags::RESYNC_ACTIVE));3712 assert_eq!(aggregate.last_sequence, latest.last_sequence);3713 assert_eq!(aggregate.frontend_queue_depth, latest.frontend_queue_depth);3714 assert_eq!(aggregate.decode_queue_depth, latest.decode_queue_depth);3715 assert_eq!(3716 aggregate.transport_delivery_p95_ms,3717 latest.transport_delivery_p95_ms3718 );3719 assert_eq!(aggregate.decode_p95_ms, latest.decode_p95_ms);3720 cursor = next_cursor;3721 assert!(3722 receiver.borrow().report_since(cursor).unwrap().is_none(),3723 "one cumulative ingress generation must be processed only once"3724 );37253726 for report_id in 4..=9 {3727 sender.send_modify(|ingress| ingress.observe(clean_feedback(report_id)));3728 }3729 receiver.changed().await.unwrap();3730 let (clamped, _) = receiver3731 .borrow_and_update()3732 .report_since(cursor)3733 .unwrap()3734 .unwrap();3735 assert_eq!(clamped.interval_ms, 5_000);3736 let mut controller =3737 ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());3738 let decision = controller3739 .decide(3740 &clamped,3741 MediaV3TelemetrySnapshot::default(),3742 Instant::now(),3743 )3744 .unwrap();3745 assert!(3746 decision3747 .reasons3748 .contains(AdaptiveBitrateReasonFlagsV1::FEEDBACK_STALE),3749 "a clamped aggregate must be stale rather than disguised as one fresh report"3750 );3751 }37523753 3754 async fn queued_adaptive_plan_keeps_latest_ids_and_recovery_barrier() {3755 let sessions = Arc::new(SessionRegistry::default());3756 let remote = endpoint(1);3757 let media = sessions3758 .claim(remote, [3; 16], InvitationGrants::VIEW)3759 .unwrap();3760 let harness = crate::source::EncoderControlTestHarness::new();3761 sessions3762 .install_encoder_control(remote, media.session_id, Some(harness.control.clone()))3763 .unwrap();3764 let mut coordinator = AdaptiveCommitCoordinator::new(3765 Arc::clone(&sessions),3766 remote,3767 media.session_id,3768 true,3769 12_000,3770 Some(native_dimensions()),3771 );3772 let mut controller =3773 ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());3774 let mut resolution_controller = None;3775 let observed_at = Instant::now();37763777 let seed = evaluate_commit_plan(3778 &mut controller,3779 &mut resolution_controller,3780 &clean_feedback(1),3781 observed_at,3782 );3783 assert!(coordinator.submit(seed).is_some());37843785 let mut first_pressure = clean_feedback(2);3786 first_pressure.decoder_dropped_delta = 1;3787 let first = evaluate_commit_plan(3788 &mut controller,3789 &mut resolution_controller,3790 &first_pressure,3791 observed_at + Duration::from_secs(1),3792 );3793 assert!(!first.force_keyframe);3794 assert!(coordinator.submit(first).is_none());37953796 let mut recovery_pressure = clean_feedback(3);3797 recovery_pressure.decoder_dropped_delta = 1;3798 recovery_pressure.flags = MediaFeedbackFlags::RESYNC_ACTIVE;3799 let recovery = evaluate_commit_plan(3800 &mut controller,3801 &mut resolution_controller,3802 &recovery_pressure,3803 observed_at + Duration::from_secs(2),3804 );3805 assert!(recovery.force_keyframe);3806 assert!(coordinator.submit(recovery).is_none());38073808 let latest = evaluate_commit_plan(3809 &mut controller,3810 &mut resolution_controller,3811 &clean_feedback(4),3812 observed_at + Duration::from_secs(3),3813 );3814 assert!(!latest.force_keyframe);3815 assert!(coordinator.submit(latest).is_none());3816 let retained = coordinator.queued.expect("latest plan must be retained");3817 assert_eq!(retained.decision.decision_id, latest.decision.decision_id);3818 assert_eq!(retained.decision.report_id, latest.decision.report_id);3819 assert!(3820 retained.force_keyframe,3821 "a coalesced recovery barrier must survive a newer hold decision"3822 );38233824 coordinator.abort_and_drain().await;3825 drop(media);3826 }38273828 3829 async fn timed_out_actuator_requests_are_overwritten_when_policy_returns_to_committed_state() {3830 let sessions = Arc::new(SessionRegistry::default());3831 let remote = endpoint(1);3832 let media = sessions3833 .claim(remote, [8; 16], InvitationGrants::VIEW)3834 .unwrap();3835 let harness = crate::source::EncoderControlTestHarness::new();3836 sessions3837 .install_encoder_control(remote, media.session_id, Some(harness.control.clone()))3838 .unwrap();3839 let native = native_dimensions();3840 let reduced = native.three_quarter().unwrap();3841 let mut coordinator = AdaptiveCommitCoordinator::new(3842 Arc::clone(&sessions),3843 remote,3844 media.session_id,3845 true,3846 12_000,3847 Some(native),3848 );3849 let timed_out = AdaptiveCommitPlan {3850 decision: AdaptiveBitrateDecisionV1 {3851 decision_id: 1,3852 report_id: 1,3853 target_kbps: 9_000,3854 floor_kbps: 1_000,3855 ceiling_kbps: 12_000,3856 state: AdaptiveBitrateStateV1::Decrease,3857 reasons: AdaptiveBitrateReasonFlagsV1::RECEIVER_QUEUE,3858 applied: false,3859 },3860 telemetry: MediaV3TelemetrySnapshot::default(),3861 resolution_decision: Some(MotionResolutionDecision {3862 tier: ResolutionTier::Reduced,3863 target: reduced,3864 trigger: ResolutionTrigger::Pressure,3865 changed: true,3866 }),3867 force_keyframe: false,3868 };3869 assert!(coordinator.submit(timed_out).is_none());3870 let stale_bitrate = harness3871 .requested_bitrate()3872 .expect("the first bitrate request must reach the actuator");3873 let stale_resolution = harness3874 .requested_resolution()3875 .expect("the first resolution request must reach the actuator");3876 assert_eq!(stale_bitrate.1, 9_000);3877 assert_eq!(3878 (stale_resolution.1, stale_resolution.2),3879 (reduced.width, reduced.height)3880 );38813882 // Feedback evaluation recovers all the way to the previously3883 // committed state while the old request is still awaiting readback.3884 let recovered = AdaptiveCommitPlan {3885 decision: AdaptiveBitrateDecisionV1 {3886 decision_id: 2,3887 report_id: 2,3888 target_kbps: 12_000,3889 floor_kbps: 1_000,3890 ceiling_kbps: 12_000,3891 state: AdaptiveBitrateStateV1::Increase,3892 reasons: AdaptiveBitrateReasonFlagsV1::CLEAN_RECOVERY,3893 applied: false,3894 },3895 telemetry: MediaV3TelemetrySnapshot::default(),3896 resolution_decision: Some(MotionResolutionDecision {3897 tier: ResolutionTier::Native,3898 target: native,3899 trigger: ResolutionTrigger::Recovery,3900 changed: true,3901 }),3902 force_keyframe: false,3903 };3904 assert!(coordinator.submit(recovered).is_none());39053906 // The first task times out its bitrate wait and then its resolution3907 // wait. Neither timeout changes the last acknowledged state.3908 tokio::task::yield_now().await;3909 tokio::time::advance(ENCODER_CONTROL_COMMIT_TIMEOUT).await;3910 tokio::task::yield_now().await;3911 tokio::time::advance(ENCODER_CONTROL_COMMIT_TIMEOUT).await;3912 tokio::task::yield_now().await;3913 let result = coordinator.acknowledgements.join_next().await;3914 let acknowledgement = coordinator.complete(result).unwrap();3915 assert!(!acknowledgement.plan.decision.applied);3916 assert_eq!(acknowledgement.resolution_applied, Some(false));3917 assert_eq!(coordinator.committed_bitrate_kbps, 12_000);3918 assert_eq!(coordinator.committed_resolution, Some(native));3919 assert_eq!(coordinator.requested_bitrate_kbps, 9_000);3920 assert_eq!(coordinator.requested_resolution, Some(reduced));39213922 // Starting the recovered plan must overwrite the stale persistent3923 // actuator watch even though its targets equal the old committed3924 // values. Otherwise the timed-out request would apply on resume.3925 assert!(coordinator.start_queued().is_none());3926 assert!(coordinator.has_pending());3927 let corrected_bitrate = harness3928 .requested_bitrate()3929 .expect("recovered bitrate must overwrite stale actuator intent");3930 let corrected_resolution = harness3931 .requested_resolution()3932 .expect("recovered resolution must overwrite stale actuator intent");3933 assert!(corrected_bitrate.0 > stale_bitrate.0);3934 assert!(corrected_resolution.0 > stale_resolution.0);3935 assert_eq!(corrected_bitrate.1, 12_000);3936 assert_eq!(3937 (corrected_resolution.1, corrected_resolution.2),3938 (native.width, native.height)3939 );3940 assert_eq!(coordinator.requested_bitrate_kbps, 12_000);3941 assert_eq!(coordinator.requested_resolution, Some(native));39423943 coordinator.abort_and_drain().await;3944 drop(media);3945 }39463947 3948 async fn feedback_loop_writes_every_decision_while_combined_commit_is_hung() {3949 let sessions = Arc::new(SessionRegistry::default());3950 let remote = endpoint(1);3951 let nonce = [9; 16];3952 let media = sessions3953 .claim(remote, nonce, InvitationGrants::VIEW)3954 .unwrap();3955 let harness = crate::source::EncoderControlTestHarness::new();3956 sessions3957 .install_encoder_control(remote, media.session_id, Some(harness.control.clone()))3958 .unwrap();3959 let feedback = sessions.claim_feedback(remote, nonce).unwrap();3960 let session_id = feedback.session_id;39613962 let (mut report_writer, report_reader) = tokio::io::duplex(4_096);3963 let (feedback_sender, mut feedback_receiver) =3964 tokio::sync::watch::channel(CumulativeFeedbackIngress::default());3965 let reader_task = tokio::spawn(forward_media_feedback_reports(3966 report_reader,3967 feedback_sender,3968 ));3969 let (mut decision_writer, mut decision_reader) = tokio::io::duplex(4_096);3970 let loop_sessions = Arc::clone(&sessions);3971 let loop_task = tokio::spawn(async move {3972 run_media_feedback_loop(3973 &mut decision_writer,3974 std::future::pending(),3975 remote,3976 &loop_sessions,3977 &feedback,3978 true,3979 12_000,3980 &mut feedback_receiver,3981 reader_task,3982 )3983 .await3984 });39853986 let mut observed = Vec::new();3987 for report_id in 1..=5 {3988 if report_id > 1 {3989 tokio::time::sleep(Duration::from_millis(800)).await;3990 }3991 let mut report = clean_feedback(report_id);3992 if report_id >= 2 {3993 report.flags = MediaFeedbackFlags::RESYNC_ACTIVE;3994 report.decoder_dropped_delta = 4;3995 }3996 sigil_protocol::write_media_feedback_report_v1(&mut report_writer, &report)3997 .await3998 .unwrap();3999 let decision = tokio::time::timeout(4000 Duration::from_secs(1),4001 sigil_protocol::read_adaptive_bitrate_decision_v1(&mut decision_reader),4002 )4003 .await4004 .expect("each evaluated report must be written before the hung commit timeout")4005 .unwrap()4006 .expect("feedback loop closed before writing its decision");4007 observed.push((decision.decision_id, decision.report_id));4008 }4009 assert_eq!(4010 observed,4011 (1..=5).map(|id| (id, id)).collect::<Vec<_>>(),4012 "slow bitrate and resolution readbacks must not coalesce Portal decisions"4013 );4014 assert!(4015 harness.requested_bitrate().is_some(),4016 "the adversarial run must actually have pending bitrate work"4017 );4018 assert!(4019 harness.requested_resolution().is_some(),4020 "the adversarial run must actually have pending resolution work"4021 );40224023 drop(media);4024 drop(report_writer);4025 tokio::time::timeout(Duration::from_secs(1), loop_task)4026 .await4027 .expect("session teardown must abort the hung commit promptly")4028 .unwrap()4029 .unwrap();4030 assert!(!sessions.is_active(remote, session_id));4031 }40324033 4034 async fn slow_adaptive_commit_does_not_pause_feedback_evaluation() {4035 let sessions = Arc::new(SessionRegistry::default());4036 let remote = endpoint(1);4037 let media = sessions4038 .claim(remote, [4; 16], InvitationGrants::VIEW)4039 .unwrap();4040 let harness = crate::source::EncoderControlTestHarness::new();4041 sessions4042 .install_encoder_control(remote, media.session_id, Some(harness.control.clone()))4043 .unwrap();4044 let mut coordinator = AdaptiveCommitCoordinator::new(4045 Arc::clone(&sessions),4046 remote,4047 media.session_id,4048 true,4049 12_000,4050 Some(native_dimensions()),4051 );4052 let mut controller =4053 ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());4054 let mut resolution_controller = None;4055 let observed_at = Instant::now();40564057 let seed = evaluate_commit_plan(4058 &mut controller,4059 &mut resolution_controller,4060 &clean_feedback(1),4061 observed_at,4062 );4063 assert_eq!(4064 coordinator.submit(seed).unwrap().plan.decision.decision_id,4065 14066 );40674068 let mut severe = clean_feedback(2);4069 severe.decoder_dropped_delta = 1;4070 let first = evaluate_commit_plan(4071 &mut controller,4072 &mut resolution_controller,4073 &severe,4074 observed_at + Duration::from_secs(1),4075 );4076 assert_eq!(first.decision.state, AdaptiveBitrateStateV1::Decrease);4077 assert!(coordinator.submit(first).is_none());4078 assert!(coordinator.has_pending());40794080 tokio::time::advance(Duration::from_secs(1)).await;4081 let evaluation_tick = tokio::time::Instant::now();4082 let mut newer_severe = clean_feedback(3);4083 newer_severe.decoder_dropped_delta = 1;4084 let queued = evaluate_commit_plan(4085 &mut controller,4086 &mut resolution_controller,4087 &newer_severe,4088 observed_at + Duration::from_secs(2),4089 );4090 assert_eq!(queued.decision.decision_id, 3);4091 assert_eq!(queued.decision.state, AdaptiveBitrateStateV1::Decrease);4092 assert!(coordinator.submit(queued).is_none());4093 assert_eq!(tokio::time::Instant::now(), evaluation_tick);4094 assert_eq!(coordinator.acknowledgements.len(), 1);4095 assert_eq!(4096 coordinator.queued.unwrap().decision.decision_id,4097 queued.decision.decision_id4098 );40994100 // Revision one is acknowledged only after another feedback window was4101 // evaluated. The decision retains that exact readback result.4102 harness.status.send_modify(|status| {4103 status.applied_bitrate_revision = Some(1);4104 status.applied_bitrate_kbps = Some(first.decision.target_kbps);4105 });4106 let commit_result = coordinator.acknowledgements.join_next().await;4107 let completed = coordinator.complete(commit_result).unwrap();4108 assert_eq!(completed.plan.decision.decision_id, 2);4109 assert!(completed.plan.decision.applied);4110 assert_eq!(4111 completed.applied_bitrate_kbps,4112 Some(first.decision.target_kbps)4113 );41144115 assert!(coordinator.start_queued().is_none());4116 assert!(coordinator.has_pending());4117 assert_eq!(coordinator.acknowledgements.len(), 1);41184119 let session_id = media.session_id;4120 drop(media);4121 assert!(!sessions.is_active(remote, session_id));4122 let session_end_result = coordinator.acknowledgements.join_next().await;4123 let session_end = coordinator.complete(session_end_result).unwrap();4124 assert!(session_end.generation_ended);4125 coordinator.abort_and_drain().await;4126 assert!(!coordinator.has_pending());4127 assert!(coordinator.queued.is_none());4128 assert!(coordinator.acknowledgements.is_empty());4129 }41304131 4132 async fn timing_out_commits_keep_uniform_hysteresis_and_bounded_work() {4133 let sessions = Arc::new(SessionRegistry::default());4134 let remote = endpoint(1);4135 let media = sessions4136 .claim(remote, [5; 16], InvitationGrants::VIEW)4137 .unwrap();4138 let harness = crate::source::EncoderControlTestHarness::new();4139 sessions4140 .install_encoder_control(remote, media.session_id, Some(harness.control.clone()))4141 .unwrap();4142 let mut coordinator = AdaptiveCommitCoordinator::new(4143 Arc::clone(&sessions),4144 remote,4145 media.session_id,4146 true,4147 12_000,4148 Some(native_dimensions()),4149 );4150 let mut controller =4151 ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());4152 let mut resolution_controller =4153 Some(MotionResolutionPolicy::new(native_dimensions()).unwrap());4154 let observed_at = Instant::now();4155 let virtual_started_at = tokio::time::Instant::now();4156 let mut evaluation_ticks = Vec::new();4157 let mut decision_ids = Vec::new();4158 let mut states = Vec::new();4159 let mut completed = Vec::new();41604161 for report_id in 1..=9 {4162 evaluation_ticks.push(tokio::time::Instant::now());4163 let mut report = clean_feedback(report_id);4164 if report_id > 1 {4165 report.frontend_queue_depth = 2;4166 }4167 let plan = evaluate_commit_plan(4168 &mut controller,4169 &mut resolution_controller,4170 &report,4171 observed_at + Duration::from_secs(report_id - 1),4172 );4173 decision_ids.push(plan.decision.decision_id);4174 states.push(plan.decision.state);4175 if let Some(immediate) = coordinator.submit(plan) {4176 completed.push(immediate);4177 }4178 assert!(4179 coordinator.acknowledgements.len() <= 1,4180 "the coordinator must own at most one commit task"4181 );4182 assert!(4183 coordinator.queued.iter().count() <= 1,4184 "the coordinator must retain only the latest plan"4185 );41864187 tokio::time::advance(Duration::from_secs(1)).await;4188 tokio::task::yield_now().await;4189 if let Some(result) = coordinator.acknowledgements.try_join_next() {4190 completed.push(coordinator.complete(Some(result)).unwrap());4191 if let Some(immediate) = coordinator.start_queued() {4192 completed.push(immediate);4193 }4194 }4195 }41964197 assert_eq!(4198 decision_ids,4199 (1..=9).collect::<Vec<_>>(),4200 "coalescing must not pause or reuse decision IDs"4201 );4202 assert_eq!(4203 states,4204 [4205 AdaptiveBitrateStateV1::Hold,4206 AdaptiveBitrateStateV1::Hold,4207 AdaptiveBitrateStateV1::Decrease,4208 AdaptiveBitrateStateV1::Hold,4209 AdaptiveBitrateStateV1::Decrease,4210 AdaptiveBitrateStateV1::Hold,4211 AdaptiveBitrateStateV1::Decrease,4212 AdaptiveBitrateStateV1::Hold,4213 AdaptiveBitrateStateV1::Decrease,4214 ],4215 "moderate-pressure hysteresis must advance once per feedback window"4216 );4217 for (index, tick) in evaluation_ticks.into_iter().enumerate() {4218 assert_eq!(4219 tick.duration_since(virtual_started_at),4220 Duration::from_secs(index as u64),4221 "feedback evaluation cadence must remain uniform during commit timeouts"4222 );4223 }4224 assert!(4225 completed.iter().any(|acknowledgement| {4226 acknowledgement.plan.decision.decision_id == 34227 && !acknowledgement.plan.decision.applied4228 && acknowledgement.resolution_applied == Some(false)4229 }),4230 "the first bitrate and resolution timeout must retain exact negative acknowledgement"4231 );4232 assert!(coordinator.has_pending());4233 assert_eq!(coordinator.acknowledgements.len(), 1);4234 assert!(coordinator.queued.is_some());42354236 coordinator.abort_and_drain().await;4237 assert!(!coordinator.has_pending());4238 assert!(coordinator.queued.is_none());4239 assert!(coordinator.acknowledgements.is_empty());42404241 let session_id = media.session_id;4242 drop(media);4243 assert!(!sessions.is_active(remote, session_id));4244 }4245}