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}603604impl std::fmt::Debug for GenerationAdaptiveCoordinator {605 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {606 formatter607 .debug_struct("GenerationAdaptiveCoordinator")608 .field("generation_id", &self.generation_id)609 .field("ceiling_kbps", &self.ceiling_kbps)610 .finish_non_exhaustive()611 }612}613614pub(super) struct GenerationFeedbackRegistration {615 coordinator: Arc<GenerationAdaptiveCoordinator>,616 key: GenerationContributorKey,617 authorization_revision: u64,618 pub(super) decisions: tokio::sync::watch::Receiver<Option<AdaptiveBitrateDecisionV1>>,619 pub(super) detach: tokio::sync::watch::Receiver<bool>,620}621622impl GenerationFeedbackRegistration {623 pub(super) fn observe(&self, report: MediaFeedbackReportV1) -> Result<()> {624 report.validate()?;625 let mut state = self626 .coordinator627 .state628 .lock()629 .map_err(|_| anyhow::anyhow!("generation adaptive state poisoned"))?;630 let contributor = state631 .contributors632 .get_mut(&self.key)633 .filter(|contributor| contributor.authorization_revision == self.authorization_revision)634 .context("feedback contributor no longer matches the active viewer generation")?;635 contributor.ingress.observe(report);636 drop(state);637 self.coordinator.changed.notify_one();638 Ok(())639 }640}641642impl Drop for GenerationFeedbackRegistration {643 fn drop(&mut self) {644 self.coordinator645 .unregister(self.key, self.authorization_revision);646 }647}648649impl GenerationAdaptiveCoordinator {650 pub(super) fn start(651 config: HostConfig,652 generation_id: u64,653 encoder_control: Option<EncoderControl>,654 producer_telemetry: Arc<super::session::MediaV3Telemetry>,655 keyframe_requests: tokio::sync::watch::Sender<Option<MediaControlRequestV3>>,656 shutdown: tokio::sync::watch::Receiver<bool>,657 recovering_viewers: Arc<AtomicUsize>,658 ) -> Result<(Arc<Self>, tokio::task::JoinHandle<Result<()>>)> {659 let ceiling_kbps = adaptive_bitrate_ceiling_kbps(&config)?;660 let coordinator = Arc::new(Self {661 generation_id,662 ceiling_kbps,663 state: Mutex::new(GenerationAggregateState {664 contributors: HashMap::with_capacity(crate::config::MAX_VIEWERS.into()),665 next_decision_id: 1,666 target_kbps: ceiling_kbps,667 }),668 changed: tokio::sync::Notify::new(),669 recovering_viewers,670 });671 let task_coordinator = Arc::clone(&coordinator);672 let task = tokio::spawn(async move {673 task_coordinator674 .run(675 config,676 encoder_control,677 producer_telemetry,678 keyframe_requests,679 shutdown,680 )681 .await682 });683 Ok((coordinator, task))684 }685686 pub(super) fn register(687 self: &Arc<Self>,688 remote: EndpointId,689 session_id: u64,690 authorization_revision: u64,691 media_generation_id: u64,692 telemetry: Arc<super::session::MediaV3Telemetry>,693 ) -> Result<GenerationFeedbackRegistration> {694 ensure!(695 media_generation_id == self.generation_id,696 "feedback media generation does not match the shared generation"697 );698 ensure!(699 authorization_revision != 0,700 "feedback authorization revision must be non-zero"701 );702 let key = GenerationContributorKey { remote, session_id };703 let (decisions, decision_rx) = tokio::sync::watch::channel(None);704 let (detach, detach_rx) = tokio::sync::watch::channel(false);705 let mut state = self706 .state707 .lock()708 .map_err(|_| anyhow::anyhow!("generation adaptive state poisoned"))?;709 ensure!(710 state.contributors.len() < usize::from(crate::config::MAX_VIEWERS)711 || state.contributors.contains_key(&key),712 "generation feedback contributor capacity is full"713 );714 state.contributors.insert(715 key,716 GenerationContributor {717 authorization_revision,718 ingress: CumulativeFeedbackIngress::default(),719 cursor: FeedbackIngressCursor::default(),720 controller: ShadowBitrateController::new(self.ceiling_kbps, telemetry.snapshot()),721 latest_report: None,722 latest_evaluation: None,723 last_received_at: None,724 recovery: ContributorRecoveryState::default(),725 telemetry,726 decisions,727 detach,728 },729 );730 self.update_recovering_count(&state);731 drop(state);732 self.changed.notify_one();733 Ok(GenerationFeedbackRegistration {734 coordinator: Arc::clone(self),735 key,736 authorization_revision,737 decisions: decision_rx,738 detach: detach_rx,739 })740 }741742 fn unregister(&self, key: GenerationContributorKey, authorization_revision: u64) {743 let Ok(mut state) = self.state.lock() else {744 return;745 };746 if state747 .contributors748 .get(&key)749 .is_some_and(|contributor| contributor.authorization_revision == authorization_revision)750 {751 state.contributors.remove(&key);752 self.update_recovering_count(&state);753 drop(state);754 self.changed.notify_one();755 }756 }757758 async fn run(759 self: Arc<Self>,760 config: HostConfig,761 encoder_control: Option<EncoderControl>,762 producer_telemetry: Arc<super::session::MediaV3Telemetry>,763 keyframe_requests: tokio::sync::watch::Sender<Option<MediaControlRequestV3>>,764 mut shutdown: tokio::sync::watch::Receiver<bool>,765 ) -> Result<()> {766 let actuation_enabled =767 adaptive_bitrate_actuation_enabled(&config) && encoder_control.is_some();768 let mut resolution_controller = if actuation_enabled {769 encoder_control770 .as_ref()771 .map(motion_resolution_policy_for_encoder)772 .transpose()?773 .flatten()774 } else {775 None776 };777 let initial_resolution = resolution_controller778 .as_ref()779 .map(MotionResolutionPolicy::target);780 let mut actuator = GenerationAdaptiveActuator::new(781 encoder_control,782 actuation_enabled,783 self.ceiling_kbps,784 initial_resolution,785 );786 let mut tick = tokio::time::interval(AGGREGATE_TICK_INTERVAL);787 tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);788 loop {789 tokio::select! {790 biased;791 changed = shutdown.changed() => {792 if changed.is_err() || *shutdown.borrow_and_update() {793 break;794 }795 }796 acknowledgement = actuator.acknowledgements.join_next(), if actuator.pending => {797 actuator.complete(acknowledgement)?;798 actuator.start_queued();799 }800 _ = self.changed.notified() => {}801 _ = tick.tick() => {}802 }803 if let Some(update) = self.evaluate(804 Instant::now(),805 producer_telemetry.snapshot(),806 actuator.bitrate_is_applied(),807 resolution_controller.as_mut(),808 )? {809 if update.request_recovery {810 let request_id = update.decision_id;811 keyframe_requests.send_replace(Some(MediaControlRequestV3::request_keyframe(812 request_id,813 update.last_sequence,814 KeyframeRequestReasonV3::FrontendBackpressure,815 )));816 }817 actuator.submit(update.plan);818 }819 }820 actuator.abort_and_drain().await;821 Ok(())822 }823824 fn evaluate(825 &self,826 now: Instant,827 producer_telemetry: MediaV3TelemetrySnapshot,828 bitrate_applied: bool,829 resolution_controller: Option<&mut MotionResolutionPolicy>,830 ) -> Result<Option<GenerationAggregateUpdate>> {831 let mut state = self832 .state833 .lock()834 .map_err(|_| anyhow::anyhow!("generation adaptive state poisoned"))?;835 let mut changed_keys = Vec::new();836 let mut request_recovery = false;837 for (key, contributor) in &mut state.contributors {838 let Some((report, next_cursor)) =839 contributor.ingress.report_since(contributor.cursor)?840 else {841 continue;842 };843 contributor.cursor = next_cursor;844 // Configured GOP supersession and producer-wide send state are not845 // charged to every viewer as local transport pressure. The shared846 // values remain available to generation diagnostics and actuation.847 let telemetry = contributor.telemetry.snapshot();848 let evaluation = contributor.controller.evaluate(&report, telemetry, now)?;849 match contributor.recovery.observe(&evaluation, now) {850 RecoveryMutation::RequestRecovery => request_recovery = true,851 RecoveryMutation::Detach => {852 contributor.detach.send_replace(true);853 }854 RecoveryMutation::None | RecoveryMutation::Recovered => {}855 }856 contributor.latest_report = Some(report);857 contributor.latest_evaluation = Some(evaluation);858 contributor.last_received_at = Some(now);859 changed_keys.push(*key);860 }861 self.update_recovering_count(&state);862863 let active = state864 .contributors865 .values()866 .filter(|contributor| {867 contributor.recovery.contributes()868 && contributor.last_received_at.is_some_and(|received| {869 now.saturating_duration_since(received) <= FEEDBACK_STALE_ARRIVAL_GAP870 })871 && contributor872 .latest_evaluation873 .is_some_and(|evaluation| evaluation.fresh)874 })875 .filter_map(|contributor| contributor.latest_evaluation.zip(contributor.latest_report))876 .collect::<Vec<_>>();877 if changed_keys.is_empty() && active.is_empty() {878 return Ok(None);879 }880 let (target_kbps, all_clean) =881 aggregate_target_and_clean(&active, state.target_kbps, self.ceiling_kbps);882 let previous_target = state.target_kbps;883 if changed_keys.is_empty() && target_kbps == previous_target {884 return Ok(None);885 }886 state.target_kbps = target_kbps;887 let global_state = match target_kbps.cmp(&previous_target) {888 std::cmp::Ordering::Less => AdaptiveBitrateStateV1::Decrease,889 std::cmp::Ordering::Greater => AdaptiveBitrateStateV1::Increase,890 std::cmp::Ordering::Equal => AdaptiveBitrateStateV1::Hold,891 };892 let worst = active893 .iter()894 .max_by_key(|(evaluation, _)| feedback_severity_rank(evaluation.pressure_severity));895 let (aggregate_evaluation, aggregate_report, aggregate_reasons) = match worst {896 Some((evaluation, report)) => {897 let mut aggregate = *evaluation;898 aggregate.clean_recovery_evidence = all_clean;899 (Some(aggregate), Some(*report), evaluation.decision.reasons)900 }901 None => (None, None, AdaptiveBitrateReasonFlagsV1::FEEDBACK_STALE),902 };903 let resolution_decision = resolution_controller.and_then(|policy| {904 aggregate_evaluation905 .zip(aggregate_report)906 .map(|(evaluation, report)| policy.observe_window(&report, &evaluation, now))907 });908 let plan_decision_id = state.next_decision_id;909 let plan_report_id = aggregate_report.map_or(1, |report| report.report_id);910 let plan = AdaptiveCommitPlan {911 decision: AdaptiveBitrateDecisionV1 {912 decision_id: plan_decision_id,913 report_id: plan_report_id,914 target_kbps,915 floor_kbps: ADAPTIVE_BITRATE_FLOOR_KBPS,916 ceiling_kbps: self.ceiling_kbps,917 state: global_state,918 reasons: aggregate_reasons,919 applied: bitrate_applied,920 },921 telemetry: producer_telemetry,922 resolution_decision,923 force_keyframe: request_recovery924 || aggregate_report925 .is_some_and(|report| report.flags.contains(MediaFeedbackFlags::RESYNC_ACTIVE)),926 };927928 for key in changed_keys {929 let Some((evaluation, decisions)) =930 state.contributors.get(&key).and_then(|contributor| {931 contributor932 .latest_evaluation933 .map(|evaluation| (evaluation, contributor.decisions.clone()))934 })935 else {936 continue;937 };938 let decision_id = state.next_decision_id;939 state.next_decision_id = state940 .next_decision_id941 .checked_add(1)942 .context("generation adaptive decision ID exhausted")?;943 let decision = AdaptiveBitrateDecisionV1 {944 decision_id,945 report_id: evaluation.decision.report_id,946 target_kbps,947 floor_kbps: ADAPTIVE_BITRATE_FLOOR_KBPS,948 ceiling_kbps: self.ceiling_kbps,949 state: global_state,950 reasons: evaluation.decision.reasons,951 applied: bitrate_applied,952 };953 decision.validate()?;954 decisions.send_replace(Some(decision));955 }956 plan.decision.validate()?;957 Ok(Some(GenerationAggregateUpdate {958 plan,959 request_recovery,960 decision_id: plan_decision_id,961 last_sequence: aggregate_report.and_then(|report| report.last_sequence),962 }))963 }964965 fn update_recovering_count(&self, state: &GenerationAggregateState) {966 let count = state967 .contributors968 .values()969 .filter(|contributor| {970 !matches!(971 contributor.recovery,972 ContributorRecoveryState::Active { .. }973 )974 })975 .count();976 debug_assert!(count <= usize::from(crate::config::MAX_VIEWERS));977 self.recovering_viewers.store(count, Ordering::Relaxed);978 }979}980981fn feedback_severity_rank(severity: FeedbackSeverity) -> u8 {982 match severity {983 FeedbackSeverity::Clean => 0,984 FeedbackSeverity::InsufficientEvidence => 1,985 FeedbackSeverity::Moderate => 2,986 FeedbackSeverity::Severe => 3,987 FeedbackSeverity::Stale => 4,988 }989}990991fn aggregate_target_and_clean(992 active: &[(AdaptiveBitrateEvaluation, MediaFeedbackReportV1)],993 current_target_kbps: u32,994 ceiling_kbps: u32,995) -> (u32, bool) {996 let target = active997 .iter()998 .map(|(evaluation, _)| evaluation.decision.target_kbps)999 .min()1000 .unwrap_or(current_target_kbps)1001 .clamp(ADAPTIVE_BITRATE_FLOOR_KBPS, ceiling_kbps);1002 let all_clean = !active.is_empty()1003 && active.iter().all(|(evaluation, _)| {1004 evaluation.pressure_severity == FeedbackSeverity::Clean1005 && evaluation.clean_recovery_evidence1006 });1007 (target, all_clean)1008}10091010struct GenerationAggregateUpdate {1011 plan: AdaptiveCommitPlan,1012 request_recovery: bool,1013 decision_id: u64,1014 last_sequence: Option<u64>,1015}101610171018struct ShadowBitrateController {1019 floor_kbps: u32,1020 ceiling_kbps: u32,1021 target_kbps: u32,1022 next_decision_id: u64,1023 last_report_id: Option<u64>,1024 last_report_received_at: Option<Instant>,1025 previous_telemetry: MediaV3TelemetrySnapshot,1026 baseline_rtt_micros: Option<u64>,1027 last_sequence: Option<u64>,1028 moderate_windows: u8,1029 clean_windows: u8,1030 cooldown_windows: u8,1031}103210331034struct AdaptiveBitrateEvaluation {1035 decision: AdaptiveBitrateDecisionV1,1036 pressure_severity: FeedbackSeverity,1037 clean_recovery_evidence: bool,1038 fresh: bool,1039}10401041impl ShadowBitrateController {1042 fn new(configured_ceiling_kbps: u32, telemetry: MediaV3TelemetrySnapshot) -> Self {1043 let ceiling_kbps = quantize_down(configured_ceiling_kbps).max(ADAPTIVE_BITRATE_FLOOR_KBPS);1044 Self {1045 floor_kbps: ADAPTIVE_BITRATE_FLOOR_KBPS,1046 ceiling_kbps,1047 target_kbps: ceiling_kbps,1048 next_decision_id: 1,1049 last_report_id: None,1050 last_report_received_at: None,1051 previous_telemetry: telemetry,1052 baseline_rtt_micros: None,1053 last_sequence: None,1054 moderate_windows: 0,1055 clean_windows: 0,1056 cooldown_windows: 0,1057 }1058 }10591060 fn evaluate(1061 &mut self,1062 report: &MediaFeedbackReportV1,1063 telemetry: MediaV3TelemetrySnapshot,1064 received_at: Instant,1065 ) -> Result<AdaptiveBitrateEvaluation> {1066 ensure!(1067 self.last_report_id1068 .is_none_or(|last| report.report_id > last),1069 "feedback report IDs must increase monotonically"1070 );1071 let (severity, pressure_severity, reasons, stale, trusted_host_pressure) =1072 self.classify(report, telemetry, received_at);1073 let mut state = AdaptiveBitrateStateV1::Hold;10741075 if self.cooldown_windows > 0 {1076 self.cooldown_windows -= 1;1077 }1078 match severity {1079 FeedbackSeverity::Severe => {1080 self.moderate_windows = 0;1081 self.clean_windows = 0;1082 self.cooldown_windows = ADAPTIVE_BITRATE_COOLDOWN_WINDOWS;1083 let next =1084 quantize_down(self.target_kbps.saturating_mul(3) / 4).max(self.floor_kbps);1085 if next < self.target_kbps {1086 self.target_kbps = next;1087 state = AdaptiveBitrateStateV1::Decrease;1088 }1089 }1090 FeedbackSeverity::Moderate => {1091 self.moderate_windows = self.moderate_windows.saturating_add(1);1092 self.clean_windows = 0;1093 if self.moderate_windows >= ADAPTIVE_BITRATE_MODERATE_WINDOWS {1094 self.moderate_windows = 0;1095 self.cooldown_windows = ADAPTIVE_BITRATE_COOLDOWN_WINDOWS;1096 let next =1097 quantize_down(self.target_kbps.saturating_mul(4) / 5).max(self.floor_kbps);1098 if next < self.target_kbps {1099 self.target_kbps = next;1100 state = AdaptiveBitrateStateV1::Decrease;1101 }1102 }1103 }1104 FeedbackSeverity::Clean => {1105 self.moderate_windows = 0;1106 self.clean_windows = self.clean_windows.saturating_add(1);1107 if self.clean_windows >= ADAPTIVE_BITRATE_CLEAN_WINDOWS1108 && self.cooldown_windows == 01109 {1110 self.clean_windows = 0;1111 let maximum_step =1112 (self.target_kbps / 20).clamp(ADAPTIVE_BITRATE_QUANTUM_KBPS, 500);1113 let next = quantize_down(1114 self.target_kbps1115 .saturating_add(maximum_step)1116 .min(self.ceiling_kbps),1117 );1118 if next > self.target_kbps {1119 self.target_kbps = next;1120 state = AdaptiveBitrateStateV1::Increase;1121 }1122 }1123 }1124 FeedbackSeverity::InsufficientEvidence | FeedbackSeverity::Stale => {1125 self.moderate_windows = 0;1126 self.clean_windows = 0;1127 }1128 }11291130 self.target_kbps = self.target_kbps.clamp(self.floor_kbps, self.ceiling_kbps);1131 let decision = AdaptiveBitrateDecisionV1 {1132 decision_id: self.next_decision_id,1133 report_id: report.report_id,1134 target_kbps: self.target_kbps,1135 floor_kbps: self.floor_kbps,1136 ceiling_kbps: self.ceiling_kbps,1137 state,1138 reasons,1139 applied: false,1140 };1141 decision.validate()?;1142 self.next_decision_id = self1143 .next_decision_id1144 .checked_add(1)1145 .context("adaptive bitrate decision ID exhausted")?;1146 self.last_report_id = Some(report.report_id);1147 self.last_report_received_at = Some(received_at);1148 if !stale1149 || !trusted_host_pressure1150 || state == AdaptiveBitrateStateV1::Decrease1151 || self.target_kbps == self.floor_kbps1152 {1153 self.previous_telemetry = telemetry;1154 }1155 if !stale && let Some(sequence) = report.last_sequence {1156 self.last_sequence = Some(1157 self.last_sequence1158 .map_or(sequence, |last| last.max(sequence)),1159 );1160 }1161 Ok(AdaptiveBitrateEvaluation {1162 decision,1163 pressure_severity,1164 clean_recovery_evidence: severity == FeedbackSeverity::Clean,1165 fresh: !stale,1166 })1167 }11681169 1170 fn decide(1171 &mut self,1172 report: &MediaFeedbackReportV1,1173 telemetry: MediaV3TelemetrySnapshot,1174 received_at: Instant,1175 ) -> Result<AdaptiveBitrateDecisionV1> {1176 self.evaluate(report, telemetry, received_at)1177 .map(|evaluation| evaluation.decision)1178 }11791180 fn classify(1181 &mut self,1182 report: &MediaFeedbackReportV1,1183 telemetry: MediaV3TelemetrySnapshot,1184 received_at: Instant,1185 ) -> (1186 FeedbackSeverity,1187 FeedbackSeverity,1188 AdaptiveBitrateReasonFlagsV1,1189 bool,1190 bool,1191 ) {1192 let stale = !(FEEDBACK_FRESH_INTERVAL_MIN_MS..=FEEDBACK_FRESH_INTERVAL_MAX_MS)1193 .contains(&report.interval_ms)1194 || self.last_report_received_at.is_none()1195 || self.last_report_received_at.is_some_and(|last| {1196 let gap = received_at.saturating_duration_since(last);1197 !(FEEDBACK_FRESH_ARRIVAL_GAP_MIN..=FEEDBACK_STALE_ARRIVAL_GAP).contains(&gap)1198 });1199 let mut severity = FeedbackSeverity::Clean;1200 let mut reasons = AdaptiveBitrateReasonFlagsV1::NONE;12011202 let lost_delta = telemetry1203 .selected_path_lost_packets1204 .saturating_sub(self.previous_telemetry.selected_path_lost_packets);1205 let congestion_delta = telemetry1206 .selected_path_congestion_events1207 .saturating_sub(self.previous_telemetry.selected_path_congestion_events);1208 let cancellation_delta = telemetry1209 .scheduler_cancellations1210 .saturating_sub(self.previous_telemetry.scheduler_cancellations);1211 let failure_delta = telemetry1212 .send_failures1213 .saturating_sub(self.previous_telemetry.send_failures);12141215 if failure_delta > 0 || lost_delta > 0 || cancellation_delta >= 2 {1216 add_feedback_signal(1217 &mut severity,1218 &mut reasons,1219 FeedbackSeverity::Severe,1220 AdaptiveBitrateReasonFlagsV1::LOSS_OR_CANCELLATION,1221 );1222 } else if cancellation_delta > 0 || congestion_delta > 0 {1223 add_feedback_signal(1224 &mut severity,1225 &mut reasons,1226 FeedbackSeverity::Moderate,1227 AdaptiveBitrateReasonFlagsV1::LOSS_OR_CANCELLATION,1228 );1229 }1230 if cancellation_delta >= 2 || failure_delta > 0 {1231 add_feedback_signal(1232 &mut severity,1233 &mut reasons,1234 FeedbackSeverity::Severe,1235 AdaptiveBitrateReasonFlagsV1::SENDER_BACKPRESSURE,1236 );1237 } else if cancellation_delta > 0 {1238 add_feedback_signal(1239 &mut severity,1240 &mut reasons,1241 FeedbackSeverity::Moderate,1242 AdaptiveBitrateReasonFlagsV1::SENDER_BACKPRESSURE,1243 );1244 }12451246 let rtt = telemetry.selected_path_rtt_micros;1247 if rtt > 0 {1248 let baseline = self.baseline_rtt_micros.get_or_insert(rtt);1249 if rtt < *baseline {1250 *baseline = rtt;1251 } else if rtt >= baseline.saturating_mul(2).max(80_000) {1252 add_feedback_signal(1253 &mut severity,1254 &mut reasons,1255 FeedbackSeverity::Severe,1256 AdaptiveBitrateReasonFlagsV1::RTT_INFLATION,1257 );1258 } else if rtt >= (baseline.saturating_mul(3) / 2).max(40_000) {1259 add_feedback_signal(1260 &mut severity,1261 &mut reasons,1262 FeedbackSeverity::Moderate,1263 AdaptiveBitrateReasonFlagsV1::RTT_INFLATION,1264 );1265 }1266 }12671268 let trusted_host_pressure = severity != FeedbackSeverity::Clean;1269 if stale {1270 reasons = reasons.union(AdaptiveBitrateReasonFlagsV1::FEEDBACK_STALE);1271 if !trusted_host_pressure {1272 severity = FeedbackSeverity::Stale;1273 }1274 return (severity, severity, reasons, true, trusted_host_pressure);1275 }12761277 let receiver_drops = report1278 .transport_dropped_delta1279 .saturating_add(report.frontend_dropped_delta)1280 .saturating_add(report.decoder_dropped_delta)1281 .saturating_add(report.presenter_dropped_delta);1282 let queue_severe = queue_ratio_at_least(1283 report.frontend_queue_depth,1284 report.frontend_queue_capacity,1285 3,1286 4,1287 ) || queue_ratio_at_least(1288 report.decode_queue_depth,1289 report.decode_queue_capacity,1290 3,1291 4,1292 ) || queue_ratio_at_least(1293 report.presenter_queue_depth,1294 report.presenter_queue_capacity,1295 3,1296 4,1297 );1298 let queue_moderate = queue_ratio_at_least(1299 report.frontend_queue_depth,1300 report.frontend_queue_capacity,1301 1,1302 2,1303 ) || queue_ratio_at_least(1304 report.decode_queue_depth,1305 report.decode_queue_capacity,1306 1,1307 2,1308 ) || queue_ratio_at_least(1309 report.presenter_queue_depth,1310 report.presenter_queue_capacity,1311 1,1312 2,1313 );1314 if report.flags.contains(MediaFeedbackFlags::RESYNC_ACTIVE)1315 || queue_severe1316 || receiver_drops >= 41317 {1318 add_feedback_signal(1319 &mut severity,1320 &mut reasons,1321 FeedbackSeverity::Severe,1322 AdaptiveBitrateReasonFlagsV1::RECEIVER_QUEUE,1323 );1324 } else if queue_moderate || receiver_drops > 0 {1325 add_feedback_signal(1326 &mut severity,1327 &mut reasons,1328 FeedbackSeverity::Moderate,1329 AdaptiveBitrateReasonFlagsV1::RECEIVER_QUEUE,1330 );1331 }1332 if report.decode_queue_depth == report.decode_queue_capacity1333 || report.decoder_dropped_delta > 01334 {1335 add_feedback_signal(1336 &mut severity,1337 &mut reasons,1338 FeedbackSeverity::Severe,1339 AdaptiveBitrateReasonFlagsV1::DECODE_BACKLOG,1340 );1341 } else if report.decode_queue_depth >= 2 {1342 add_feedback_signal(1343 &mut severity,1344 &mut reasons,1345 FeedbackSeverity::Moderate,1346 AdaptiveBitrateReasonFlagsV1::DECODE_BACKLOG,1347 );1348 }13491350 let delivery_p95 = report.transport_delivery_p95_ms.unwrap_or(0);1351 let decode_p95 = report.decode_p95_ms.unwrap_or(0);1352 let presentation_p95 = report.presentation_p95_ms.unwrap_or(0);1353 if delivery_p95 >= 200 || decode_p95 >= 100 || presentation_p95 >= 100 {1354 add_feedback_signal(1355 &mut severity,1356 &mut reasons,1357 FeedbackSeverity::Severe,1358 AdaptiveBitrateReasonFlagsV1::DELIVERY_LATENCY,1359 );1360 } else if delivery_p95 >= 100 || decode_p95 >= 50 || presentation_p95 >= 50 {1361 add_feedback_signal(1362 &mut severity,1363 &mut reasons,1364 FeedbackSeverity::Moderate,1365 AdaptiveBitrateReasonFlagsV1::DELIVERY_LATENCY,1366 );1367 }13681369 let pressure_severity = severity;1370 if severity == FeedbackSeverity::Clean {1371 let sequence_progressed = self1372 .last_sequence1373 .zip(report.last_sequence)1374 .is_some_and(|(last, current)| current > last);1375 let complete_latency = report.transport_delivery_p95_ms.is_some()1376 && report.decode_p95_ms.is_some()1377 && report.presentation_p95_ms.is_some();1378 if sequence_progressed && complete_latency {1379 if self.target_kbps < self.ceiling_kbps {1380 reasons = reasons.union(AdaptiveBitrateReasonFlagsV1::CLEAN_RECOVERY);1381 }1382 } else {1383 severity = FeedbackSeverity::InsufficientEvidence;1384 }1385 }1386 (1387 severity,1388 pressure_severity,1389 reasons,1390 false,1391 trusted_host_pressure,1392 )1393 }1394}13951396fn add_feedback_signal(1397 severity: &mut FeedbackSeverity,1398 reasons: &mut AdaptiveBitrateReasonFlagsV1,1399 signal: FeedbackSeverity,1400 reason: AdaptiveBitrateReasonFlagsV1,1401) {1402 *reasons = reasons.union(reason);1403 if signal == FeedbackSeverity::Severe1404 || (*severity == FeedbackSeverity::Clean && signal == FeedbackSeverity::Moderate)1405 {1406 *severity = signal;1407 }1408}14091410fn quantize_down(kbps: u32) -> u32 {1411 kbps / ADAPTIVE_BITRATE_QUANTUM_KBPS * ADAPTIVE_BITRATE_QUANTUM_KBPS1412}14131414fn queue_ratio_at_least(depth: u8, capacity: u8, numerator: u16, denominator: u16) -> bool {1415 capacity > 01416 && u16::from(depth).saturating_mul(denominator)1417 >= u16::from(capacity).saturating_mul(numerator)1418}141914201421struct AdaptiveCommitPlan {1422 decision: AdaptiveBitrateDecisionV1,1423 telemetry: MediaV3TelemetrySnapshot,1424 resolution_decision: Option<MotionResolutionDecision>,1425 force_keyframe: bool,1426}142714281429struct AdaptiveCommitAcknowledgement {1430 plan: AdaptiveCommitPlan,1431 generation_ended: bool,1432 recovery_acknowledged: Option<bool>,1433 resolution_applied: Option<bool>,1434 applied_bitrate_kbps: Option<u32>,1435 applied_resolution: Option<VideoDimensions>,1436}14371438impl AdaptiveCommitAcknowledgement {1439 fn immediate(plan: AdaptiveCommitPlan) -> Self {1440 Self {1441 plan,1442 generation_ended: false,1443 recovery_acknowledged: None,1444 resolution_applied: None,1445 applied_bitrate_kbps: None,1446 applied_resolution: None,1447 }1448 }1449}14501451/// Bounded adaptive-control actuator.1452///1453/// Feedback evaluation remains synchronous and deterministic, while this1454/// coordinator owns the slower encoder acknowledgements. At most one commit1455/// task and one latest coalesced plan exist. A queued plan carries the latest1456/// desired bitrate and resolution even when its feedback decision is a hold,1457/// so a slow or failed predecessor cannot strand an uncommitted target.1458struct AdaptiveCommitCoordinator {1459 sessions: Arc<SessionRegistry>,1460 remote: EndpointId,1461 session_id: u64,1462 actuation_enabled: bool,1463 desired_bitrate_kbps: u32,1464 requested_bitrate_kbps: u32,1465 committed_bitrate_kbps: u32,1466 desired_resolution: Option<VideoDimensions>,1467 requested_resolution: Option<VideoDimensions>,1468 committed_resolution: Option<VideoDimensions>,1469 pending: bool,1470 queued: Option<AdaptiveCommitPlan>,1471 acknowledgements: tokio::task::JoinSet<AdaptiveCommitAcknowledgement>,1472}147314741475struct GenerationAdaptiveAcknowledgement {1476 bitrate_kbps: Option<u32>,1477 resolution: Option<VideoDimensions>,1478}14791480struct GenerationAdaptiveActuator {1481 control: Option<EncoderControl>,1482 actuation_enabled: bool,1483 desired_bitrate_kbps: u32,1484 requested_bitrate_kbps: u32,1485 committed_bitrate_kbps: u32,1486 desired_resolution: Option<VideoDimensions>,1487 requested_resolution: Option<VideoDimensions>,1488 committed_resolution: Option<VideoDimensions>,1489 pending: bool,1490 queued: Option<AdaptiveCommitPlan>,1491 acknowledgements: tokio::task::JoinSet<Result<GenerationAdaptiveAcknowledgement>>,1492}14931494impl GenerationAdaptiveActuator {1495 fn new(1496 control: Option<EncoderControl>,1497 actuation_enabled: bool,1498 bitrate_kbps: u32,1499 resolution: Option<VideoDimensions>,1500 ) -> Self {1501 Self {1502 control,1503 actuation_enabled,1504 desired_bitrate_kbps: bitrate_kbps,1505 requested_bitrate_kbps: bitrate_kbps,1506 committed_bitrate_kbps: bitrate_kbps,1507 desired_resolution: resolution,1508 requested_resolution: resolution,1509 committed_resolution: resolution,1510 pending: false,1511 queued: None,1512 acknowledgements: tokio::task::JoinSet::new(),1513 }1514 }15151516 fn bitrate_is_applied(&self) -> bool {1517 self.actuation_enabled1518 && !self.pending1519 && self.desired_bitrate_kbps == self.committed_bitrate_kbps1520 }15211522 fn submit(&mut self, mut plan: AdaptiveCommitPlan) {1523 self.desired_bitrate_kbps = plan.decision.target_kbps;1524 if let Some(resolution) = plan.resolution_decision {1525 self.desired_resolution = Some(resolution.target);1526 }1527 if self.pending {1528 if let Some(queued) = self.queued {1529 plan.force_keyframe |= queued.force_keyframe;1530 }1531 self.queued = Some(plan);1532 return;1533 }1534 self.start(plan);1535 }15361537 fn start_queued(&mut self) {1538 if let Some(plan) = self.queued.take() {1539 self.start(plan);1540 }1541 }15421543 fn start(&mut self, plan: AdaptiveCommitPlan) {1544 if !self.actuation_enabled {1545 return;1546 }1547 let Some(control) = self.control.clone() else {1548 return;1549 };1550 let bitrate_target = plan.decision.target_kbps;1551 let bitrate_revision = if bitrate_target != self.requested_bitrate_kbps1552 || bitrate_target != self.committed_bitrate_kbps1553 {1554 match control.request_bitrate_kbps(bitrate_target) {1555 Ok(revision) => {1556 self.requested_bitrate_kbps = bitrate_target;1557 Some(revision)1558 }1559 Err(error) => {1560 warn!(%error, bitrate_target, "generation adaptive bitrate request failed");1561 None1562 }1563 }1564 } else {1565 None1566 };1567 let resolution_target = plan.resolution_decision.map(|decision| decision.target);1568 let resolution_revision = if let Some(target) = resolution_target.filter(|target| {1569 Some(*target) != self.requested_resolution || Some(*target) != self.committed_resolution1570 }) {1571 match control.request_resolution(target.width, target.height) {1572 Ok(revision) => {1573 self.requested_resolution = Some(target);1574 Some((revision, target))1575 }1576 Err(error) => {1577 warn!(%error, width = target.width, height = target.height, "generation adaptive resolution request failed");1578 None1579 }1580 }1581 } else {1582 None1583 };1584 if bitrate_revision.is_none() && resolution_revision.is_none() {1585 return;1586 }1587 self.pending = true;1588 self.acknowledgements.spawn(async move {1589 let bitrate_kbps = if let Some(revision) = bitrate_revision {1590 let status = control1591 .wait_for_bitrate_applied(revision, ENCODER_CONTROL_COMMIT_TIMEOUT)1592 .await?;1593 ensure!(1594 status.applied_bitrate_revision == Some(revision)1595 && status.applied_bitrate_kbps == Some(bitrate_target),1596 "generation bitrate acknowledgement did not match revision {revision} at {bitrate_target} kbps"1597 );1598 Some(bitrate_target)1599 } else {1600 None1601 };1602 let resolution = if let Some((revision, target)) = resolution_revision {1603 let status = control1604 .wait_for_resolution_applied(1605 revision,1606 target.width,1607 target.height,1608 ENCODER_CONTROL_COMMIT_TIMEOUT,1609 )1610 .await?;1611 ensure!(1612 status.applied_resolution_revision == Some(revision)1613 && status.applied_width == Some(target.width)1614 && status.applied_height == Some(target.height),1615 "generation resolution acknowledgement did not match revision {revision} at {}x{}",1616 target.width,1617 target.height1618 );1619 Some(target)1620 } else {1621 None1622 };1623 Ok(GenerationAdaptiveAcknowledgement {1624 bitrate_kbps,1625 resolution,1626 })1627 });1628 }16291630 fn complete(1631 &mut self,1632 result: Option<Result<Result<GenerationAdaptiveAcknowledgement>, tokio::task::JoinError>>,1633 ) -> Result<()> {1634 ensure!(1635 self.pending,1636 "generation adaptive acknowledgement was not pending"1637 );1638 self.pending = false;1639 match result1640 .context("generation adaptive task ended without a result")?1641 .context("generation adaptive task failed")?1642 {1643 Ok(acknowledgement) => {1644 if let Some(bitrate_kbps) = acknowledgement.bitrate_kbps {1645 self.committed_bitrate_kbps = bitrate_kbps;1646 }1647 if let Some(resolution) = acknowledgement.resolution {1648 self.committed_resolution = Some(resolution);1649 }1650 }1651 Err(error) => {1652 warn!(%error, "generation adaptive commit was not applied");1653 }1654 }1655 Ok(())1656 }16571658 async fn abort_and_drain(&mut self) {1659 self.pending = false;1660 self.queued = None;1661 self.acknowledgements.abort_all();1662 while self.acknowledgements.join_next().await.is_some() {}1663 }1664}16651666impl AdaptiveCommitCoordinator {1667 fn new(1668 sessions: Arc<SessionRegistry>,1669 remote: EndpointId,1670 session_id: u64,1671 actuation_enabled: bool,1672 committed_bitrate_kbps: u32,1673 committed_resolution: Option<VideoDimensions>,1674 ) -> Self {1675 Self {1676 sessions,1677 remote,1678 session_id,1679 actuation_enabled,1680 desired_bitrate_kbps: committed_bitrate_kbps,1681 requested_bitrate_kbps: committed_bitrate_kbps,1682 committed_bitrate_kbps,1683 desired_resolution: committed_resolution,1684 requested_resolution: committed_resolution,1685 committed_resolution,1686 pending: false,1687 queued: None,1688 acknowledgements: tokio::task::JoinSet::new(),1689 }1690 }16911692 fn has_pending(&self) -> bool {1693 self.pending1694 }16951696 fn bitrate_is_applied(&self, target_kbps: u32) -> bool {1697 self.actuation_enabled1698 && !self.pending1699 && target_kbps == self.requested_bitrate_kbps1700 && target_kbps == self.committed_bitrate_kbps1701 }17021703 fn submit(&mut self, mut plan: AdaptiveCommitPlan) -> Option<AdaptiveCommitAcknowledgement> {1704 self.desired_bitrate_kbps = plan.decision.target_kbps;1705 if let Some(resolution) = plan.resolution_decision {1706 self.desired_resolution = Some(resolution.target);1707 }1708 if self.pending {1709 if let Some(queued) = self.queued {1710 // The latest decision replaces older diagnostics, but a1711 // recovery barrier requested by any coalesced desired update1712 // must survive until that desired state is actuated.1713 plan.force_keyframe |= queued.force_keyframe;1714 }1715 self.queued = Some(plan);1716 return None;1717 }1718 if let Some(queued) = self.queued.take() {1719 // A completion can deliberately defer this slot while an unread1720 // feedback generation is waiting. The newly evaluated plan is the1721 // actual latest state, but it must inherit any recovery barrier1722 // carried by the superseded slot.1723 plan.force_keyframe |= queued.force_keyframe;1724 }1725 self.start(plan)1726 }17271728 fn start_queued(&mut self) -> Option<AdaptiveCommitAcknowledgement> {1729 let plan = self.queued.take()?;1730 self.start(plan)1731 }17321733 fn start(&mut self, plan: AdaptiveCommitPlan) -> Option<AdaptiveCommitAcknowledgement> {1734 if !self.actuation_enabled {1735 return Some(AdaptiveCommitAcknowledgement::immediate(plan));1736 }17371738 let desired_resolution = plan.resolution_decision.map(|change| change.target);1739 debug_assert_eq!(plan.decision.target_kbps, self.desired_bitrate_kbps);1740 if desired_resolution.is_some() {1741 debug_assert_eq!(desired_resolution, self.desired_resolution);1742 }17431744 // A proposal mutates EncoderControl's persistent desired watch before1745 // its acknowledgement arrives. Reconcile against both the last1746 // requested and last committed values: after a timeout, returning to1747 // the old committed target must overwrite the still-pending stale1748 // request rather than allowing it to apply later.1749 let reconciles_resolution = desired_resolution.is_some()1750 && (desired_resolution != self.requested_resolution1751 || desired_resolution != self.committed_resolution);1752 let resolution_proposal = if reconciles_resolution {1753 let target = desired_resolution.expect("resolution target was checked as present");1754 match self1755 .sessions1756 .propose_resolution_update(self.remote, self.session_id, target)1757 {1758 Ok(Some(proposal)) => {1759 self.requested_resolution = Some(target);1760 Some(proposal)1761 }1762 Ok(None) => {1763 warn!(1764 remote = %self.remote,1765 session_id = self.session_id,1766 "active adaptive session lost encoder control; retaining committed resolution"1767 );1768 None1769 }1770 Err(error) => {1771 warn!(1772 remote = %self.remote,1773 session_id = self.session_id,1774 %error,1775 "adaptive resolution proposal failed; retaining committed resolution"1776 );1777 None1778 }1779 }1780 } else {1781 None1782 };17831784 let reconciles_bitrate = plan.decision.target_kbps != self.requested_bitrate_kbps1785 || plan.decision.target_kbps != self.committed_bitrate_kbps;1786 let force_keyframe = plan.force_keyframe && resolution_proposal.is_none();1787 let bitrate_proposal = if reconciles_bitrate || force_keyframe {1788 match self.sessions.propose_adaptive_encoder_update(1789 self.remote,1790 self.session_id,1791 plan.decision.target_kbps,1792 force_keyframe,1793 ) {1794 Ok(Some(proposal)) => {1795 self.requested_bitrate_kbps = plan.decision.target_kbps;1796 Some(proposal)1797 }1798 Ok(None) => {1799 warn!(1800 remote = %self.remote,1801 session_id = self.session_id,1802 "active adaptive session lost encoder control; retaining committed bitrate"1803 );1804 None1805 }1806 Err(error) => {1807 warn!(1808 remote = %self.remote,1809 session_id = self.session_id,1810 %error,1811 "adaptive encoder proposal failed; retaining committed bitrate"1812 );1813 None1814 }1815 }1816 } else {1817 None1818 };18191820 if bitrate_proposal.is_none() && resolution_proposal.is_none() {1821 return Some(AdaptiveCommitAcknowledgement::immediate(plan));1822 }18231824 self.pending = true;1825 let sessions = Arc::clone(&self.sessions);1826 let remote = self.remote;1827 let session_id = self.session_id;1828 self.acknowledgements.spawn(async move {1829 run_adaptive_commit(1830 sessions,1831 remote,1832 session_id,1833 plan,1834 bitrate_proposal,1835 resolution_proposal,1836 )1837 .await1838 });1839 None1840 }18411842 fn complete(1843 &mut self,1844 result: Option<std::result::Result<AdaptiveCommitAcknowledgement, tokio::task::JoinError>>,1845 ) -> Result<AdaptiveCommitAcknowledgement> {1846 ensure!(1847 self.pending,1848 "adaptive commit completed without pending work"1849 );1850 self.pending = false;1851 let acknowledgement = result1852 .context("adaptive commit task ended without a result")?1853 .context("adaptive commit task failed")?;1854 if let Some(applied) = acknowledgement.applied_bitrate_kbps {1855 self.committed_bitrate_kbps = applied;1856 }1857 if let Some(applied) = acknowledgement.applied_resolution {1858 self.committed_resolution = Some(applied);1859 }1860 Ok(acknowledgement)1861 }18621863 async fn abort_and_drain(&mut self) {1864 self.pending = false;1865 self.queued = None;1866 self.acknowledgements.abort_all();1867 while self.acknowledgements.join_next().await.is_some() {}1868 }1869}18701871pub(super) async fn serve_media_feedback(1872 connection: Connection,1873 config: &HostConfig,1874 sessions: &Arc<SessionRegistry>,1875 generations: &Arc<MediaGenerationManager>,1876 authorization: &AuthorizationPolicy,1877 input_operations: &Arc<InputOperations>,1878) -> Result<()> {1879 let remote = connection.remote_id();1880 let handshake_permit = sessions1881 .pending_handshakes1882 .try_acquire()1883 .context("too many pending handshakes")?;1884 let (mut send, mut recv) = tokio::time::timeout(HANDSHAKE_TIMEOUT, connection.accept_bi())1885 .await1886 .context("timed out accepting media feedback stream")?1887 .context("accepting media feedback stream")?;1888 let hello = receive_hello(&mut recv, Capability::VideoH264).await?;1889 drop(handshake_permit);1890 debug!(%remote, agent = %hello.agent, "media feedback hello received");1891 ensure!(1892 hello.invitation.is_none(),1893 "invitations are accepted only on the first media handshake"1894 );18951896 let grants = match authorization.authorize_or_redeem(remote, None, unix_timestamp_now()?) {1897 Ok(grants) => grants,1898 Err(error) => {1899 send_rejection(&mut send, "Portal peer is not authorized").await?;1900 return Err(error.context("authorizing media feedback peer"));1901 }1902 };1903 ensure!(1904 grants.contains(InvitationGrants::VIEW),1905 "authorized feedback peer lacks view permission"1906 );1907 let ceiling_kbps = adaptive_bitrate_ceiling_kbps(config)?;1908 let lease = match sessions.claim_feedback(remote, hello.nonce) {1909 Ok(lease) => lease,1910 Err(error) => {1911 send_rejection(&mut send, error.to_string()).await?;1912 return Err(error);1913 }1914 };1915 let encoder_actuation_available =1916 adaptive_bitrate_actuation_enabled(config) && lease.encoder_control.is_some();1917 // Shadow mode computes every decision and applies none. Say why, or a host1918 // silently never adapts and nothing in the logs explains it.1919 if !encoder_actuation_available {1920 let reason = if !adaptive_bitrate_actuation_enabled(config) {1921 match config.gamescope_pipewire.as_ref() {1922 Some(gamescope)1923 if gamescope.encoder_backend != GamescopeEncoderBackend::InProcessGstreamer =>1924 {1925 "gamescope_pipewire.encoder_backend is not in-process-gstreamer"1926 }1927 Some(gamescope) if gamescope.rate_control != VaapiRateControl::Cbr => {1928 "gamescope_pipewire.rate_control is not cbr"1929 }1930 Some(_) => "adaptive bitrate actuation is unavailable for this configuration",1931 None => "the configured source is not gamescope-pipewire",1932 }1933 } else {1934 "this session has no encoder control handle"1935 };1936 warn!(1937 %remote,1938 session_id = lease.session_id,1939 reason,1940 "adaptive bitrate and resolution run in shadow: decisions are computed but never applied"1941 );1942 }19431944 write_host_hello(1945 &mut send,1946 &HostHello::accepted(1947 lease.session_id,1948 negotiated_capabilities(&hello, MEDIA_CAPABILITIES),1949 ),1950 )1951 .await?;1952 if lease.scope == FeedbackScope::SharedGeneration {1953 let adaptive = generations.adaptive(lease.media_generation_id).await?;1954 let registration = adaptive.register(1955 remote,1956 lease.session_id,1957 lease.authorization_revision,1958 lease.media_generation_id,1959 Arc::clone(&lease.telemetry),1960 )?;1961 info!(1962 %remote,1963 session_id = lease.session_id,1964 media_generation_id = lease.media_generation_id,1965 authorization_revision = lease.authorization_revision,1966 ceiling_kbps,1967 mode = if encoder_actuation_available { "aggregate-active" } else { "aggregate-shadow" },1968 "media feedback contributor attached to shared generation"1969 );1970 let result = run_generation_feedback_session(1971 &connection,1972 &mut send,1973 recv,1974 remote,1975 sessions,1976 input_operations,1977 &lease,1978 registration,1979 )1980 .await;1981 let _ = send.finish();1982 drop(lease);1983 info!(%remote, "shared-generation media feedback contributor released");1984 return result;1985 }1986 info!(1987 %remote,1988 session_id = lease.session_id,1989 authorization_revision = lease.authorization_revision,1990 ceiling_kbps,1991 applied = false,1992 mode = if encoder_actuation_available { "active" } else { "shadow" },1993 "media feedback client accepted for bitrate control"1994 );19951996 let (feedback_sender, mut feedback_receiver) =1997 tokio::sync::watch::channel(CumulativeFeedbackIngress::default());1998 let reader_task = tokio::spawn(forward_media_feedback_reports(recv, feedback_sender));1999 let connection_closed = async {2000 let closed = connection.closed().await;2001 debug!(%remote, ?closed, "media feedback connection closed");2002 };2003 let result = run_media_feedback_loop(2004 &mut send,2005 connection_closed,2006 remote,2007 sessions,2008 &lease,2009 encoder_actuation_available,2010 ceiling_kbps,2011 &mut feedback_receiver,2012 reader_task,2013 )2014 .await;2015 let _ = send.finish();2016 drop(lease);2017 info!(%remote, "media feedback client released");2018 result2019}202020212022async fn run_generation_feedback_session<W>(2023 connection: &Connection,2024 send: &mut W,2025 recv: iroh::endpoint::RecvStream,2026 remote: EndpointId,2027 sessions: &Arc<SessionRegistry>,2028 input_operations: &Arc<InputOperations>,2029 lease: &FeedbackLease,2030 mut registration: GenerationFeedbackRegistration,2031) -> Result<()>2032where2033 W: tokio::io::AsyncWrite + Unpin,2034{2035 let (feedback_sender, mut feedback_receiver) =2036 tokio::sync::watch::channel(CumulativeFeedbackIngress::default());2037 let mut reader_task = tokio::spawn(forward_media_feedback_reports(recv, feedback_sender));2038 let mut feedback_cursor = FeedbackIngressCursor::default();2039 let result = loop {2040 let session_changed = sessions.session_changed.notified();2041 tokio::pin!(session_changed);2042 session_changed.as_mut().enable();2043 if !sessions.is_active(remote, lease.session_id) {2044 break Ok(());2045 }2046 tokio::select! {2047 reader_result = &mut reader_task => {2048 break reader_result2049 .context("shared-generation feedback reader task failed")?2050 .context("reading shared-generation feedback reports");2051 }2052 changed = feedback_receiver.changed() => {2053 if changed.is_err() {2054 break Ok(());2055 }2056 let ingress = *feedback_receiver.borrow_and_update();2057 let Some((report, next_cursor)) = ingress.report_since(feedback_cursor)? else {2058 continue;2059 };2060 feedback_cursor = next_cursor;2061 registration.observe(report)?;2062 }2063 changed = registration.decisions.changed() => {2064 changed.context("shared-generation adaptive decision publisher stopped")?;2065 let Some(decision) = *registration.decisions.borrow_and_update() else {2066 continue;2067 };2068 tokio::time::timeout(2069 FEEDBACK_WRITE_TIMEOUT,2070 write_adaptive_bitrate_decision_v1(send, &decision),2071 )2072 .await2073 .context("timed out writing shared-generation adaptive decision")??;2074 }2075 changed = registration.detach.changed() => {2076 changed.context("shared-generation recovery state publisher stopped")?;2077 if !*registration.detach.borrow_and_update() {2078 continue;2079 }2080 if let Some(transition) = sessions.invalidate_v2_focus(2081 remote,2082 lease.session_id,2083 FocusTransitionReasonV2::Disconnected,2084 )? {2085 input_operations.neutralize_focus_transition(2086 sessions,2087 transition.transition_id,2088 FocusTransitionReasonV2::Disconnected,2089 )?;2090 }2091 sessions.disconnect_v2_viewer(remote, lease.session_id)?;2092 info!(2093 %remote,2094 session_id = lease.session_id,2095 media_generation_id = lease.media_generation_id,2096 "detached persistently below-floor viewer after bounded isolated recovery"2097 );2098 break Ok(());2099 }2100 _ = &mut session_changed => {}2101 reason = connection.closed() => {2102 debug!(%remote, ?reason, "shared-generation feedback connection closed");2103 break Ok(());2104 }2105 }2106 };2107 if !reader_task.is_finished() {2108 reader_task.abort();2109 let _ = reader_task.await;2110 }2111 result2112}211321142115async fn run_media_feedback_loop<W, C>(2116 send: &mut W,2117 connection_closed: C,2118 remote: EndpointId,2119 sessions: &Arc<SessionRegistry>,2120 lease: &FeedbackLease,2121 encoder_actuation_available: bool,2122 ceiling_kbps: u32,2123 feedback_receiver: &mut tokio::sync::watch::Receiver<CumulativeFeedbackIngress>,2124 mut reader_task: tokio::task::JoinHandle<Result<()>>,2125) -> Result<()>2126where2127 W: tokio::io::AsyncWrite + Unpin,2128 C: Future<Output = ()>,2129{2130 let mut reader_finished = false;2131 let mut receiver_open = true;2132 let mut controller = ShadowBitrateController::new(ceiling_kbps, lease.telemetry.snapshot());2133 let mut resolution_controller = if encoder_actuation_available {2134 lease2135 .encoder_control2136 .as_ref()2137 .map(motion_resolution_policy_for_encoder)2138 .transpose()?2139 .flatten()2140 } else {2141 None2142 };2143 let initial_resolution = resolution_controller2144 .as_ref()2145 .map(MotionResolutionPolicy::target);2146 let mut commit_coordinator = AdaptiveCommitCoordinator::new(2147 Arc::clone(sessions),2148 remote,2149 lease.session_id,2150 encoder_actuation_available,2151 ceiling_kbps,2152 initial_resolution,2153 );2154 let mut feedback_cursor = FeedbackIngressCursor::default();2155 tokio::pin!(connection_closed);21562157 // Keep cleanup on every error path: `?` below exits this async block, not2158 // the outer function that owns the reader and actuator tasks.2159 let result: Result<()> = async {2160 loop {2161 let session_changed = sessions.session_changed.notified();2162 tokio::pin!(session_changed);2163 session_changed.as_mut().enable();2164 if !sessions.is_active(remote, lease.session_id) {2165 break Ok(());2166 }2167 tokio::select! {2168 biased;2169 reader_result = &mut reader_task, if !reader_finished => {2170 reader_finished = true;2171 match reader_result {2172 Ok(Ok(())) => {2173 debug!(%remote, "media feedback report stream closed");2174 }2175 Ok(Err(error)) => {2176 break Err(error).context("reading media feedback reports");2177 }2178 Err(error) => {2179 break Err(error).context("media feedback reader task failed");2180 }2181 }2182 }2183 commit_result = commit_coordinator.acknowledgements.join_next(),2184 if commit_coordinator.has_pending() =>2185 {2186 let acknowledgement = commit_coordinator.complete(commit_result)?;2187 if acknowledgement.generation_ended {2188 break Ok(());2189 }2190 log_adaptive_acknowledgement(2191 remote,2192 lease.session_id,2193 encoder_actuation_available,2194 &commit_coordinator,2195 &acknowledgement,2196 );2197 // If a newer report is already waiting, let its evaluation2198 // replace the queued actuator slot before starting more2199 // work. This avoids spending another full timeout on a2200 // plan that was stale at completion time.2201 if !matches!(feedback_receiver.has_changed(), Ok(true))2202 && let Some(immediate) = commit_coordinator.start_queued()2203 {2204 log_adaptive_acknowledgement(2205 remote,2206 lease.session_id,2207 encoder_actuation_available,2208 &commit_coordinator,2209 &immediate,2210 );2211 }2212 }2213 changed = feedback_receiver.changed(), if receiver_open => {2214 if changed.is_err() {2215 receiver_open = false;2216 if reader_finished {2217 break Ok(());2218 }2219 continue;2220 }2221 let ingress = *feedback_receiver.borrow_and_update();2222 let Some((report, next_cursor)) = ingress.report_since(feedback_cursor)? else {2223 continue;2224 };2225 // This branch now owns the complete aggregate. Advance the2226 // ingress baseline exactly once before evaluation; reports2227 // arriving during an encoder wait accumulate beyond it.2228 feedback_cursor = next_cursor;2229 let telemetry = lease.telemetry.snapshot();2230 let observed_at = Instant::now();2231 let mut candidate = controller.clone();2232 let evaluation = candidate.evaluate(&report, telemetry, observed_at)?;2233 let mut decision = evaluation.decision;2234 decision.applied =2235 commit_coordinator.bitrate_is_applied(decision.target_kbps);2236 let mut resolution_candidate = resolution_controller.clone();2237 let resolution_decision = resolution_candidate2238 .as_mut()2239 .map(|policy| policy.observe_window(&report, &evaluation, observed_at));2240 let force_keyframe =2241 adaptive_recovery_keyframe_required(&report, &decision);2242 let plan = AdaptiveCommitPlan {2243 decision,2244 telemetry,2245 resolution_decision,2246 force_keyframe,2247 };22482249 // One fixed-width decision is written for every evaluated2250 // report before actuator work is submitted. Encoder2251 // acknowledgements are correlated in bounded host2252 // diagnostics and never gate the Portal-facing cadence.2253 write_adaptive_decision(2254 send,2255 remote,2256 lease.session_id,2257 encoder_actuation_available,2258 &plan,2259 )2260 .await?;22612262 // Evaluation state advances at feedback cadence. The2263 // actuator independently retains the exact requested,2264 // committed, and latest desired targets.2265 controller = candidate;2266 resolution_controller = resolution_candidate;2267 if let Some(immediate) = commit_coordinator.submit(plan) {2268 log_adaptive_acknowledgement(2269 remote,2270 lease.session_id,2271 encoder_actuation_available,2272 &commit_coordinator,2273 &immediate,2274 );2275 }2276 }2277 _ = &mut session_changed => {2278 if !sessions.is_active(remote, lease.session_id) {2279 break Ok(());2280 }2281 }2282 _ = &mut connection_closed => {2283 break Ok(());2284 }2285 }2286 if reader_finished2287 && !receiver_open2288 && !commit_coordinator.has_pending()2289 && commit_coordinator.queued.is_none()2290 {2291 break Ok(());2292 }2293 }2294 }2295 .await;22962297 if !reader_finished {2298 reader_task.abort();2299 let _ = reader_task.await;2300 }2301 commit_coordinator.abort_and_drain().await;2302 result2303}23042305async fn write_adaptive_decision<W>(2306 send: &mut W,2307 remote: EndpointId,2308 session_id: u64,2309 encoder_actuation_available: bool,2310 plan: &AdaptiveCommitPlan,2311) -> Result<()>2312where2313 W: tokio::io::AsyncWrite + Unpin,2314{2315 let decision = plan.decision;2316 let telemetry = plan.telemetry;2317 let resolution_decision = plan.resolution_decision;2318 tokio::time::timeout(2319 FEEDBACK_WRITE_TIMEOUT,2320 write_adaptive_bitrate_decision_v1(send, &decision),2321 )2322 .await2323 .context("timed out writing adaptive bitrate decision")??;2324 debug!(2325 %remote,2326 session_id,2327 report_id = decision.report_id,2328 decision_id = decision.decision_id,2329 target_kbps = decision.target_kbps,2330 ?decision.state,2331 reasons = decision.reasons.bits(),2332 applied = decision.applied,2333 resolution_width = resolution_decision.map(|change| change.target.width),2334 resolution_height = resolution_decision.map(|change| change.target.height),2335 resolution_trigger = ?resolution_decision.map(|change| change.trigger),2336 path_rtt_micros = telemetry.selected_path_rtt_micros,2337 path_lost_packets = telemetry.selected_path_lost_packets,2338 path_congestion_events = telemetry.selected_path_congestion_events,2339 scheduler_cancellations = telemetry.scheduler_cancellations,2340 send_failures = telemetry.send_failures,2341 mode = if encoder_actuation_available { "active" } else { "shadow" },2342 "adaptive bitrate decision"2343 );2344 Ok(())2345}23462347fn log_adaptive_acknowledgement(2348 remote: EndpointId,2349 session_id: u64,2350 encoder_actuation_available: bool,2351 coordinator: &AdaptiveCommitCoordinator,2352 acknowledgement: &AdaptiveCommitAcknowledgement,2353) {2354 let decision = acknowledgement.plan.decision;2355 let resolution_decision = acknowledgement.plan.resolution_decision;2356 debug!(2357 %remote,2358 session_id,2359 report_id = decision.report_id,2360 decision_id = decision.decision_id,2361 target_kbps = decision.target_kbps,2362 bitrate_applied = acknowledgement.applied_bitrate_kbps.is_some(),2363 applied_bitrate_kbps = ?acknowledgement.applied_bitrate_kbps,2364 recovery_acknowledged = ?acknowledgement.recovery_acknowledged,2365 resolution_width = resolution_decision.map(|change| change.target.width),2366 resolution_height = resolution_decision.map(|change| change.target.height),2367 resolution_applied = ?acknowledgement.resolution_applied,2368 applied_resolution = ?acknowledgement.applied_resolution,2369 desired_bitrate_kbps = coordinator.desired_bitrate_kbps,2370 requested_bitrate_kbps = coordinator.requested_bitrate_kbps,2371 committed_bitrate_kbps = coordinator.committed_bitrate_kbps,2372 desired_resolution = ?coordinator.desired_resolution,2373 requested_resolution = ?coordinator.requested_resolution,2374 committed_resolution = ?coordinator.committed_resolution,2375 mode = if encoder_actuation_available { "active" } else { "shadow" },2376 "adaptive actuator acknowledgement"2377 );2378}23792380async fn run_adaptive_commit(2381 sessions: Arc<SessionRegistry>,2382 remote: EndpointId,2383 session_id: u64,2384 plan: AdaptiveCommitPlan,2385 bitrate_proposal: Option<AdaptiveEncoderProposal>,2386 resolution_proposal: Option<ResolutionEncoderProposal>,2387) -> AdaptiveCommitAcknowledgement {2388 let mut acknowledgement = AdaptiveCommitAcknowledgement::immediate(plan);2389 if let Some(proposal) = bitrate_proposal {2390 match commit_adaptive_encoder_proposal(&sessions, remote, session_id, &proposal).await {2391 AdaptiveEncoderCommit::GenerationEnded => {2392 acknowledgement.generation_ended = true;2393 return acknowledgement;2394 }2395 AdaptiveEncoderCommit::NotApplied(error) => {2396 warn!(2397 %remote,2398 session_id,2399 %error,2400 "adaptive encoder application failed; retaining committed bitrate"2401 );2402 }2403 AdaptiveEncoderCommit::Applied => {2404 acknowledgement.plan.decision.applied = true;2405 acknowledgement.applied_bitrate_kbps = Some(proposal.target_kbps);2406 if proposal.force_keyframe_revision.is_some() {2407 match await_adaptive_recovery_keyframe(&sessions, remote, session_id, &proposal)2408 .await2409 {2410 None => {2411 acknowledgement.generation_ended = true;2412 return acknowledgement;2413 }2414 Some(Ok(())) => acknowledgement.recovery_acknowledged = Some(true),2415 Some(Err(error)) => {2416 acknowledgement.recovery_acknowledged = Some(false);2417 warn!(2418 %remote,2419 session_id,2420 %error,2421 "adaptive bitrate applied but forced-IDR recovery was not acknowledged"2422 );2423 }2424 }2425 }2426 }2427 }2428 }24292430 if let Some(proposal) = resolution_proposal {2431 match commit_resolution_encoder_proposal(&sessions, remote, session_id, &proposal).await {2432 AdaptiveEncoderCommit::GenerationEnded => {2433 acknowledgement.generation_ended = true;2434 return acknowledgement;2435 }2436 AdaptiveEncoderCommit::NotApplied(error) => {2437 acknowledgement.resolution_applied = Some(false);2438 warn!(2439 %remote,2440 session_id,2441 target_width = proposal.target.width,2442 target_height = proposal.target.height,2443 %error,2444 "adaptive resolution application failed; retaining committed resolution"2445 );2446 }2447 AdaptiveEncoderCommit::Applied => {2448 acknowledgement.resolution_applied = Some(true);2449 acknowledgement.applied_resolution = Some(proposal.target);2450 }2451 }2452 }2453 acknowledgement2454}24552456async fn forward_media_feedback_reports<R>(2457 mut reader: R,2458 sender: tokio::sync::watch::Sender<CumulativeFeedbackIngress>,2459) -> Result<()>2460where2461 R: tokio::io::AsyncRead + Unpin,2462{2463 let mut last_report_id = None;2464 let mut next_read_at = tokio::time::Instant::now();2465 loop {2466 tokio::time::sleep_until(next_read_at).await;2467 let Some(report) = read_media_feedback_report_v1(&mut reader).await? else {2468 return Ok(());2469 };2470 ensure!(2471 last_report_id.is_none_or(|last| report.report_id > last),2472 "feedback report IDs must increase monotonically"2473 );2474 last_report_id = Some(report.report_id);2475 sender.send_modify(|ingress| ingress.observe(report));2476 if sender.is_closed() {2477 return Ok(());2478 }2479 next_read_at = tokio::time::Instant::now() + FEEDBACK_MIN_READ_INTERVAL;2480 }2481}24822483fn adaptive_bitrate_ceiling_kbps(config: &HostConfig) -> Result<u32> {2484 let configured = match config.source {2485 VideoSource::TestPattern => SHADOW_BITRATE_CEILING_KBPS,2486 VideoSource::GamescopePipewire => {2487 let gamescope = config2488 .gamescope_pipewire2489 .as_ref()2490 .context("Gamescope feedback requires gamescope_pipewire configuration")?;2491 match gamescope.rate_control {2492 VaapiRateControl::Cbr => gamescope2493 .bitrate_kbps2494 .context("adaptive bitrate shadow mode requires a Gamescope CBR bitrate")?,2495 VaapiRateControl::Cqp => SHADOW_BITRATE_CEILING_KBPS,2496 }2497 }2498 };2499 let ceiling = quantize_down(configured);2500 ensure!(2501 ceiling >= ADAPTIVE_BITRATE_FLOOR_KBPS,2502 "adaptive bitrate ceiling is below the 1000 kbps floor"2503 );2504 Ok(ceiling)2505}25062507fn adaptive_bitrate_actuation_enabled(config: &HostConfig) -> bool {2508 config.source == VideoSource::GamescopePipewire2509 && config.gamescope_pipewire.as_ref().is_some_and(|gamescope| {2510 gamescope.encoder_backend == GamescopeEncoderBackend::InProcessGstreamer2511 && gamescope.rate_control == VaapiRateControl::Cbr2512 })2513}25142515fn adaptive_recovery_keyframe_required(2516 report: &MediaFeedbackReportV1,2517 decision: &AdaptiveBitrateDecisionV1,2518) -> bool {2519 decision.state != AdaptiveBitrateStateV1::Hold2520 && report.flags.contains(MediaFeedbackFlags::RESYNC_ACTIVE)2521}252225232524enum AdaptiveEncoderCommit {2525 GenerationEnded,2526 NotApplied(anyhow::Error),2527 Applied,2528}25292530async fn await_while_session_active<F, T>(2531 sessions: &SessionRegistry,2532 remote: EndpointId,2533 session_id: u64,2534 future: F,2535) -> Option<T>2536where2537 F: Future<Output = T>,2538{2539 tokio::pin!(future);2540 loop {2541 let session_changed = sessions.session_changed.notified();2542 tokio::pin!(session_changed);2543 session_changed.as_mut().enable();2544 if !sessions.is_active(remote, session_id) {2545 return None;2546 }2547 tokio::select! {2548 result = &mut future => return sessions.is_active(remote, session_id).then_some(result),2549 _ = &mut session_changed => {}2550 }2551 }2552}25532554async fn commit_adaptive_encoder_proposal(2555 sessions: &SessionRegistry,2556 remote: EndpointId,2557 session_id: u64,2558 proposal: &AdaptiveEncoderProposal,2559) -> AdaptiveEncoderCommit {2560 let bitrate_result = await_while_session_active(2561 sessions,2562 remote,2563 session_id,2564 proposal2565 .control2566 .wait_for_bitrate_applied(proposal.bitrate_revision, ENCODER_CONTROL_COMMIT_TIMEOUT),2567 )2568 .await;2569 let Some(bitrate_result) = bitrate_result else {2570 return AdaptiveEncoderCommit::GenerationEnded;2571 };2572 let status = match bitrate_result {2573 Ok(status) => status,2574 Err(error) => return AdaptiveEncoderCommit::NotApplied(error),2575 };2576 if status.applied_bitrate_revision != Some(proposal.bitrate_revision)2577 || status.applied_bitrate_kbps != Some(proposal.target_kbps)2578 {2579 return AdaptiveEncoderCommit::NotApplied(anyhow::anyhow!(2580 "encoder bitrate commit mismatch: revision {:?}, readback {:?}, expected revision {} at {} kbps",2581 status.applied_bitrate_revision,2582 status.applied_bitrate_kbps,2583 proposal.bitrate_revision,2584 proposal.target_kbps2585 ));2586 }25872588 AdaptiveEncoderCommit::Applied2589}25902591async fn commit_resolution_encoder_proposal(2592 sessions: &SessionRegistry,2593 remote: EndpointId,2594 session_id: u64,2595 proposal: &ResolutionEncoderProposal,2596) -> AdaptiveEncoderCommit {2597 let result = await_while_session_active(2598 sessions,2599 remote,2600 session_id,2601 proposal.control.wait_for_resolution_applied(2602 proposal.revision,2603 proposal.target.width,2604 proposal.target.height,2605 ENCODER_CONTROL_COMMIT_TIMEOUT,2606 ),2607 )2608 .await;2609 let Some(result) = result else {2610 return AdaptiveEncoderCommit::GenerationEnded;2611 };2612 match result {2613 Ok(status)2614 if status.applied_resolution_revision == Some(proposal.revision)2615 && status.applied_width == Some(proposal.target.width)2616 && status.applied_height == Some(proposal.target.height) =>2617 {2618 AdaptiveEncoderCommit::Applied2619 }2620 Ok(status) => AdaptiveEncoderCommit::NotApplied(anyhow::anyhow!(2621 "encoder resolution commit mismatch: revision {:?}, dimensions {:?}x{:?}, expected revision {} at {}x{}",2622 status.applied_resolution_revision,2623 status.applied_width,2624 status.applied_height,2625 proposal.revision,2626 proposal.target.width,2627 proposal.target.height2628 )),2629 Err(error) => AdaptiveEncoderCommit::NotApplied(error),2630 }2631}26322633async fn await_adaptive_recovery_keyframe(2634 sessions: &SessionRegistry,2635 remote: EndpointId,2636 session_id: u64,2637 proposal: &AdaptiveEncoderProposal,2638) -> Option<Result<()>> {2639 let force_keyframe_revision = proposal.force_keyframe_revision?;2640 await_while_session_active(2641 sessions,2642 remote,2643 session_id,2644 proposal.control.wait_for_force_keyframe_acknowledged(2645 force_keyframe_revision,2646 ENCODER_CONTROL_COMMIT_TIMEOUT,2647 ),2648 )2649 .await2650 .map(|result| result.map(|_| ()))2651}265226532654mod tests {2655 use super::super::{endpoint, moq_test_config};2656 use super::*;26572658 fn clean_feedback(report_id: u64) -> MediaFeedbackReportV1 {2659 MediaFeedbackReportV1 {2660 report_id,2661 interval_ms: 1_000,2662 flags: MediaFeedbackFlags::NONE,2663 last_sequence: Some(report_id),2664 transport_dropped_delta: 0,2665 frontend_dropped_delta: 0,2666 decoder_dropped_delta: 0,2667 presenter_dropped_delta: 0,2668 frontend_queue_depth: 0,2669 frontend_queue_capacity: 4,2670 decode_queue_depth: 0,2671 decode_queue_capacity: 4,2672 presenter_queue_depth: 0,2673 presenter_queue_capacity: 2,2674 transport_delivery_p95_ms: Some(10),2675 decode_p95_ms: Some(3),2676 presentation_p95_ms: Some(5),2677 }2678 }26792680 fn aggregate_evaluation(2681 report_id: u64,2682 target_kbps: u32,2683 severity: FeedbackSeverity,2684 clean: bool,2685 ) -> AdaptiveBitrateEvaluation {2686 AdaptiveBitrateEvaluation {2687 decision: AdaptiveBitrateDecisionV1 {2688 decision_id: report_id,2689 report_id,2690 target_kbps,2691 floor_kbps: ADAPTIVE_BITRATE_FLOOR_KBPS,2692 ceiling_kbps: 12_000,2693 state: AdaptiveBitrateStateV1::Hold,2694 reasons: if severity == FeedbackSeverity::Clean {2695 AdaptiveBitrateReasonFlagsV1::CLEAN_RECOVERY2696 } else {2697 AdaptiveBitrateReasonFlagsV1::RECEIVER_QUEUE2698 },2699 applied: false,2700 },2701 pressure_severity: severity,2702 clean_recovery_evidence: clean,2703 fresh: severity != FeedbackSeverity::Stale,2704 }2705 }27062707 2708 fn aggregate_uses_worst_fresh_target_and_requires_every_contributor_clean() {2709 let clean = aggregate_evaluation(1, 10_000, FeedbackSeverity::Clean, true);2710 let pressured = aggregate_evaluation(2, 6_000, FeedbackSeverity::Moderate, false);2711 let active = vec![(clean, clean_feedback(1)), (pressured, clean_feedback(2))];2712 assert_eq!(2713 aggregate_target_and_clean(&active, 12_000, 12_000),2714 (6_000, false)2715 );27162717 let recovered = vec![2718 (clean, clean_feedback(3)),2719 (2720 aggregate_evaluation(4, 8_000, FeedbackSeverity::Clean, true),2721 clean_feedback(4),2722 ),2723 ];2724 assert_eq!(2725 aggregate_target_and_clean(&recovered, 6_000, 12_000),2726 (8_000, true)2727 );2728 }27292730 2731 fn sub_millisecond_rtt_jitter_does_not_become_permanent_moderate_pressure() {2732 let now = Instant::now();2733 let baseline = MediaV3TelemetrySnapshot {2734 selected_path_rtt_micros: 500,2735 ..Default::default()2736 };2737 let mut controller = ShadowBitrateController::new(12_000, baseline);2738 controller2739 .evaluate(&clean_feedback(1), baseline, now)2740 .unwrap();2741 let jitter = MediaV3TelemetrySnapshot {2742 selected_path_rtt_micros: 1_000,2743 ..Default::default()2744 };2745 let evaluation = controller2746 .evaluate(&clean_feedback(2), jitter, now + Duration::from_secs(1))2747 .unwrap();2748 assert!(2749 !evaluation2750 .decision2751 .reasons2752 .contains(AdaptiveBitrateReasonFlagsV1::RTT_INFLATION)2753 );2754 assert_eq!(evaluation.pressure_severity, FeedbackSeverity::Clean);2755 }27562757 2758 fn persistent_floor_breach_retries_then_detaches_without_holding_survivors_down() {2759 let now = Instant::now();2760 let severe = aggregate_evaluation(2761 1,2762 ADAPTIVE_BITRATE_FLOOR_KBPS,2763 FeedbackSeverity::Severe,2764 false,2765 );2766 let mut recovery = ContributorRecoveryState::default();2767 assert_eq!(recovery.observe(&severe, now), RecoveryMutation::None);2768 assert_eq!(recovery.observe(&severe, now), RecoveryMutation::None);2769 assert_eq!(2770 recovery.observe(&severe, now),2771 RecoveryMutation::RequestRecovery2772 );2773 assert!(!recovery.contributes());27742775 let first_retry = now + ISOLATED_RECOVERY_BACKOFF;2776 assert_eq!(2777 recovery.observe(&severe, first_retry),2778 RecoveryMutation::None2779 );2780 assert_eq!(2781 recovery.observe(&severe, first_retry),2782 RecoveryMutation::None2783 );2784 assert_eq!(2785 recovery.observe(&severe, first_retry),2786 RecoveryMutation::RequestRecovery2787 );2788 let second_retry = first_retry + ISOLATED_RECOVERY_BACKOFF.saturating_mul(2);2789 assert_eq!(2790 recovery.observe(&severe, second_retry),2791 RecoveryMutation::None2792 );2793 assert_eq!(2794 recovery.observe(&severe, second_retry),2795 RecoveryMutation::None2796 );2797 assert_eq!(2798 recovery.observe(&severe, second_retry),2799 RecoveryMutation::Detach2800 );2801 assert_eq!(recovery, ContributorRecoveryState::Detaching);2802 }28032804 2805 fn isolated_viewer_rejoins_only_after_bounded_clean_recovery() {2806 let now = Instant::now();2807 let severe = aggregate_evaluation(2808 1,2809 ADAPTIVE_BITRATE_FLOOR_KBPS,2810 FeedbackSeverity::Severe,2811 false,2812 );2813 let clean = aggregate_evaluation(2, 2_000, FeedbackSeverity::Clean, true);2814 let mut recovery = ContributorRecoveryState::default();2815 for _ in 0..FLOOR_BREACH_WINDOWS {2816 let _ = recovery.observe(&severe, now);2817 }2818 let retry = now + ISOLATED_RECOVERY_BACKOFF;2819 assert_eq!(recovery.observe(&clean, retry), RecoveryMutation::None);2820 assert_eq!(recovery.observe(&clean, retry), RecoveryMutation::None);2821 assert_eq!(recovery.observe(&clean, retry), RecoveryMutation::Recovered);2822 assert!(recovery.contributes());2823 }28242825 fn controller_decide(2826 controller: &mut ShadowBitrateController,2827 report: &MediaFeedbackReportV1,2828 second: u64,2829 ) -> AdaptiveBitrateDecisionV1 {2830 controller2831 .decide(2832 report,2833 MediaV3TelemetrySnapshot::default(),2834 Instant::now() + Duration::from_secs(second),2835 )2836 .unwrap()2837 }28382839 fn evaluate_commit_plan(2840 controller: &mut ShadowBitrateController,2841 resolution_controller: &mut Option<MotionResolutionPolicy>,2842 report: &MediaFeedbackReportV1,2843 observed_at: Instant,2844 ) -> AdaptiveCommitPlan {2845 let telemetry = MediaV3TelemetrySnapshot::default();2846 let evaluation = controller.evaluate(report, telemetry, observed_at).unwrap();2847 let resolution_decision = resolution_controller2848 .as_mut()2849 .map(|policy| policy.observe_window(report, &evaluation, observed_at));2850 AdaptiveCommitPlan {2851 decision: evaluation.decision,2852 telemetry,2853 resolution_decision,2854 force_keyframe: adaptive_recovery_keyframe_required(report, &evaluation.decision),2855 }2856 }28572858 fn native_dimensions() -> VideoDimensions {2859 VideoDimensions {2860 width: 1_280,2861 height: 800,2862 }2863 }28642865 2866 fn motion_resolution_tier_is_exact_even_and_same_aspect() {2867 let native = native_dimensions();2868 let policy = MotionResolutionPolicy::new(native).unwrap();2869 assert_eq!(policy.target(), native);2870 assert_eq!(2871 policy.reduced,2872 VideoDimensions {2873 width: 960,2874 height: 600,2875 }2876 );2877 assert_eq!(2878 u32::from(native.width) * u32::from(policy.reduced.height),2879 u32::from(native.height) * u32::from(policy.reduced.width)2880 );2881 assert!(policy.reduced.width.is_multiple_of(2));2882 assert!(policy.reduced.height.is_multiple_of(2));28832884 assert!(2885 MotionResolutionPolicy::new(VideoDimensions {2886 width: 1_278,2887 height: 800,2888 })2889 .is_err()2890 );2891 assert!(2892 MotionResolutionPolicy::new(VideoDimensions {2893 width: 80,2894 height: 64,2895 })2896 .is_err()2897 );2898 assert!(motion_resolution_policy(1_920, 1_080).unwrap().is_some());2899 assert!(motion_resolution_policy(1_366, 768).unwrap().is_none());2900 }29012902 2903 fn feedback_resolution_policy_uses_encoder_generation_dimensions() {2904 let full_hd = crate::source::EncoderControlTestHarness::with_dimensions(1_920, 1_080);2905 let policy = motion_resolution_policy_for_encoder(&full_hd.control)2906 .unwrap()2907 .unwrap();2908 assert_eq!(2909 policy.native,2910 VideoDimensions {2911 width: 1_920,2912 height: 1_080,2913 }2914 );29152916 let no_exact_tier = crate::source::EncoderControlTestHarness::with_dimensions(1_366, 768);2917 assert!(2918 motion_resolution_policy_for_encoder(&no_exact_tier.control)2919 .unwrap()2920 .is_none()2921 );2922 }29232924 2925 fn motion_downshifts_immediately_and_recovers_only_after_settle_and_clean_windows() {2926 let now = Instant::now();2927 let mut policy = MotionResolutionPolicy::new(native_dimensions()).unwrap();2928 let down = policy.observe_motion(true, now);2929 assert_eq!(down.tier, ResolutionTier::Reduced);2930 assert_eq!(down.trigger, ResolutionTrigger::Motion);2931 assert!(down.changed);29322933 let still = policy.observe_motion(false, now + Duration::from_secs(1));2934 assert!(!still.changed);2935 for second in [2, 3] {2936 assert!(2937 !policy2938 .observe_feedback(2939 FeedbackSeverity::Clean,2940 true,2941 now + Duration::from_secs(second),2942 )2943 .changed2944 );2945 }2946 let recovered =2947 policy.observe_feedback(FeedbackSeverity::Clean, true, now + Duration::from_secs(4));2948 assert_eq!(recovered.tier, ResolutionTier::Native);2949 assert_eq!(recovered.target, native_dimensions());2950 assert_eq!(recovered.trigger, ResolutionTrigger::Recovery);2951 assert!(recovered.changed);2952 }29532954 2955 fn repeated_still_observations_do_not_postpone_resolution_recovery() {2956 let now = Instant::now();2957 let mut policy = MotionResolutionPolicy::new(native_dimensions()).unwrap();2958 policy.observe_motion(true, now);2959 policy.observe_motion(false, now + Duration::from_secs(1));2960 policy.observe_feedback(FeedbackSeverity::Clean, true, now + Duration::from_secs(2));2961 policy.observe_motion(false, now + Duration::from_millis(2_500));2962 policy.observe_feedback(FeedbackSeverity::Clean, true, now + Duration::from_secs(3));2963 let recovered =2964 policy.observe_feedback(FeedbackSeverity::Clean, true, now + Duration::from_secs(4));2965 assert!(recovered.changed);2966 assert_eq!(recovered.tier, ResolutionTier::Native);2967 }29682969 2970 fn resolution_pressure_requires_fresh_hysteresis_but_severe_is_immediate() {2971 let now = Instant::now();2972 let mut moderate = MotionResolutionPolicy::new(native_dimensions()).unwrap();2973 moderate.observe_motion(false, now);2974 assert!(2975 !moderate2976 .observe_feedback(2977 FeedbackSeverity::Moderate,2978 true,2979 now + Duration::from_secs(1),2980 )2981 .changed2982 );2983 assert!(2984 !moderate2985 .observe_feedback(2986 FeedbackSeverity::Moderate,2987 false,2988 now + Duration::from_secs(2),2989 )2990 .changed2991 );2992 assert!(2993 !moderate2994 .observe_feedback(2995 FeedbackSeverity::Moderate,2996 true,2997 now + Duration::from_secs(3),2998 )2999 .changed3000 );3001 let down = moderate.observe_feedback(3002 FeedbackSeverity::Moderate,3003 true,3004 now + Duration::from_secs(4),3005 );3006 assert!(down.changed);3007 assert_eq!(down.trigger, ResolutionTrigger::Pressure);30083009 let mut severe = MotionResolutionPolicy::new(native_dimensions()).unwrap();3010 let immediate =3011 severe.observe_feedback(FeedbackSeverity::Severe, true, now + Duration::from_secs(1));3012 assert!(immediate.changed);3013 assert_eq!(immediate.tier, ResolutionTier::Reduced);3014 }30153016 3017 fn stale_feedback_cannot_restore_native_resolution() {3018 let now = Instant::now();3019 let mut policy = MotionResolutionPolicy::new(native_dimensions()).unwrap();3020 policy.observe_motion(false, now);3021 policy.observe_feedback(FeedbackSeverity::Severe, true, now + Duration::from_secs(1));3022 for second in [2, 3, 4, 5] {3023 let decision = policy.observe_feedback(3024 FeedbackSeverity::Clean,3025 false,3026 now + Duration::from_secs(second),3027 );3028 assert_eq!(decision.tier, ResolutionTier::Reduced);3029 assert!(!decision.changed);3030 }3031 assert!(policy.pressure_active);30323033 for second in [6, 7] {3034 assert!(3035 !policy3036 .observe_feedback(3037 FeedbackSeverity::Clean,3038 true,3039 now + Duration::from_secs(second),3040 )3041 .changed3042 );3043 }3044 let recovered =3045 policy.observe_feedback(FeedbackSeverity::Clean, true, now + Duration::from_secs(8));3046 assert!(recovered.changed);3047 assert_eq!(recovered.tier, ResolutionTier::Native);3048 }30493050 3051 fn motion_keeps_resolution_reduced_after_pressure_clears() {3052 let now = Instant::now();3053 let mut policy = MotionResolutionPolicy::new(native_dimensions()).unwrap();3054 policy.observe_motion(true, now);3055 policy.observe_feedback(FeedbackSeverity::Severe, true, now + Duration::from_secs(1));3056 for second in 2..=5 {3057 let decision = policy.observe_feedback(3058 FeedbackSeverity::Clean,3059 true,3060 now + Duration::from_secs(second),3061 );3062 assert_eq!(decision.tier, ResolutionTier::Reduced);3063 assert!(!decision.changed);3064 }3065 assert!(!policy.pressure_active);3066 assert!(policy.motion_active);3067 }30683069 3070 fn damage_driven_motion_proxy_seeds_then_downshifts_and_stale_blocks_upscale() {3071 let now = Instant::now();3072 let telemetry = MediaV3TelemetrySnapshot::default();3073 let mut bitrate = ShadowBitrateController::new(12_000, telemetry);3074 let mut resolution = MotionResolutionPolicy::new(native_dimensions()).unwrap();30753076 let mut first = clean_feedback(1);3077 first.last_sequence = Some(100);3078 let first_evaluation = bitrate.evaluate(&first, telemetry, now).unwrap();3079 assert!(!first_evaluation.fresh);3080 let seeded = resolution.observe_window(&first, &first_evaluation, now);3081 assert_eq!(seeded.tier, ResolutionTier::Native);3082 assert!(!seeded.changed);30833084 let mut second = clean_feedback(2);3085 second.last_sequence = Some(110);3086 let second_evaluation = bitrate3087 .evaluate(&second, telemetry, now + Duration::from_secs(1))3088 .unwrap();3089 assert!(second_evaluation.fresh);3090 let down =3091 resolution.observe_window(&second, &second_evaluation, now + Duration::from_secs(1));3092 assert_eq!(down.trigger, ResolutionTrigger::Motion);3093 assert_eq!(down.tier, ResolutionTier::Reduced);3094 assert!(down.changed);30953096 let mut stale = clean_feedback(3);3097 stale.interval_ms = 5_000;3098 stale.last_sequence = Some(110);3099 let stale_evaluation = bitrate3100 .evaluate(&stale, telemetry, now + Duration::from_secs(5))3101 .unwrap();3102 assert!(!stale_evaluation.fresh);3103 let held =3104 resolution.observe_window(&stale, &stale_evaluation, now + Duration::from_secs(5));3105 assert_eq!(held.tier, ResolutionTier::Reduced);3106 assert!(!held.changed);3107 }31083109 3110 fn damage_driven_no_progress_cannot_restore_native_resolution() {3111 let now = Instant::now();3112 let telemetry = MediaV3TelemetrySnapshot::default();3113 let mut bitrate = ShadowBitrateController::new(12_000, telemetry);3114 let mut resolution = MotionResolutionPolicy::new(native_dimensions()).unwrap();31153116 let mut seed = clean_feedback(1);3117 seed.last_sequence = Some(100);3118 let seed_evaluation = bitrate.evaluate(&seed, telemetry, now).unwrap();3119 resolution.observe_window(&seed, &seed_evaluation, now);31203121 let mut pressure = clean_feedback(2);3122 pressure.last_sequence = Some(110);3123 pressure.flags = MediaFeedbackFlags::RESYNC_ACTIVE;3124 let pressure_evaluation = bitrate3125 .evaluate(&pressure, telemetry, now + Duration::from_secs(1))3126 .unwrap();3127 let downshift = resolution.observe_window(3128 &pressure,3129 &pressure_evaluation,3130 now + Duration::from_secs(1),3131 );3132 assert_eq!(downshift.tier, ResolutionTier::Reduced);3133 assert!(downshift.changed);31343135 for second in 2..=4 {3136 let mut no_progress = clean_feedback(second + 1);3137 no_progress.last_sequence = Some(110);3138 let evaluation = bitrate3139 .evaluate(&no_progress, telemetry, now + Duration::from_secs(second))3140 .unwrap();3141 assert_eq!(evaluation.pressure_severity, FeedbackSeverity::Clean);3142 assert!(!evaluation.clean_recovery_evidence);3143 let decision = resolution.observe_window(3144 &no_progress,3145 &evaluation,3146 now + Duration::from_secs(second),3147 );3148 assert_eq!(decision.tier, ResolutionTier::Reduced);3149 assert_eq!(decision.trigger, ResolutionTrigger::Pressure);3150 assert!(!decision.changed);3151 assert_eq!(resolution.clean_windows, 0);3152 }31533154 for (report_id, second, sequence) in [(6, 5, 111), (7, 6, 112), (8, 7, 113)] {3155 let mut low_motion = clean_feedback(report_id);3156 low_motion.last_sequence = Some(sequence);3157 let evaluation = bitrate3158 .evaluate(&low_motion, telemetry, now + Duration::from_secs(second))3159 .unwrap();3160 assert!(evaluation.clean_recovery_evidence);3161 let decision = resolution.observe_window(3162 &low_motion,3163 &evaluation,3164 now + Duration::from_secs(second),3165 );3166 if report_id < 8 {3167 assert_eq!(decision.tier, ResolutionTier::Reduced);3168 assert!(!decision.changed);3169 } else {3170 assert_eq!(decision.tier, ResolutionTier::Native);3171 assert_eq!(decision.trigger, ResolutionTrigger::Recovery);3172 assert!(decision.changed);3173 }3174 }3175 }31763177 3178 fn shadow_controller_immediately_decreases_for_severe_feedback_and_stays_bounded() {3179 let mut controller =3180 ShadowBitrateController::new(12_137, MediaV3TelemetrySnapshot::default());3181 assert_eq!(controller.ceiling_kbps, 12_000);3182 let first = controller_decide(&mut controller, &clean_feedback(1), 0);3183 assert_eq!(first.state, AdaptiveBitrateStateV1::Hold);3184 assert!(3185 first3186 .reasons3187 .contains(AdaptiveBitrateReasonFlagsV1::FEEDBACK_STALE)3188 );3189 for report_id in 2..=21 {3190 let mut report = clean_feedback(report_id);3191 report.flags = MediaFeedbackFlags::RESYNC_ACTIVE;3192 let decision = controller_decide(&mut controller, &report, report_id - 1);3193 assert!(!decision.applied);3194 assert!(decision.target_kbps >= ADAPTIVE_BITRATE_FLOOR_KBPS);3195 assert!(decision.target_kbps <= 12_000);3196 assert_eq!(decision.target_kbps % ADAPTIVE_BITRATE_QUANTUM_KBPS, 0);3197 assert!(3198 decision3199 .reasons3200 .contains(AdaptiveBitrateReasonFlagsV1::RECEIVER_QUEUE)3201 );3202 }3203 assert_eq!(controller.target_kbps, ADAPTIVE_BITRATE_FLOOR_KBPS);3204 }32053206 3207 fn shadow_controller_requires_two_moderate_windows() {3208 let mut controller =3209 ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());3210 controller_decide(&mut controller, &clean_feedback(1), 0);3211 let mut first = clean_feedback(2);3212 first.frontend_queue_depth = 2;3213 assert_eq!(3214 controller_decide(&mut controller, &first, 1).state,3215 AdaptiveBitrateStateV1::Hold3216 );3217 let mut second = clean_feedback(3);3218 second.frontend_queue_depth = 2;3219 let decision = controller_decide(&mut controller, &second, 2);3220 assert_eq!(decision.state, AdaptiveBitrateStateV1::Decrease);3221 assert_eq!(decision.target_kbps, 9_500);3222 }32233224 3225 fn shadow_controller_increases_only_after_clean_cooldown() {3226 let mut controller =3227 ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());3228 controller_decide(&mut controller, &clean_feedback(1), 0);3229 let mut severe = clean_feedback(2);3230 severe.decoder_dropped_delta = 1;3231 assert_eq!(3232 controller_decide(&mut controller, &severe, 1).target_kbps,3233 9_0003234 );3235 for report_id in 3..=11 {3236 let decision =3237 controller_decide(&mut controller, &clean_feedback(report_id), report_id - 1);3238 assert_eq!(decision.state, AdaptiveBitrateStateV1::Hold);3239 assert_eq!(decision.target_kbps, 9_000);3240 }3241 let decision = controller_decide(&mut controller, &clean_feedback(12), 11);3242 assert_eq!(decision.state, AdaptiveBitrateStateV1::Increase);3243 assert_eq!(decision.target_kbps, 9_250);3244 assert!(3245 decision3246 .reasons3247 .contains(AdaptiveBitrateReasonFlagsV1::CLEAN_RECOVERY)3248 );3249 }32503251 3252 fn stale_or_non_monotonic_feedback_cannot_drive_an_increase() {3253 let mut controller =3254 ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());3255 controller_decide(&mut controller, &clean_feedback(1), 0);3256 let mut severe = clean_feedback(2);3257 severe.flags = MediaFeedbackFlags::RESYNC_ACTIVE;3258 controller_decide(&mut controller, &severe, 1);3259 let stale = controller_decide(&mut controller, &clean_feedback(3), 4);3260 assert_eq!(stale.state, AdaptiveBitrateStateV1::Hold);3261 assert_eq!(stale.target_kbps, 9_000);3262 assert!(3263 stale3264 .reasons3265 .contains(AdaptiveBitrateReasonFlagsV1::FEEDBACK_STALE)3266 );3267 assert!(3268 controller3269 .decide(3270 &clean_feedback(2),3271 MediaV3TelemetrySnapshot::default(),3272 Instant::now() + Duration::from_secs(4),3273 )3274 .is_err()3275 );32763277 let now = Instant::now();3278 let mut rapid = ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());3279 rapid3280 .decide(&clean_feedback(1), MediaV3TelemetrySnapshot::default(), now)3281 .unwrap();3282 let too_fast = rapid3283 .decide(3284 &clean_feedback(2),3285 MediaV3TelemetrySnapshot::default(),3286 now + Duration::from_millis(500),3287 )3288 .unwrap();3289 assert_eq!(too_fast.state, AdaptiveBitrateStateV1::Hold);3290 assert!(3291 too_fast3292 .reasons3293 .contains(AdaptiveBitrateReasonFlagsV1::FEEDBACK_STALE)3294 );3295 }32963297 3298 fn full_decode_queue_is_severe_even_at_capacity_two() {3299 let mut controller =3300 ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());3301 controller_decide(&mut controller, &clean_feedback(1), 0);3302 let mut report = clean_feedback(2);3303 report.decode_queue_capacity = 2;3304 report.decode_queue_depth = 2;3305 let decision = controller_decide(&mut controller, &report, 1);3306 assert_eq!(decision.state, AdaptiveBitrateStateV1::Decrease);3307 assert!(3308 decision3309 .reasons3310 .contains(AdaptiveBitrateReasonFlagsV1::DECODE_BACKLOG)3311 );3312 }33133314 3315 fn decision_id_exhaustion_fails_instead_of_repeating() {3316 let mut controller =3317 ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());3318 controller.next_decision_id = u64::MAX;3319 assert!(3320 controller3321 .decide(3322 &clean_feedback(1),3323 MediaV3TelemetrySnapshot::default(),3324 Instant::now(),3325 )3326 .is_err()3327 );3328 assert_eq!(controller.next_decision_id, u64::MAX);3329 }33303331 3332 fn stale_feedback_applies_trusted_loss_and_send_failure_pressure_once() {3333 for telemetry in [3334 MediaV3TelemetrySnapshot {3335 selected_path_lost_packets: 1,3336 ..MediaV3TelemetrySnapshot::default()3337 },3338 MediaV3TelemetrySnapshot {3339 send_failures: 1,3340 ..MediaV3TelemetrySnapshot::default()3341 },3342 ] {3343 let now = Instant::now();3344 let mut controller =3345 ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());3346 let decision = controller3347 .decide(&clean_feedback(1), telemetry, now)3348 .unwrap();3349 assert_eq!(decision.state, AdaptiveBitrateStateV1::Decrease);3350 assert_eq!(decision.target_kbps, 9_000);3351 assert!(3352 decision3353 .reasons3354 .contains(AdaptiveBitrateReasonFlagsV1::FEEDBACK_STALE)3355 );3356 assert!(3357 decision3358 .reasons3359 .contains(AdaptiveBitrateReasonFlagsV1::LOSS_OR_CANCELLATION)3360 );33613362 let repeated = controller3363 .decide(3364 &clean_feedback(2),3365 telemetry,3366 now + Duration::from_millis(500),3367 )3368 .unwrap();3369 assert_eq!(repeated.state, AdaptiveBitrateStateV1::Hold);3370 assert_eq!(repeated.target_kbps, 9_000);3371 }3372 }33733374 3375 fn stale_moderate_cancellation_is_retained_until_hysteresis_reduces() {3376 let now = Instant::now();3377 let telemetry = MediaV3TelemetrySnapshot {3378 scheduler_cancellations: 1,3379 ..MediaV3TelemetrySnapshot::default()3380 };3381 let mut controller =3382 ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());3383 let first = controller3384 .decide(&clean_feedback(1), telemetry, now)3385 .unwrap();3386 assert_eq!(first.state, AdaptiveBitrateStateV1::Hold);3387 assert_eq!(controller.previous_telemetry.scheduler_cancellations, 0);33883389 let second = controller3390 .decide(3391 &clean_feedback(2),3392 telemetry,3393 now + Duration::from_millis(500),3394 )3395 .unwrap();3396 assert_eq!(second.state, AdaptiveBitrateStateV1::Decrease);3397 assert_eq!(second.target_kbps, 9_500);3398 assert_eq!(controller.previous_telemetry.scheduler_cancellations, 1);3399 }34003401 3402 fn clean_recovery_below_five_mbps_advances_by_one_quantum() {3403 let mut controller =3404 ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());3405 controller.target_kbps = 4_000;3406 controller_decide(&mut controller, &clean_feedback(1), 0);3407 controller_decide(&mut controller, &clean_feedback(2), 1);3408 for report_id in 3..=11 {3409 let decision =3410 controller_decide(&mut controller, &clean_feedback(report_id), report_id - 1);3411 assert_eq!(decision.state, AdaptiveBitrateStateV1::Hold);3412 }3413 let decision = controller_decide(&mut controller, &clean_feedback(12), 11);3414 assert_eq!(decision.state, AdaptiveBitrateStateV1::Increase);3415 assert_eq!(decision.target_kbps, 4_250);3416 }34173418 3419 fn recovery_requires_latency_samples_and_sequence_progress_but_pressure_still_reduces() {3420 let mut controller =3421 ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());3422 controller.target_kbps = 9_000;3423 controller_decide(&mut controller, &clean_feedback(1), 0);3424 controller_decide(&mut controller, &clean_feedback(2), 1);34253426 for report_id in 3..=14 {3427 let mut report = clean_feedback(report_id);3428 report.last_sequence = Some(2);3429 let decision = controller_decide(&mut controller, &report, report_id - 1);3430 assert_eq!(decision.state, AdaptiveBitrateStateV1::Hold);3431 assert_eq!(decision.target_kbps, 9_000);3432 }3433 assert_eq!(controller.clean_windows, 0);34343435 for report_id in 15..=26 {3436 let mut report = clean_feedback(report_id);3437 report.decode_p95_ms = None;3438 let decision = controller_decide(&mut controller, &report, report_id - 1);3439 assert_eq!(decision.state, AdaptiveBitrateStateV1::Hold);3440 assert_eq!(decision.target_kbps, 9_000);3441 }3442 assert_eq!(controller.clean_windows, 0);34433444 let mut pressure = clean_feedback(27);3445 pressure.decode_p95_ms = None;3446 pressure.decode_queue_capacity = 2;3447 pressure.decode_queue_depth = 2;3448 let decision = controller_decide(&mut controller, &pressure, 26);3449 assert_eq!(decision.state, AdaptiveBitrateStateV1::Decrease);3450 assert_eq!(decision.target_kbps, 6_750);3451 assert!(3452 decision3453 .reasons3454 .contains(AdaptiveBitrateReasonFlagsV1::DECODE_BACKLOG)3455 );3456 }34573458 fn gamescope_feedback_test_config(rate_control: VaapiRateControl) -> HostConfig {3459 let (bitrate_kbps, quantizer) = match rate_control {3460 VaapiRateControl::Cbr => (Some(12_123), None),3461 VaapiRateControl::Cqp => (None, Some(24)),3462 };3463 let mut config = moq_test_config();3464 config.source = VideoSource::GamescopePipewire;3465 config.width = None;3466 config.height = None;3467 config.gamescope_pipewire = Some(crate::config::GamescopePipewireConfig {3468 node_name: "gamescope".into(),3469 media_class: "Video/Source".into(),3470 match_properties: std::collections::BTreeMap::new(),3471 xwayland_display: None,3472 pw_dump_path: "/usr/bin/pw-dump".into(),3473 gst_launch_path: "/usr/bin/gst-launch-1.0".into(),3474 gst_inspect_path: "/usr/bin/gst-inspect-1.0".into(),3475 encoder_backend: GamescopeEncoderBackend::ExternalGstLaunch,3476 vaapi_encoder: "vah264enc".into(),3477 vaapi_render_node: "/dev/dri/renderD128".into(),3478 rate_control,3479 bitrate_kbps,3480 quantizer,3481 });3482 config.validate().unwrap();3483 config3484 }34853486 3487 fn external_cqp_uses_a_fixed_shadow_bitrate_ceiling() {3488 let config = gamescope_feedback_test_config(VaapiRateControl::Cqp);34893490 assert_eq!(3491 adaptive_bitrate_ceiling_kbps(&config).unwrap(),3492 SHADOW_BITRATE_CEILING_KBPS3493 );3494 assert!(!adaptive_bitrate_actuation_enabled(&config));3495 }34963497 3498 fn gamescope_cbr_keeps_its_quantized_configured_ceiling() {3499 let config = gamescope_feedback_test_config(VaapiRateControl::Cbr);35003501 assert_eq!(adaptive_bitrate_ceiling_kbps(&config).unwrap(), 12_000);3502 }35033504 3505 fn malformed_cbr_without_a_bitrate_still_fails_closed() {3506 let mut config = gamescope_feedback_test_config(VaapiRateControl::Cbr);3507 config.gamescope_pipewire.as_mut().unwrap().bitrate_kbps = None;35083509 assert_eq!(3510 adaptive_bitrate_ceiling_kbps(&config)3511 .unwrap_err()3512 .to_string(),3513 "adaptive bitrate shadow mode requires a Gamescope CBR bitrate"3514 );3515 }35163517 3518 async fn adaptive_encoder_commit_requires_exact_readback_and_tracks_recovery_separately() {3519 let sessions = Arc::new(SessionRegistry::default());3520 let remote = endpoint(1);3521 let media = sessions3522 .claim(remote, [4; 16], InvitationGrants::VIEW)3523 .unwrap();3524 let harness = crate::source::EncoderControlTestHarness::new();3525 sessions3526 .install_encoder_control(remote, media.session_id, Some(harness.control.clone()))3527 .unwrap();35283529 let proposal = sessions3530 .propose_adaptive_encoder_update(remote, media.session_id, 8_000, true)3531 .unwrap()3532 .unwrap();3533 harness.status.send_modify(|status| {3534 status.applied_bitrate_revision = Some(proposal.bitrate_revision);3535 status.applied_bitrate_kbps = Some(8_000);3536 status.requested_force_keyframe_revision = proposal.force_keyframe_revision;3537 status.acknowledged_force_keyframe_revision = proposal.force_keyframe_revision;3538 });3539 assert!(matches!(3540 commit_adaptive_encoder_proposal(&sessions, remote, media.session_id, &proposal).await,3541 AdaptiveEncoderCommit::Applied3542 ));3543 assert!(matches!(3544 await_adaptive_recovery_keyframe(&sessions, remote, media.session_id, &proposal).await,3545 Some(Ok(()))3546 ));35473548 let mismatch = sessions3549 .propose_adaptive_encoder_update(remote, media.session_id, 7_000, false)3550 .unwrap()3551 .unwrap();3552 harness.status.send_modify(|status| {3553 status.applied_bitrate_revision = Some(mismatch.bitrate_revision);3554 status.applied_bitrate_kbps = Some(7_250);3555 });3556 assert!(matches!(3557 commit_adaptive_encoder_proposal(&sessions, remote, media.session_id, &mismatch).await,3558 AdaptiveEncoderCommit::NotApplied(_)3559 ));35603561 let stale_generation = sessions3562 .propose_adaptive_encoder_update(remote, media.session_id, 6_000, false)3563 .unwrap()3564 .unwrap();3565 let session_id = media.session_id;3566 drop(media);3567 assert!(matches!(3568 commit_adaptive_encoder_proposal(&sessions, remote, session_id, &stale_generation)3569 .await,3570 AdaptiveEncoderCommit::GenerationEnded3571 ));3572 }35733574 3575 fn adaptive_recovery_idr_is_only_requested_for_a_bitrate_change_during_resync() {3576 let mut report = clean_feedback(1);3577 let mut decision = AdaptiveBitrateDecisionV1 {3578 decision_id: 1,3579 report_id: 1,3580 target_kbps: 8_000,3581 floor_kbps: 1_000,3582 ceiling_kbps: 12_000,3583 state: AdaptiveBitrateStateV1::Decrease,3584 reasons: AdaptiveBitrateReasonFlagsV1::RECEIVER_QUEUE,3585 applied: false,3586 };3587 assert!(!adaptive_recovery_keyframe_required(&report, &decision));3588 report.flags = MediaFeedbackFlags::RESYNC_ACTIVE;3589 assert!(adaptive_recovery_keyframe_required(&report, &decision));3590 decision.state = AdaptiveBitrateStateV1::Hold;3591 assert!(!adaptive_recovery_keyframe_required(&report, &decision));3592 }35933594 3595 async fn stalled_adaptive_apply_coalesces_feedback_without_losing_pressure() {3596 use tokio::io::AsyncWriteExt as _;35973598 let (sender, mut receiver) =3599 tokio::sync::watch::channel(CumulativeFeedbackIngress::default());3600 let (mut writer, reader) = tokio::io::duplex(512);3601 let reader_task = tokio::spawn(forward_media_feedback_reports(reader, sender.clone()));3602 let mut cursor = FeedbackIngressCursor::default();36033604 sigil_protocol::write_media_feedback_report_v1(&mut writer, &clean_feedback(1))3605 .await3606 .unwrap();3607 receiver.changed().await.unwrap();3608 let (first, next_cursor) = receiver3609 .borrow_and_update()3610 .report_since(cursor)3611 .unwrap()3612 .unwrap();3613 assert_eq!(first.report_id, 1);3614 cursor = next_cursor;36153616 // The consumer is now notionally stalled on an encoder commit. Two3617 // reports overwrite one watch notification, but their interval,3618 // deltas, and resync pressure must remain cumulative.3619 let mut pressured = clean_feedback(2);3620 pressured.flags = MediaFeedbackFlags::RESYNC_ACTIVE;3621 pressured.transport_dropped_delta = 2;3622 pressured.decoder_dropped_delta = 1;3623 pressured.decode_queue_depth = 3;3624 pressured.transport_delivery_p95_ms = Some(180);3625 sigil_protocol::write_media_feedback_report_v1(&mut writer, &pressured)3626 .await3627 .unwrap();36283629 let mut latest = clean_feedback(3);3630 latest.frontend_dropped_delta = 4;3631 latest.presenter_dropped_delta = 3;3632 latest.frontend_queue_depth = 3;3633 latest.decode_queue_depth = 1;3634 latest.transport_delivery_p95_ms = Some(40);3635 latest.decode_p95_ms = Some(9);3636 sigil_protocol::write_media_feedback_report_v1(&mut writer, &latest)3637 .await3638 .unwrap();3639 writer.shutdown().await.unwrap();3640 reader_task.await.unwrap().unwrap();36413642 receiver.changed().await.unwrap();3643 let (aggregate, next_cursor) = receiver3644 .borrow_and_update()3645 .report_since(cursor)3646 .unwrap()3647 .unwrap();3648 assert_eq!(aggregate.report_id, 3);3649 assert_eq!(aggregate.interval_ms, 2_000);3650 assert_eq!(aggregate.transport_dropped_delta, 2);3651 assert_eq!(aggregate.frontend_dropped_delta, 4);3652 assert_eq!(aggregate.decoder_dropped_delta, 1);3653 assert_eq!(aggregate.presenter_dropped_delta, 3);3654 assert!(aggregate.flags.contains(MediaFeedbackFlags::RESYNC_ACTIVE));3655 assert_eq!(aggregate.last_sequence, latest.last_sequence);3656 assert_eq!(aggregate.frontend_queue_depth, latest.frontend_queue_depth);3657 assert_eq!(aggregate.decode_queue_depth, latest.decode_queue_depth);3658 assert_eq!(3659 aggregate.transport_delivery_p95_ms,3660 latest.transport_delivery_p95_ms3661 );3662 assert_eq!(aggregate.decode_p95_ms, latest.decode_p95_ms);3663 cursor = next_cursor;3664 assert!(3665 receiver.borrow().report_since(cursor).unwrap().is_none(),3666 "one cumulative ingress generation must be processed only once"3667 );36683669 for report_id in 4..=9 {3670 sender.send_modify(|ingress| ingress.observe(clean_feedback(report_id)));3671 }3672 receiver.changed().await.unwrap();3673 let (clamped, _) = receiver3674 .borrow_and_update()3675 .report_since(cursor)3676 .unwrap()3677 .unwrap();3678 assert_eq!(clamped.interval_ms, 5_000);3679 let mut controller =3680 ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());3681 let decision = controller3682 .decide(3683 &clamped,3684 MediaV3TelemetrySnapshot::default(),3685 Instant::now(),3686 )3687 .unwrap();3688 assert!(3689 decision3690 .reasons3691 .contains(AdaptiveBitrateReasonFlagsV1::FEEDBACK_STALE),3692 "a clamped aggregate must be stale rather than disguised as one fresh report"3693 );3694 }36953696 3697 async fn queued_adaptive_plan_keeps_latest_ids_and_recovery_barrier() {3698 let sessions = Arc::new(SessionRegistry::default());3699 let remote = endpoint(1);3700 let media = sessions3701 .claim(remote, [3; 16], InvitationGrants::VIEW)3702 .unwrap();3703 let harness = crate::source::EncoderControlTestHarness::new();3704 sessions3705 .install_encoder_control(remote, media.session_id, Some(harness.control.clone()))3706 .unwrap();3707 let mut coordinator = AdaptiveCommitCoordinator::new(3708 Arc::clone(&sessions),3709 remote,3710 media.session_id,3711 true,3712 12_000,3713 Some(native_dimensions()),3714 );3715 let mut controller =3716 ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());3717 let mut resolution_controller = None;3718 let observed_at = Instant::now();37193720 let seed = evaluate_commit_plan(3721 &mut controller,3722 &mut resolution_controller,3723 &clean_feedback(1),3724 observed_at,3725 );3726 assert!(coordinator.submit(seed).is_some());37273728 let mut first_pressure = clean_feedback(2);3729 first_pressure.decoder_dropped_delta = 1;3730 let first = evaluate_commit_plan(3731 &mut controller,3732 &mut resolution_controller,3733 &first_pressure,3734 observed_at + Duration::from_secs(1),3735 );3736 assert!(!first.force_keyframe);3737 assert!(coordinator.submit(first).is_none());37383739 let mut recovery_pressure = clean_feedback(3);3740 recovery_pressure.decoder_dropped_delta = 1;3741 recovery_pressure.flags = MediaFeedbackFlags::RESYNC_ACTIVE;3742 let recovery = evaluate_commit_plan(3743 &mut controller,3744 &mut resolution_controller,3745 &recovery_pressure,3746 observed_at + Duration::from_secs(2),3747 );3748 assert!(recovery.force_keyframe);3749 assert!(coordinator.submit(recovery).is_none());37503751 let latest = evaluate_commit_plan(3752 &mut controller,3753 &mut resolution_controller,3754 &clean_feedback(4),3755 observed_at + Duration::from_secs(3),3756 );3757 assert!(!latest.force_keyframe);3758 assert!(coordinator.submit(latest).is_none());3759 let retained = coordinator.queued.expect("latest plan must be retained");3760 assert_eq!(retained.decision.decision_id, latest.decision.decision_id);3761 assert_eq!(retained.decision.report_id, latest.decision.report_id);3762 assert!(3763 retained.force_keyframe,3764 "a coalesced recovery barrier must survive a newer hold decision"3765 );37663767 coordinator.abort_and_drain().await;3768 drop(media);3769 }37703771 3772 async fn timed_out_actuator_requests_are_overwritten_when_policy_returns_to_committed_state() {3773 let sessions = Arc::new(SessionRegistry::default());3774 let remote = endpoint(1);3775 let media = sessions3776 .claim(remote, [8; 16], InvitationGrants::VIEW)3777 .unwrap();3778 let harness = crate::source::EncoderControlTestHarness::new();3779 sessions3780 .install_encoder_control(remote, media.session_id, Some(harness.control.clone()))3781 .unwrap();3782 let native = native_dimensions();3783 let reduced = native.three_quarter().unwrap();3784 let mut coordinator = AdaptiveCommitCoordinator::new(3785 Arc::clone(&sessions),3786 remote,3787 media.session_id,3788 true,3789 12_000,3790 Some(native),3791 );3792 let timed_out = AdaptiveCommitPlan {3793 decision: AdaptiveBitrateDecisionV1 {3794 decision_id: 1,3795 report_id: 1,3796 target_kbps: 9_000,3797 floor_kbps: 1_000,3798 ceiling_kbps: 12_000,3799 state: AdaptiveBitrateStateV1::Decrease,3800 reasons: AdaptiveBitrateReasonFlagsV1::RECEIVER_QUEUE,3801 applied: false,3802 },3803 telemetry: MediaV3TelemetrySnapshot::default(),3804 resolution_decision: Some(MotionResolutionDecision {3805 tier: ResolutionTier::Reduced,3806 target: reduced,3807 trigger: ResolutionTrigger::Pressure,3808 changed: true,3809 }),3810 force_keyframe: false,3811 };3812 assert!(coordinator.submit(timed_out).is_none());3813 let stale_bitrate = harness3814 .requested_bitrate()3815 .expect("the first bitrate request must reach the actuator");3816 let stale_resolution = harness3817 .requested_resolution()3818 .expect("the first resolution request must reach the actuator");3819 assert_eq!(stale_bitrate.1, 9_000);3820 assert_eq!(3821 (stale_resolution.1, stale_resolution.2),3822 (reduced.width, reduced.height)3823 );38243825 // Feedback evaluation recovers all the way to the previously3826 // committed state while the old request is still awaiting readback.3827 let recovered = AdaptiveCommitPlan {3828 decision: AdaptiveBitrateDecisionV1 {3829 decision_id: 2,3830 report_id: 2,3831 target_kbps: 12_000,3832 floor_kbps: 1_000,3833 ceiling_kbps: 12_000,3834 state: AdaptiveBitrateStateV1::Increase,3835 reasons: AdaptiveBitrateReasonFlagsV1::CLEAN_RECOVERY,3836 applied: false,3837 },3838 telemetry: MediaV3TelemetrySnapshot::default(),3839 resolution_decision: Some(MotionResolutionDecision {3840 tier: ResolutionTier::Native,3841 target: native,3842 trigger: ResolutionTrigger::Recovery,3843 changed: true,3844 }),3845 force_keyframe: false,3846 };3847 assert!(coordinator.submit(recovered).is_none());38483849 // The first task times out its bitrate wait and then its resolution3850 // wait. Neither timeout changes the last acknowledged state.3851 tokio::task::yield_now().await;3852 tokio::time::advance(ENCODER_CONTROL_COMMIT_TIMEOUT).await;3853 tokio::task::yield_now().await;3854 tokio::time::advance(ENCODER_CONTROL_COMMIT_TIMEOUT).await;3855 tokio::task::yield_now().await;3856 let result = coordinator.acknowledgements.join_next().await;3857 let acknowledgement = coordinator.complete(result).unwrap();3858 assert!(!acknowledgement.plan.decision.applied);3859 assert_eq!(acknowledgement.resolution_applied, Some(false));3860 assert_eq!(coordinator.committed_bitrate_kbps, 12_000);3861 assert_eq!(coordinator.committed_resolution, Some(native));3862 assert_eq!(coordinator.requested_bitrate_kbps, 9_000);3863 assert_eq!(coordinator.requested_resolution, Some(reduced));38643865 // Starting the recovered plan must overwrite the stale persistent3866 // actuator watch even though its targets equal the old committed3867 // values. Otherwise the timed-out request would apply on resume.3868 assert!(coordinator.start_queued().is_none());3869 assert!(coordinator.has_pending());3870 let corrected_bitrate = harness3871 .requested_bitrate()3872 .expect("recovered bitrate must overwrite stale actuator intent");3873 let corrected_resolution = harness3874 .requested_resolution()3875 .expect("recovered resolution must overwrite stale actuator intent");3876 assert!(corrected_bitrate.0 > stale_bitrate.0);3877 assert!(corrected_resolution.0 > stale_resolution.0);3878 assert_eq!(corrected_bitrate.1, 12_000);3879 assert_eq!(3880 (corrected_resolution.1, corrected_resolution.2),3881 (native.width, native.height)3882 );3883 assert_eq!(coordinator.requested_bitrate_kbps, 12_000);3884 assert_eq!(coordinator.requested_resolution, Some(native));38853886 coordinator.abort_and_drain().await;3887 drop(media);3888 }38893890 3891 async fn feedback_loop_writes_every_decision_while_combined_commit_is_hung() {3892 let sessions = Arc::new(SessionRegistry::default());3893 let remote = endpoint(1);3894 let nonce = [9; 16];3895 let media = sessions3896 .claim(remote, nonce, InvitationGrants::VIEW)3897 .unwrap();3898 let harness = crate::source::EncoderControlTestHarness::new();3899 sessions3900 .install_encoder_control(remote, media.session_id, Some(harness.control.clone()))3901 .unwrap();3902 let feedback = sessions.claim_feedback(remote, nonce).unwrap();3903 let session_id = feedback.session_id;39043905 let (mut report_writer, report_reader) = tokio::io::duplex(4_096);3906 let (feedback_sender, mut feedback_receiver) =3907 tokio::sync::watch::channel(CumulativeFeedbackIngress::default());3908 let reader_task = tokio::spawn(forward_media_feedback_reports(3909 report_reader,3910 feedback_sender,3911 ));3912 let (mut decision_writer, mut decision_reader) = tokio::io::duplex(4_096);3913 let loop_sessions = Arc::clone(&sessions);3914 let loop_task = tokio::spawn(async move {3915 run_media_feedback_loop(3916 &mut decision_writer,3917 std::future::pending(),3918 remote,3919 &loop_sessions,3920 &feedback,3921 true,3922 12_000,3923 &mut feedback_receiver,3924 reader_task,3925 )3926 .await3927 });39283929 let mut observed = Vec::new();3930 for report_id in 1..=5 {3931 if report_id > 1 {3932 tokio::time::sleep(Duration::from_millis(800)).await;3933 }3934 let mut report = clean_feedback(report_id);3935 if report_id >= 2 {3936 report.flags = MediaFeedbackFlags::RESYNC_ACTIVE;3937 report.decoder_dropped_delta = 4;3938 }3939 sigil_protocol::write_media_feedback_report_v1(&mut report_writer, &report)3940 .await3941 .unwrap();3942 let decision = tokio::time::timeout(3943 Duration::from_secs(1),3944 sigil_protocol::read_adaptive_bitrate_decision_v1(&mut decision_reader),3945 )3946 .await3947 .expect("each evaluated report must be written before the hung commit timeout")3948 .unwrap()3949 .expect("feedback loop closed before writing its decision");3950 observed.push((decision.decision_id, decision.report_id));3951 }3952 assert_eq!(3953 observed,3954 (1..=5).map(|id| (id, id)).collect::<Vec<_>>(),3955 "slow bitrate and resolution readbacks must not coalesce Portal decisions"3956 );3957 assert!(3958 harness.requested_bitrate().is_some(),3959 "the adversarial run must actually have pending bitrate work"3960 );3961 assert!(3962 harness.requested_resolution().is_some(),3963 "the adversarial run must actually have pending resolution work"3964 );39653966 drop(media);3967 drop(report_writer);3968 tokio::time::timeout(Duration::from_secs(1), loop_task)3969 .await3970 .expect("session teardown must abort the hung commit promptly")3971 .unwrap()3972 .unwrap();3973 assert!(!sessions.is_active(remote, session_id));3974 }39753976 3977 async fn slow_adaptive_commit_does_not_pause_feedback_evaluation() {3978 let sessions = Arc::new(SessionRegistry::default());3979 let remote = endpoint(1);3980 let media = sessions3981 .claim(remote, [4; 16], InvitationGrants::VIEW)3982 .unwrap();3983 let harness = crate::source::EncoderControlTestHarness::new();3984 sessions3985 .install_encoder_control(remote, media.session_id, Some(harness.control.clone()))3986 .unwrap();3987 let mut coordinator = AdaptiveCommitCoordinator::new(3988 Arc::clone(&sessions),3989 remote,3990 media.session_id,3991 true,3992 12_000,3993 Some(native_dimensions()),3994 );3995 let mut controller =3996 ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());3997 let mut resolution_controller = None;3998 let observed_at = Instant::now();39994000 let seed = evaluate_commit_plan(4001 &mut controller,4002 &mut resolution_controller,4003 &clean_feedback(1),4004 observed_at,4005 );4006 assert_eq!(4007 coordinator.submit(seed).unwrap().plan.decision.decision_id,4008 14009 );40104011 let mut severe = clean_feedback(2);4012 severe.decoder_dropped_delta = 1;4013 let first = evaluate_commit_plan(4014 &mut controller,4015 &mut resolution_controller,4016 &severe,4017 observed_at + Duration::from_secs(1),4018 );4019 assert_eq!(first.decision.state, AdaptiveBitrateStateV1::Decrease);4020 assert!(coordinator.submit(first).is_none());4021 assert!(coordinator.has_pending());40224023 tokio::time::advance(Duration::from_secs(1)).await;4024 let evaluation_tick = tokio::time::Instant::now();4025 let mut newer_severe = clean_feedback(3);4026 newer_severe.decoder_dropped_delta = 1;4027 let queued = evaluate_commit_plan(4028 &mut controller,4029 &mut resolution_controller,4030 &newer_severe,4031 observed_at + Duration::from_secs(2),4032 );4033 assert_eq!(queued.decision.decision_id, 3);4034 assert_eq!(queued.decision.state, AdaptiveBitrateStateV1::Decrease);4035 assert!(coordinator.submit(queued).is_none());4036 assert_eq!(tokio::time::Instant::now(), evaluation_tick);4037 assert_eq!(coordinator.acknowledgements.len(), 1);4038 assert_eq!(4039 coordinator.queued.unwrap().decision.decision_id,4040 queued.decision.decision_id4041 );40424043 // Revision one is acknowledged only after another feedback window was4044 // evaluated. The decision retains that exact readback result.4045 harness.status.send_modify(|status| {4046 status.applied_bitrate_revision = Some(1);4047 status.applied_bitrate_kbps = Some(first.decision.target_kbps);4048 });4049 let commit_result = coordinator.acknowledgements.join_next().await;4050 let completed = coordinator.complete(commit_result).unwrap();4051 assert_eq!(completed.plan.decision.decision_id, 2);4052 assert!(completed.plan.decision.applied);4053 assert_eq!(4054 completed.applied_bitrate_kbps,4055 Some(first.decision.target_kbps)4056 );40574058 assert!(coordinator.start_queued().is_none());4059 assert!(coordinator.has_pending());4060 assert_eq!(coordinator.acknowledgements.len(), 1);40614062 let session_id = media.session_id;4063 drop(media);4064 assert!(!sessions.is_active(remote, session_id));4065 let session_end_result = coordinator.acknowledgements.join_next().await;4066 let session_end = coordinator.complete(session_end_result).unwrap();4067 assert!(session_end.generation_ended);4068 coordinator.abort_and_drain().await;4069 assert!(!coordinator.has_pending());4070 assert!(coordinator.queued.is_none());4071 assert!(coordinator.acknowledgements.is_empty());4072 }40734074 4075 async fn timing_out_commits_keep_uniform_hysteresis_and_bounded_work() {4076 let sessions = Arc::new(SessionRegistry::default());4077 let remote = endpoint(1);4078 let media = sessions4079 .claim(remote, [5; 16], InvitationGrants::VIEW)4080 .unwrap();4081 let harness = crate::source::EncoderControlTestHarness::new();4082 sessions4083 .install_encoder_control(remote, media.session_id, Some(harness.control.clone()))4084 .unwrap();4085 let mut coordinator = AdaptiveCommitCoordinator::new(4086 Arc::clone(&sessions),4087 remote,4088 media.session_id,4089 true,4090 12_000,4091 Some(native_dimensions()),4092 );4093 let mut controller =4094 ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());4095 let mut resolution_controller =4096 Some(MotionResolutionPolicy::new(native_dimensions()).unwrap());4097 let observed_at = Instant::now();4098 let virtual_started_at = tokio::time::Instant::now();4099 let mut evaluation_ticks = Vec::new();4100 let mut decision_ids = Vec::new();4101 let mut states = Vec::new();4102 let mut completed = Vec::new();41034104 for report_id in 1..=9 {4105 evaluation_ticks.push(tokio::time::Instant::now());4106 let mut report = clean_feedback(report_id);4107 if report_id > 1 {4108 report.frontend_queue_depth = 2;4109 }4110 let plan = evaluate_commit_plan(4111 &mut controller,4112 &mut resolution_controller,4113 &report,4114 observed_at + Duration::from_secs(report_id - 1),4115 );4116 decision_ids.push(plan.decision.decision_id);4117 states.push(plan.decision.state);4118 if let Some(immediate) = coordinator.submit(plan) {4119 completed.push(immediate);4120 }4121 assert!(4122 coordinator.acknowledgements.len() <= 1,4123 "the coordinator must own at most one commit task"4124 );4125 assert!(4126 coordinator.queued.iter().count() <= 1,4127 "the coordinator must retain only the latest plan"4128 );41294130 tokio::time::advance(Duration::from_secs(1)).await;4131 tokio::task::yield_now().await;4132 if let Some(result) = coordinator.acknowledgements.try_join_next() {4133 completed.push(coordinator.complete(Some(result)).unwrap());4134 if let Some(immediate) = coordinator.start_queued() {4135 completed.push(immediate);4136 }4137 }4138 }41394140 assert_eq!(4141 decision_ids,4142 (1..=9).collect::<Vec<_>>(),4143 "coalescing must not pause or reuse decision IDs"4144 );4145 assert_eq!(4146 states,4147 [4148 AdaptiveBitrateStateV1::Hold,4149 AdaptiveBitrateStateV1::Hold,4150 AdaptiveBitrateStateV1::Decrease,4151 AdaptiveBitrateStateV1::Hold,4152 AdaptiveBitrateStateV1::Decrease,4153 AdaptiveBitrateStateV1::Hold,4154 AdaptiveBitrateStateV1::Decrease,4155 AdaptiveBitrateStateV1::Hold,4156 AdaptiveBitrateStateV1::Decrease,4157 ],4158 "moderate-pressure hysteresis must advance once per feedback window"4159 );4160 for (index, tick) in evaluation_ticks.into_iter().enumerate() {4161 assert_eq!(4162 tick.duration_since(virtual_started_at),4163 Duration::from_secs(index as u64),4164 "feedback evaluation cadence must remain uniform during commit timeouts"4165 );4166 }4167 assert!(4168 completed.iter().any(|acknowledgement| {4169 acknowledgement.plan.decision.decision_id == 34170 && !acknowledgement.plan.decision.applied4171 && acknowledgement.resolution_applied == Some(false)4172 }),4173 "the first bitrate and resolution timeout must retain exact negative acknowledgement"4174 );4175 assert!(coordinator.has_pending());4176 assert_eq!(coordinator.acknowledgements.len(), 1);4177 assert!(coordinator.queued.is_some());41784179 coordinator.abort_and_drain().await;4180 assert!(!coordinator.has_pending());4181 assert!(coordinator.queued.is_none());4182 assert!(coordinator.acknowledgements.is_empty());41834184 let session_id = media.session_id;4185 drop(media);4186 assert!(!sessions.is_active(remote, session_id));4187 }4188}