1use super::session::{2 AdaptiveEncoderProposal, FeedbackLease, MediaV3TelemetrySnapshot, ResolutionEncoderProposal,3 SessionRegistry,4};5use super::*;67const ADAPTIVE_BITRATE_FLOOR_KBPS: u32 = 1_000;8const SHADOW_BITRATE_CEILING_KBPS: u32 = 12_000;9const ADAPTIVE_BITRATE_QUANTUM_KBPS: u32 = 250;10const ADAPTIVE_BITRATE_CLEAN_WINDOWS: u8 = 10;11const ADAPTIVE_BITRATE_MODERATE_WINDOWS: u8 = 2;12const ADAPTIVE_BITRATE_COOLDOWN_WINDOWS: u8 = 10;13const MOTION_RESOLUTION_MODERATE_WINDOWS: u8 = 2;14const MOTION_RESOLUTION_CLEAN_WINDOWS: u8 = 3;15const MOTION_RESOLUTION_ACTIVITY_FPS: u64 = 10;16const MOTION_RESOLUTION_STILL_SETTLE: Duration = Duration::from_secs(2);17const MOTION_RESOLUTION_UPSCALE_COOLDOWN: Duration = Duration::from_secs(3);18const FEEDBACK_FRESH_INTERVAL_MIN_MS: u16 = 750;19const FEEDBACK_FRESH_INTERVAL_MAX_MS: u16 = 1_500;20const FEEDBACK_FRESH_ARRIVAL_GAP_MIN: Duration = Duration::from_millis(750);21const FEEDBACK_STALE_ARRIVAL_GAP: Duration = Duration::from_millis(2_500);22const FEEDBACK_MIN_READ_INTERVAL: Duration = Duration::from_millis(250);23const FEEDBACK_WRITE_TIMEOUT: Duration = Duration::from_secs(1);24// Mirrors the fixed v1 wire bound. Aggregates clamp here so a stalled25// consumer observes a stale 5-second window instead of a deceptively fresh26// latest report.27const MEDIA_FEEDBACK_INTERVAL_MAX_MS: u64 = 5_000;282930enum FeedbackSeverity {31 Clean,32 InsufficientEvidence,33 Moderate,34 Severe,35 Stale,36}373839pub(crate) struct VideoDimensions {40 pub width: u16,41 pub height: u16,42}4344impl VideoDimensions {45 fn three_quarter(self) -> Result<Self> {46 ensure!(47 self.width.is_multiple_of(4) && self.height.is_multiple_of(4),48 "native dimensions must be divisible by four for an exact three-quarter tier"49 );50 let reduced = Self {51 width: self.width / 4 * 3,52 height: self.height / 4 * 3,53 };54 ensure!(55 reduced.width >= 64 && reduced.height >= 64,56 "three-quarter dimensions are below the H.264 minimum"57 );58 ensure!(59 reduced.width.is_multiple_of(2) && reduced.height.is_multiple_of(2),60 "three-quarter H.264 dimensions must be even"61 );62 ensure!(63 u32::from(self.width) * u32::from(reduced.height)64 == u32::from(self.height) * u32::from(reduced.width),65 "three-quarter dimensions must preserve the native aspect ratio"66 );67 Ok(reduced)68 }69}707172enum ResolutionTier {73 Native,74 Reduced,75}767778enum ResolutionTrigger {79 Hold,80 Motion,81 Pressure,82 Recovery,83}848586struct MotionResolutionDecision {87 tier: ResolutionTier,88 target: VideoDimensions,89 trigger: ResolutionTrigger,90 changed: bool,91}9293/// Pure two-tier resolution policy. Fresh sequence progress from Gamescope's94/// damage-driven stream is the bounded motion proxy while the authenticated95/// feedback path supplies pressure. A downshift is safety-biased and96/// immediate; an upscale requires both settled motion and fresh clean97/// feedback, so stale reports can never restore detail.9899pub(crate) struct MotionResolutionPolicy {100 native: VideoDimensions,101 reduced: VideoDimensions,102 tier: ResolutionTier,103 motion_active: bool,104 still_since: Option<Instant>,105 pressure_active: bool,106 moderate_windows: u8,107 clean_windows: u8,108 last_transition_at: Option<Instant>,109 last_motion_sequence: Option<u64>,110}111112impl MotionResolutionPolicy {113 pub(crate) fn new(native: VideoDimensions) -> Result<Self> {114 ensure!(115 native.width >= 64 && native.height >= 64,116 "native dimensions are too small"117 );118 ensure!(119 native.width.is_multiple_of(2) && native.height.is_multiple_of(2),120 "native H.264 dimensions must be even"121 );122 Ok(Self {123 native,124 reduced: native.three_quarter()?,125 tier: ResolutionTier::Native,126 motion_active: false,127 still_since: None,128 pressure_active: false,129 moderate_windows: 0,130 clean_windows: 0,131 last_transition_at: None,132 last_motion_sequence: None,133 })134 }135136 pub(crate) fn target(&self) -> VideoDimensions {137 match self.tier {138 ResolutionTier::Native => self.native,139 ResolutionTier::Reduced => self.reduced,140 }141 }142143 /// Host motion hook. Repeated still observations preserve the original144 /// settle boundary instead of indefinitely postponing recovery.145 146 fn observe_motion(147 &mut self,148 motion_active: bool,149 observed_at: Instant,150 ) -> MotionResolutionDecision {151 self.update_motion(motion_active, observed_at);152 self.evaluate_target(observed_at)153 }154155 fn update_motion(&mut self, motion_active: bool, observed_at: Instant) {156 if motion_active {157 self.motion_active = true;158 self.still_since = None;159 return;160 }161 if self.motion_active || self.still_since.is_none() {162 self.still_since = Some(observed_at);163 }164 self.motion_active = false;165 }166167 /// Observe one classified feedback window. Only fresh moderate/severe168 /// pressure can initiate a downshift, and only fresh clean windows count169 /// toward restoring native detail.170 171 fn observe_feedback(172 &mut self,173 severity: FeedbackSeverity,174 fresh: bool,175 observed_at: Instant,176 ) -> MotionResolutionDecision {177 self.update_feedback(severity, fresh);178 self.evaluate_target(observed_at)179 }180181 /// Gamescope output is damage-driven, so fresh media sequence progress is182 /// a conservative constant-space motion proxy for this first slice. A183 /// future raw-frame observer can call `observe_motion` without changing184 /// pressure hysteresis or encoder actuation.185 fn observe_window(186 &mut self,187 report: &MediaFeedbackReportV1,188 evaluation: &AdaptiveBitrateEvaluation,189 observed_at: Instant,190 ) -> MotionResolutionDecision {191 let previous_sequence = self.last_motion_sequence;192 if let Some(sequence) = report.last_sequence {193 self.last_motion_sequence = Some(194 self.last_motion_sequence195 .map_or(sequence, |last| last.max(sequence)),196 );197 }198 if evaluation.fresh {199 let motion_active = report.last_sequence.is_some_and(|sequence| {200 let delta = previous_sequence.map_or(0, |last| sequence.saturating_sub(last));201 delta.saturating_mul(1_000)202 >= MOTION_RESOLUTION_ACTIVITY_FPS.saturating_mul(u64::from(report.interval_ms))203 });204 self.update_motion(motion_active, observed_at);205 }206 let resolution_severity = if evaluation.pressure_severity == FeedbackSeverity::Clean207 && !evaluation.clean_recovery_evidence208 {209 FeedbackSeverity::InsufficientEvidence210 } else {211 evaluation.pressure_severity212 };213 self.update_feedback(resolution_severity, evaluation.fresh);214 self.evaluate_target(observed_at)215 }216217 fn update_feedback(&mut self, severity: FeedbackSeverity, fresh: bool) {218 if !fresh219 || matches!(220 severity,221 FeedbackSeverity::InsufficientEvidence | FeedbackSeverity::Stale222 )223 {224 self.moderate_windows = 0;225 self.clean_windows = 0;226 return;227 }228229 match severity {230 FeedbackSeverity::Severe => {231 self.pressure_active = true;232 self.moderate_windows = 0;233 self.clean_windows = 0;234 }235 FeedbackSeverity::Moderate => {236 self.clean_windows = 0;237 self.moderate_windows = self.moderate_windows.saturating_add(1);238 if self.moderate_windows >= MOTION_RESOLUTION_MODERATE_WINDOWS {239 self.moderate_windows = 0;240 self.pressure_active = true;241 }242 }243 FeedbackSeverity::Clean => {244 self.moderate_windows = 0;245 self.clean_windows = self.clean_windows.saturating_add(1);246 if self.clean_windows >= MOTION_RESOLUTION_CLEAN_WINDOWS {247 self.pressure_active = false;248 }249 }250 FeedbackSeverity::InsufficientEvidence | FeedbackSeverity::Stale => {}251 }252 }253254 fn evaluate_target(&mut self, observed_at: Instant) -> MotionResolutionDecision {255 if self.motion_active {256 self.downshift(observed_at, ResolutionTrigger::Motion)257 } else if self.pressure_active {258 self.downshift(observed_at, ResolutionTrigger::Pressure)259 } else {260 self.maybe_restore(observed_at)261 }262 }263264 fn downshift(265 &mut self,266 observed_at: Instant,267 trigger: ResolutionTrigger,268 ) -> MotionResolutionDecision {269 let changed = self.tier != ResolutionTier::Reduced;270 if changed {271 self.tier = ResolutionTier::Reduced;272 self.last_transition_at = Some(observed_at);273 }274 MotionResolutionDecision {275 tier: self.tier,276 target: self.target(),277 trigger,278 changed,279 }280 }281282 fn maybe_restore(&mut self, observed_at: Instant) -> MotionResolutionDecision {283 let still_settled = !self.motion_active284 && self.still_since.is_some_and(|still_since| {285 observed_at.saturating_duration_since(still_since) >= MOTION_RESOLUTION_STILL_SETTLE286 });287 let cooldown_complete = self.last_transition_at.is_none_or(|last| {288 observed_at.saturating_duration_since(last) >= MOTION_RESOLUTION_UPSCALE_COOLDOWN289 });290 let clean = self.clean_windows >= MOTION_RESOLUTION_CLEAN_WINDOWS;291 if self.tier == ResolutionTier::Reduced292 && still_settled293 && cooldown_complete294 && clean295 && !self.pressure_active296 {297 self.tier = ResolutionTier::Native;298 self.clean_windows = 0;299 self.last_transition_at = Some(observed_at);300 return MotionResolutionDecision {301 tier: self.tier,302 target: self.target(),303 trigger: ResolutionTrigger::Recovery,304 changed: true,305 };306 }307 self.hold()308 }309310 fn hold(&self) -> MotionResolutionDecision {311 MotionResolutionDecision {312 tier: self.tier,313 target: self.target(),314 trigger: ResolutionTrigger::Hold,315 changed: false,316 }317 }318}319320fn motion_resolution_policy(width: u32, height: u32) -> Result<Option<MotionResolutionPolicy>> {321 let native = VideoDimensions {322 width: u16::try_from(width).context("native width exceeds protocol")?,323 height: u16::try_from(height).context("native height exceeds protocol")?,324 };325 match MotionResolutionPolicy::new(native) {326 Ok(policy) => Ok(Some(policy)),327 Err(error) => {328 warn!(329 %error,330 width,331 height,332 "motion resolution disabled because this native mode has no exact reduced tier"333 );334 Ok(None)335 }336 }337}338339fn motion_resolution_policy_for_encoder(340 control: &EncoderControl,341) -> Result<Option<MotionResolutionPolicy>> {342 let (width, height) = control.initial_dimensions();343 motion_resolution_policy(u32::from(width), u32::from(height))344}345346347struct FeedbackIngressCursor {348 last_report_id: Option<u64>,349 interval_ms: u64,350 transport_dropped: u64,351 frontend_dropped: u64,352 decoder_dropped: u64,353 presenter_dropped: u64,354 resync_reports: u64,355}356357358struct CumulativeFeedbackIngress {359 latest: Option<MediaFeedbackReportV1>,360 interval_ms: u64,361 transport_dropped: u64,362 frontend_dropped: u64,363 decoder_dropped: u64,364 presenter_dropped: u64,365 resync_reports: u64,366}367368impl CumulativeFeedbackIngress {369 fn observe(&mut self, report: MediaFeedbackReportV1) {370 self.interval_ms = self.interval_ms.wrapping_add(u64::from(report.interval_ms));371 self.transport_dropped = self372 .transport_dropped373 .wrapping_add(u64::from(report.transport_dropped_delta));374 self.frontend_dropped = self375 .frontend_dropped376 .wrapping_add(u64::from(report.frontend_dropped_delta));377 self.decoder_dropped = self378 .decoder_dropped379 .wrapping_add(u64::from(report.decoder_dropped_delta));380 self.presenter_dropped = self381 .presenter_dropped382 .wrapping_add(u64::from(report.presenter_dropped_delta));383 if report.flags.contains(MediaFeedbackFlags::RESYNC_ACTIVE) {384 self.resync_reports = self.resync_reports.wrapping_add(1);385 }386 self.latest = Some(report);387 }388389 fn report_since(390 &self,391 cursor: FeedbackIngressCursor,392 ) -> Result<Option<(MediaFeedbackReportV1, FeedbackIngressCursor)>> {393 let Some(latest) = self.latest else {394 return Ok(None);395 };396 if cursor397 .last_report_id398 .is_some_and(|report_id| latest.report_id <= report_id)399 {400 return Ok(None);401 }402 let mut report = latest;403 report.interval_ms = u16::try_from(404 self.interval_ms405 .wrapping_sub(cursor.interval_ms)406 .min(MEDIA_FEEDBACK_INTERVAL_MAX_MS),407 )408 .expect("feedback interval is clamped to u16");409 report.transport_dropped_delta =410 cumulative_feedback_delta(self.transport_dropped, cursor.transport_dropped);411 report.frontend_dropped_delta =412 cumulative_feedback_delta(self.frontend_dropped, cursor.frontend_dropped);413 report.decoder_dropped_delta =414 cumulative_feedback_delta(self.decoder_dropped, cursor.decoder_dropped);415 report.presenter_dropped_delta =416 cumulative_feedback_delta(self.presenter_dropped, cursor.presenter_dropped);417 report.flags = if self.resync_reports.wrapping_sub(cursor.resync_reports) > 0 {418 MediaFeedbackFlags::RESYNC_ACTIVE419 } else {420 MediaFeedbackFlags::NONE421 };422 report.validate()?;423 Ok(Some((424 report,425 FeedbackIngressCursor {426 last_report_id: Some(latest.report_id),427 interval_ms: self.interval_ms,428 transport_dropped: self.transport_dropped,429 frontend_dropped: self.frontend_dropped,430 decoder_dropped: self.decoder_dropped,431 presenter_dropped: self.presenter_dropped,432 resync_reports: self.resync_reports,433 },434 )))435 }436}437438fn cumulative_feedback_delta(total: u64, baseline: u64) -> u32 {439 u32::try_from(total.wrapping_sub(baseline).min(u64::from(u32::MAX)))440 .expect("feedback delta is clamped to u32")441}442443444struct ShadowBitrateController {445 floor_kbps: u32,446 ceiling_kbps: u32,447 target_kbps: u32,448 next_decision_id: u64,449 last_report_id: Option<u64>,450 last_report_received_at: Option<Instant>,451 previous_telemetry: MediaV3TelemetrySnapshot,452 baseline_rtt_micros: Option<u64>,453 last_sequence: Option<u64>,454 moderate_windows: u8,455 clean_windows: u8,456 cooldown_windows: u8,457}458459460struct AdaptiveBitrateEvaluation {461 decision: AdaptiveBitrateDecisionV1,462 pressure_severity: FeedbackSeverity,463 clean_recovery_evidence: bool,464 fresh: bool,465}466467impl ShadowBitrateController {468 fn new(configured_ceiling_kbps: u32, telemetry: MediaV3TelemetrySnapshot) -> Self {469 let ceiling_kbps = quantize_down(configured_ceiling_kbps).max(ADAPTIVE_BITRATE_FLOOR_KBPS);470 Self {471 floor_kbps: ADAPTIVE_BITRATE_FLOOR_KBPS,472 ceiling_kbps,473 target_kbps: ceiling_kbps,474 next_decision_id: 1,475 last_report_id: None,476 last_report_received_at: None,477 previous_telemetry: telemetry,478 baseline_rtt_micros: None,479 last_sequence: None,480 moderate_windows: 0,481 clean_windows: 0,482 cooldown_windows: 0,483 }484 }485486 fn evaluate(487 &mut self,488 report: &MediaFeedbackReportV1,489 telemetry: MediaV3TelemetrySnapshot,490 received_at: Instant,491 ) -> Result<AdaptiveBitrateEvaluation> {492 ensure!(493 self.last_report_id494 .is_none_or(|last| report.report_id > last),495 "feedback report IDs must increase monotonically"496 );497 let (severity, pressure_severity, reasons, stale, trusted_host_pressure) =498 self.classify(report, telemetry, received_at);499 let mut state = AdaptiveBitrateStateV1::Hold;500501 if self.cooldown_windows > 0 {502 self.cooldown_windows -= 1;503 }504 match severity {505 FeedbackSeverity::Severe => {506 self.moderate_windows = 0;507 self.clean_windows = 0;508 self.cooldown_windows = ADAPTIVE_BITRATE_COOLDOWN_WINDOWS;509 let next =510 quantize_down(self.target_kbps.saturating_mul(3) / 4).max(self.floor_kbps);511 if next < self.target_kbps {512 self.target_kbps = next;513 state = AdaptiveBitrateStateV1::Decrease;514 }515 }516 FeedbackSeverity::Moderate => {517 self.moderate_windows = self.moderate_windows.saturating_add(1);518 self.clean_windows = 0;519 if self.moderate_windows >= ADAPTIVE_BITRATE_MODERATE_WINDOWS {520 self.moderate_windows = 0;521 self.cooldown_windows = ADAPTIVE_BITRATE_COOLDOWN_WINDOWS;522 let next =523 quantize_down(self.target_kbps.saturating_mul(4) / 5).max(self.floor_kbps);524 if next < self.target_kbps {525 self.target_kbps = next;526 state = AdaptiveBitrateStateV1::Decrease;527 }528 }529 }530 FeedbackSeverity::Clean => {531 self.moderate_windows = 0;532 self.clean_windows = self.clean_windows.saturating_add(1);533 if self.clean_windows >= ADAPTIVE_BITRATE_CLEAN_WINDOWS534 && self.cooldown_windows == 0535 {536 self.clean_windows = 0;537 let maximum_step =538 (self.target_kbps / 20).clamp(ADAPTIVE_BITRATE_QUANTUM_KBPS, 500);539 let next = quantize_down(540 self.target_kbps541 .saturating_add(maximum_step)542 .min(self.ceiling_kbps),543 );544 if next > self.target_kbps {545 self.target_kbps = next;546 state = AdaptiveBitrateStateV1::Increase;547 }548 }549 }550 FeedbackSeverity::InsufficientEvidence | FeedbackSeverity::Stale => {551 self.moderate_windows = 0;552 self.clean_windows = 0;553 }554 }555556 self.target_kbps = self.target_kbps.clamp(self.floor_kbps, self.ceiling_kbps);557 let decision = AdaptiveBitrateDecisionV1 {558 decision_id: self.next_decision_id,559 report_id: report.report_id,560 target_kbps: self.target_kbps,561 floor_kbps: self.floor_kbps,562 ceiling_kbps: self.ceiling_kbps,563 state,564 reasons,565 applied: false,566 };567 decision.validate()?;568 self.next_decision_id = self569 .next_decision_id570 .checked_add(1)571 .context("adaptive bitrate decision ID exhausted")?;572 self.last_report_id = Some(report.report_id);573 self.last_report_received_at = Some(received_at);574 if !stale575 || !trusted_host_pressure576 || state == AdaptiveBitrateStateV1::Decrease577 || self.target_kbps == self.floor_kbps578 {579 self.previous_telemetry = telemetry;580 }581 if !stale && let Some(sequence) = report.last_sequence {582 self.last_sequence = Some(583 self.last_sequence584 .map_or(sequence, |last| last.max(sequence)),585 );586 }587 Ok(AdaptiveBitrateEvaluation {588 decision,589 pressure_severity,590 clean_recovery_evidence: severity == FeedbackSeverity::Clean,591 fresh: !stale,592 })593 }594595 596 fn decide(597 &mut self,598 report: &MediaFeedbackReportV1,599 telemetry: MediaV3TelemetrySnapshot,600 received_at: Instant,601 ) -> Result<AdaptiveBitrateDecisionV1> {602 self.evaluate(report, telemetry, received_at)603 .map(|evaluation| evaluation.decision)604 }605606 fn classify(607 &mut self,608 report: &MediaFeedbackReportV1,609 telemetry: MediaV3TelemetrySnapshot,610 received_at: Instant,611 ) -> (612 FeedbackSeverity,613 FeedbackSeverity,614 AdaptiveBitrateReasonFlagsV1,615 bool,616 bool,617 ) {618 let stale = !(FEEDBACK_FRESH_INTERVAL_MIN_MS..=FEEDBACK_FRESH_INTERVAL_MAX_MS)619 .contains(&report.interval_ms)620 || self.last_report_received_at.is_none()621 || self.last_report_received_at.is_some_and(|last| {622 let gap = received_at.saturating_duration_since(last);623 !(FEEDBACK_FRESH_ARRIVAL_GAP_MIN..=FEEDBACK_STALE_ARRIVAL_GAP).contains(&gap)624 });625 let mut severity = FeedbackSeverity::Clean;626 let mut reasons = AdaptiveBitrateReasonFlagsV1::NONE;627628 let lost_delta = telemetry629 .selected_path_lost_packets630 .saturating_sub(self.previous_telemetry.selected_path_lost_packets);631 let congestion_delta = telemetry632 .selected_path_congestion_events633 .saturating_sub(self.previous_telemetry.selected_path_congestion_events);634 let cancellation_delta = telemetry635 .scheduler_cancellations636 .saturating_sub(self.previous_telemetry.scheduler_cancellations);637 let failure_delta = telemetry638 .send_failures639 .saturating_sub(self.previous_telemetry.send_failures);640641 if failure_delta > 0 || lost_delta > 0 || cancellation_delta >= 2 {642 add_feedback_signal(643 &mut severity,644 &mut reasons,645 FeedbackSeverity::Severe,646 AdaptiveBitrateReasonFlagsV1::LOSS_OR_CANCELLATION,647 );648 } else if cancellation_delta > 0 || congestion_delta > 0 {649 add_feedback_signal(650 &mut severity,651 &mut reasons,652 FeedbackSeverity::Moderate,653 AdaptiveBitrateReasonFlagsV1::LOSS_OR_CANCELLATION,654 );655 }656 if cancellation_delta >= 2 || failure_delta > 0 {657 add_feedback_signal(658 &mut severity,659 &mut reasons,660 FeedbackSeverity::Severe,661 AdaptiveBitrateReasonFlagsV1::SENDER_BACKPRESSURE,662 );663 } else if cancellation_delta > 0 {664 add_feedback_signal(665 &mut severity,666 &mut reasons,667 FeedbackSeverity::Moderate,668 AdaptiveBitrateReasonFlagsV1::SENDER_BACKPRESSURE,669 );670 }671672 let rtt = telemetry.selected_path_rtt_micros;673 if rtt > 0 {674 let baseline = self.baseline_rtt_micros.get_or_insert(rtt);675 if rtt < *baseline {676 *baseline = rtt;677 } else if rtt >= baseline.saturating_mul(2).max(80_000) {678 add_feedback_signal(679 &mut severity,680 &mut reasons,681 FeedbackSeverity::Severe,682 AdaptiveBitrateReasonFlagsV1::RTT_INFLATION,683 );684 } else if rtt >= baseline.saturating_mul(3) / 2 {685 add_feedback_signal(686 &mut severity,687 &mut reasons,688 FeedbackSeverity::Moderate,689 AdaptiveBitrateReasonFlagsV1::RTT_INFLATION,690 );691 }692 }693694 let trusted_host_pressure = severity != FeedbackSeverity::Clean;695 if stale {696 reasons = reasons.union(AdaptiveBitrateReasonFlagsV1::FEEDBACK_STALE);697 if !trusted_host_pressure {698 severity = FeedbackSeverity::Stale;699 }700 return (severity, severity, reasons, true, trusted_host_pressure);701 }702703 let receiver_drops = report704 .transport_dropped_delta705 .saturating_add(report.frontend_dropped_delta)706 .saturating_add(report.decoder_dropped_delta)707 .saturating_add(report.presenter_dropped_delta);708 let queue_severe = queue_ratio_at_least(709 report.frontend_queue_depth,710 report.frontend_queue_capacity,711 3,712 4,713 ) || queue_ratio_at_least(714 report.decode_queue_depth,715 report.decode_queue_capacity,716 3,717 4,718 ) || queue_ratio_at_least(719 report.presenter_queue_depth,720 report.presenter_queue_capacity,721 3,722 4,723 );724 let queue_moderate = queue_ratio_at_least(725 report.frontend_queue_depth,726 report.frontend_queue_capacity,727 1,728 2,729 ) || queue_ratio_at_least(730 report.decode_queue_depth,731 report.decode_queue_capacity,732 1,733 2,734 ) || queue_ratio_at_least(735 report.presenter_queue_depth,736 report.presenter_queue_capacity,737 1,738 2,739 );740 if report.flags.contains(MediaFeedbackFlags::RESYNC_ACTIVE)741 || queue_severe742 || receiver_drops >= 4743 {744 add_feedback_signal(745 &mut severity,746 &mut reasons,747 FeedbackSeverity::Severe,748 AdaptiveBitrateReasonFlagsV1::RECEIVER_QUEUE,749 );750 } else if queue_moderate || receiver_drops > 0 {751 add_feedback_signal(752 &mut severity,753 &mut reasons,754 FeedbackSeverity::Moderate,755 AdaptiveBitrateReasonFlagsV1::RECEIVER_QUEUE,756 );757 }758 if report.decode_queue_depth == report.decode_queue_capacity759 || report.decoder_dropped_delta > 0760 {761 add_feedback_signal(762 &mut severity,763 &mut reasons,764 FeedbackSeverity::Severe,765 AdaptiveBitrateReasonFlagsV1::DECODE_BACKLOG,766 );767 } else if report.decode_queue_depth >= 2 {768 add_feedback_signal(769 &mut severity,770 &mut reasons,771 FeedbackSeverity::Moderate,772 AdaptiveBitrateReasonFlagsV1::DECODE_BACKLOG,773 );774 }775776 let delivery_p95 = report.transport_delivery_p95_ms.unwrap_or(0);777 let decode_p95 = report.decode_p95_ms.unwrap_or(0);778 let presentation_p95 = report.presentation_p95_ms.unwrap_or(0);779 if delivery_p95 >= 200 || decode_p95 >= 100 || presentation_p95 >= 100 {780 add_feedback_signal(781 &mut severity,782 &mut reasons,783 FeedbackSeverity::Severe,784 AdaptiveBitrateReasonFlagsV1::DELIVERY_LATENCY,785 );786 } else if delivery_p95 >= 100 || decode_p95 >= 50 || presentation_p95 >= 50 {787 add_feedback_signal(788 &mut severity,789 &mut reasons,790 FeedbackSeverity::Moderate,791 AdaptiveBitrateReasonFlagsV1::DELIVERY_LATENCY,792 );793 }794795 let pressure_severity = severity;796 if severity == FeedbackSeverity::Clean {797 let sequence_progressed = self798 .last_sequence799 .zip(report.last_sequence)800 .is_some_and(|(last, current)| current > last);801 let complete_latency = report.transport_delivery_p95_ms.is_some()802 && report.decode_p95_ms.is_some()803 && report.presentation_p95_ms.is_some();804 if sequence_progressed && complete_latency {805 if self.target_kbps < self.ceiling_kbps {806 reasons = reasons.union(AdaptiveBitrateReasonFlagsV1::CLEAN_RECOVERY);807 }808 } else {809 severity = FeedbackSeverity::InsufficientEvidence;810 }811 }812 (813 severity,814 pressure_severity,815 reasons,816 false,817 trusted_host_pressure,818 )819 }820}821822fn add_feedback_signal(823 severity: &mut FeedbackSeverity,824 reasons: &mut AdaptiveBitrateReasonFlagsV1,825 signal: FeedbackSeverity,826 reason: AdaptiveBitrateReasonFlagsV1,827) {828 *reasons = reasons.union(reason);829 if signal == FeedbackSeverity::Severe830 || (*severity == FeedbackSeverity::Clean && signal == FeedbackSeverity::Moderate)831 {832 *severity = signal;833 }834}835836fn quantize_down(kbps: u32) -> u32 {837 kbps / ADAPTIVE_BITRATE_QUANTUM_KBPS * ADAPTIVE_BITRATE_QUANTUM_KBPS838}839840fn queue_ratio_at_least(depth: u8, capacity: u8, numerator: u16, denominator: u16) -> bool {841 capacity > 0842 && u16::from(depth).saturating_mul(denominator)843 >= u16::from(capacity).saturating_mul(numerator)844}845846847struct AdaptiveCommitPlan {848 decision: AdaptiveBitrateDecisionV1,849 telemetry: MediaV3TelemetrySnapshot,850 resolution_decision: Option<MotionResolutionDecision>,851 force_keyframe: bool,852}853854855struct AdaptiveCommitAcknowledgement {856 plan: AdaptiveCommitPlan,857 generation_ended: bool,858 recovery_acknowledged: Option<bool>,859 resolution_applied: Option<bool>,860 applied_bitrate_kbps: Option<u32>,861 applied_resolution: Option<VideoDimensions>,862}863864impl AdaptiveCommitAcknowledgement {865 fn immediate(plan: AdaptiveCommitPlan) -> Self {866 Self {867 plan,868 generation_ended: false,869 recovery_acknowledged: None,870 resolution_applied: None,871 applied_bitrate_kbps: None,872 applied_resolution: None,873 }874 }875}876877/// Bounded adaptive-control actuator.878///879/// Feedback evaluation remains synchronous and deterministic, while this880/// coordinator owns the slower encoder acknowledgements. At most one commit881/// task and one latest coalesced plan exist. A queued plan carries the latest882/// desired bitrate and resolution even when its feedback decision is a hold,883/// so a slow or failed predecessor cannot strand an uncommitted target.884struct AdaptiveCommitCoordinator {885 sessions: Arc<SessionRegistry>,886 remote: EndpointId,887 session_id: u64,888 actuation_enabled: bool,889 desired_bitrate_kbps: u32,890 requested_bitrate_kbps: u32,891 committed_bitrate_kbps: u32,892 desired_resolution: Option<VideoDimensions>,893 requested_resolution: Option<VideoDimensions>,894 committed_resolution: Option<VideoDimensions>,895 pending: bool,896 queued: Option<AdaptiveCommitPlan>,897 acknowledgements: tokio::task::JoinSet<AdaptiveCommitAcknowledgement>,898}899900impl AdaptiveCommitCoordinator {901 fn new(902 sessions: Arc<SessionRegistry>,903 remote: EndpointId,904 session_id: u64,905 actuation_enabled: bool,906 committed_bitrate_kbps: u32,907 committed_resolution: Option<VideoDimensions>,908 ) -> Self {909 Self {910 sessions,911 remote,912 session_id,913 actuation_enabled,914 desired_bitrate_kbps: committed_bitrate_kbps,915 requested_bitrate_kbps: committed_bitrate_kbps,916 committed_bitrate_kbps,917 desired_resolution: committed_resolution,918 requested_resolution: committed_resolution,919 committed_resolution,920 pending: false,921 queued: None,922 acknowledgements: tokio::task::JoinSet::new(),923 }924 }925926 fn has_pending(&self) -> bool {927 self.pending928 }929930 fn bitrate_is_applied(&self, target_kbps: u32) -> bool {931 self.actuation_enabled932 && !self.pending933 && target_kbps == self.requested_bitrate_kbps934 && target_kbps == self.committed_bitrate_kbps935 }936937 fn submit(&mut self, mut plan: AdaptiveCommitPlan) -> Option<AdaptiveCommitAcknowledgement> {938 self.desired_bitrate_kbps = plan.decision.target_kbps;939 if let Some(resolution) = plan.resolution_decision {940 self.desired_resolution = Some(resolution.target);941 }942 if self.pending {943 if let Some(queued) = self.queued {944 // The latest decision replaces older diagnostics, but a945 // recovery barrier requested by any coalesced desired update946 // must survive until that desired state is actuated.947 plan.force_keyframe |= queued.force_keyframe;948 }949 self.queued = Some(plan);950 return None;951 }952 if let Some(queued) = self.queued.take() {953 // A completion can deliberately defer this slot while an unread954 // feedback generation is waiting. The newly evaluated plan is the955 // actual latest state, but it must inherit any recovery barrier956 // carried by the superseded slot.957 plan.force_keyframe |= queued.force_keyframe;958 }959 self.start(plan)960 }961962 fn start_queued(&mut self) -> Option<AdaptiveCommitAcknowledgement> {963 let plan = self.queued.take()?;964 self.start(plan)965 }966967 fn start(&mut self, plan: AdaptiveCommitPlan) -> Option<AdaptiveCommitAcknowledgement> {968 if !self.actuation_enabled {969 return Some(AdaptiveCommitAcknowledgement::immediate(plan));970 }971972 let desired_resolution = plan.resolution_decision.map(|change| change.target);973 debug_assert_eq!(plan.decision.target_kbps, self.desired_bitrate_kbps);974 if desired_resolution.is_some() {975 debug_assert_eq!(desired_resolution, self.desired_resolution);976 }977978 // A proposal mutates EncoderControl's persistent desired watch before979 // its acknowledgement arrives. Reconcile against both the last980 // requested and last committed values: after a timeout, returning to981 // the old committed target must overwrite the still-pending stale982 // request rather than allowing it to apply later.983 let reconciles_resolution = desired_resolution.is_some()984 && (desired_resolution != self.requested_resolution985 || desired_resolution != self.committed_resolution);986 let resolution_proposal = if reconciles_resolution {987 let target = desired_resolution.expect("resolution target was checked as present");988 match self989 .sessions990 .propose_resolution_update(self.remote, self.session_id, target)991 {992 Ok(Some(proposal)) => {993 self.requested_resolution = Some(target);994 Some(proposal)995 }996 Ok(None) => {997 warn!(998 remote = %self.remote,999 session_id = self.session_id,1000 "active adaptive session lost encoder control; retaining committed resolution"1001 );1002 None1003 }1004 Err(error) => {1005 warn!(1006 remote = %self.remote,1007 session_id = self.session_id,1008 %error,1009 "adaptive resolution proposal failed; retaining committed resolution"1010 );1011 None1012 }1013 }1014 } else {1015 None1016 };10171018 let reconciles_bitrate = plan.decision.target_kbps != self.requested_bitrate_kbps1019 || plan.decision.target_kbps != self.committed_bitrate_kbps;1020 let force_keyframe = plan.force_keyframe && resolution_proposal.is_none();1021 let bitrate_proposal = if reconciles_bitrate || force_keyframe {1022 match self.sessions.propose_adaptive_encoder_update(1023 self.remote,1024 self.session_id,1025 plan.decision.target_kbps,1026 force_keyframe,1027 ) {1028 Ok(Some(proposal)) => {1029 self.requested_bitrate_kbps = plan.decision.target_kbps;1030 Some(proposal)1031 }1032 Ok(None) => {1033 warn!(1034 remote = %self.remote,1035 session_id = self.session_id,1036 "active adaptive session lost encoder control; retaining committed bitrate"1037 );1038 None1039 }1040 Err(error) => {1041 warn!(1042 remote = %self.remote,1043 session_id = self.session_id,1044 %error,1045 "adaptive encoder proposal failed; retaining committed bitrate"1046 );1047 None1048 }1049 }1050 } else {1051 None1052 };10531054 if bitrate_proposal.is_none() && resolution_proposal.is_none() {1055 return Some(AdaptiveCommitAcknowledgement::immediate(plan));1056 }10571058 self.pending = true;1059 let sessions = Arc::clone(&self.sessions);1060 let remote = self.remote;1061 let session_id = self.session_id;1062 self.acknowledgements.spawn(async move {1063 run_adaptive_commit(1064 sessions,1065 remote,1066 session_id,1067 plan,1068 bitrate_proposal,1069 resolution_proposal,1070 )1071 .await1072 });1073 None1074 }10751076 fn complete(1077 &mut self,1078 result: Option<std::result::Result<AdaptiveCommitAcknowledgement, tokio::task::JoinError>>,1079 ) -> Result<AdaptiveCommitAcknowledgement> {1080 ensure!(1081 self.pending,1082 "adaptive commit completed without pending work"1083 );1084 self.pending = false;1085 let acknowledgement = result1086 .context("adaptive commit task ended without a result")?1087 .context("adaptive commit task failed")?;1088 if let Some(applied) = acknowledgement.applied_bitrate_kbps {1089 self.committed_bitrate_kbps = applied;1090 }1091 if let Some(applied) = acknowledgement.applied_resolution {1092 self.committed_resolution = Some(applied);1093 }1094 Ok(acknowledgement)1095 }10961097 async fn abort_and_drain(&mut self) {1098 self.pending = false;1099 self.queued = None;1100 self.acknowledgements.abort_all();1101 while self.acknowledgements.join_next().await.is_some() {}1102 }1103}11041105pub(super) async fn serve_media_feedback(1106 connection: Connection,1107 config: &HostConfig,1108 sessions: &Arc<SessionRegistry>,1109 authorization: &AuthorizationPolicy,1110) -> Result<()> {1111 let remote = connection.remote_id();1112 let handshake_permit = sessions1113 .pending_handshakes1114 .try_acquire()1115 .context("too many pending handshakes")?;1116 let (mut send, mut recv) = tokio::time::timeout(HANDSHAKE_TIMEOUT, connection.accept_bi())1117 .await1118 .context("timed out accepting media feedback stream")?1119 .context("accepting media feedback stream")?;1120 let hello = receive_hello(&mut recv, Capability::VideoH264).await?;1121 drop(handshake_permit);1122 debug!(%remote, agent = %hello.agent, "media feedback hello received");1123 ensure!(1124 hello.invitation.is_none(),1125 "invitations are accepted only on the first media handshake"1126 );11271128 let grants = match authorization.authorize_or_redeem(remote, None, unix_timestamp_now()?) {1129 Ok(grants) => grants,1130 Err(error) => {1131 send_rejection(&mut send, "Portal peer is not authorized").await?;1132 return Err(error.context("authorizing media feedback peer"));1133 }1134 };1135 ensure!(1136 grants.contains(InvitationGrants::VIEW),1137 "authorized feedback peer lacks view permission"1138 );1139 let ceiling_kbps = adaptive_bitrate_ceiling_kbps(config)?;1140 let lease = match sessions.claim_feedback(remote, hello.nonce) {1141 Ok(lease) => lease,1142 Err(error) => {1143 send_rejection(&mut send, error.to_string()).await?;1144 return Err(error);1145 }1146 };1147 let encoder_actuation_available =1148 adaptive_bitrate_actuation_enabled(config) && lease.encoder_control.is_some();11491150 write_host_hello(1151 &mut send,1152 &HostHello::accepted(1153 lease.session_id,1154 negotiated_capabilities(&hello, MEDIA_CAPABILITIES),1155 ),1156 )1157 .await?;1158 info!(1159 %remote,1160 session_id = lease.session_id,1161 ceiling_kbps,1162 applied = false,1163 mode = if encoder_actuation_available { "active" } else { "shadow" },1164 "media feedback client accepted for bitrate control"1165 );11661167 let (feedback_sender, mut feedback_receiver) =1168 tokio::sync::watch::channel(CumulativeFeedbackIngress::default());1169 let reader_task = tokio::spawn(forward_media_feedback_reports(recv, feedback_sender));1170 let connection_closed = async {1171 let closed = connection.closed().await;1172 debug!(%remote, ?closed, "media feedback connection closed");1173 };1174 let result = run_media_feedback_loop(1175 &mut send,1176 connection_closed,1177 remote,1178 sessions,1179 &lease,1180 encoder_actuation_available,1181 ceiling_kbps,1182 &mut feedback_receiver,1183 reader_task,1184 )1185 .await;1186 let _ = send.finish();1187 drop(lease);1188 info!(%remote, "media feedback client released");1189 result1190}119111921193async fn run_media_feedback_loop<W, C>(1194 send: &mut W,1195 connection_closed: C,1196 remote: EndpointId,1197 sessions: &Arc<SessionRegistry>,1198 lease: &FeedbackLease,1199 encoder_actuation_available: bool,1200 ceiling_kbps: u32,1201 feedback_receiver: &mut tokio::sync::watch::Receiver<CumulativeFeedbackIngress>,1202 mut reader_task: tokio::task::JoinHandle<Result<()>>,1203) -> Result<()>1204where1205 W: tokio::io::AsyncWrite + Unpin,1206 C: Future<Output = ()>,1207{1208 let mut reader_finished = false;1209 let mut receiver_open = true;1210 let mut controller = ShadowBitrateController::new(ceiling_kbps, lease.telemetry.snapshot());1211 let mut resolution_controller = if encoder_actuation_available {1212 lease1213 .encoder_control1214 .as_ref()1215 .map(motion_resolution_policy_for_encoder)1216 .transpose()?1217 .flatten()1218 } else {1219 None1220 };1221 let initial_resolution = resolution_controller1222 .as_ref()1223 .map(MotionResolutionPolicy::target);1224 let mut commit_coordinator = AdaptiveCommitCoordinator::new(1225 Arc::clone(sessions),1226 remote,1227 lease.session_id,1228 encoder_actuation_available,1229 ceiling_kbps,1230 initial_resolution,1231 );1232 let mut feedback_cursor = FeedbackIngressCursor::default();1233 tokio::pin!(connection_closed);12341235 // Keep cleanup on every error path: `?` below exits this async block, not1236 // the outer function that owns the reader and actuator tasks.1237 let result: Result<()> = async {1238 loop {1239 let session_changed = sessions.session_changed.notified();1240 tokio::pin!(session_changed);1241 session_changed.as_mut().enable();1242 if !sessions.is_active(remote, lease.session_id) {1243 break Ok(());1244 }1245 tokio::select! {1246 biased;1247 reader_result = &mut reader_task, if !reader_finished => {1248 reader_finished = true;1249 match reader_result {1250 Ok(Ok(())) => {1251 debug!(%remote, "media feedback report stream closed");1252 }1253 Ok(Err(error)) => {1254 break Err(error).context("reading media feedback reports");1255 }1256 Err(error) => {1257 break Err(error).context("media feedback reader task failed");1258 }1259 }1260 }1261 commit_result = commit_coordinator.acknowledgements.join_next(),1262 if commit_coordinator.has_pending() =>1263 {1264 let acknowledgement = commit_coordinator.complete(commit_result)?;1265 if acknowledgement.generation_ended {1266 break Ok(());1267 }1268 log_adaptive_acknowledgement(1269 remote,1270 lease.session_id,1271 encoder_actuation_available,1272 &commit_coordinator,1273 &acknowledgement,1274 );1275 // If a newer report is already waiting, let its evaluation1276 // replace the queued actuator slot before starting more1277 // work. This avoids spending another full timeout on a1278 // plan that was stale at completion time.1279 if !matches!(feedback_receiver.has_changed(), Ok(true))1280 && let Some(immediate) = commit_coordinator.start_queued()1281 {1282 log_adaptive_acknowledgement(1283 remote,1284 lease.session_id,1285 encoder_actuation_available,1286 &commit_coordinator,1287 &immediate,1288 );1289 }1290 }1291 changed = feedback_receiver.changed(), if receiver_open => {1292 if changed.is_err() {1293 receiver_open = false;1294 if reader_finished {1295 break Ok(());1296 }1297 continue;1298 }1299 let ingress = *feedback_receiver.borrow_and_update();1300 let Some((report, next_cursor)) = ingress.report_since(feedback_cursor)? else {1301 continue;1302 };1303 // This branch now owns the complete aggregate. Advance the1304 // ingress baseline exactly once before evaluation; reports1305 // arriving during an encoder wait accumulate beyond it.1306 feedback_cursor = next_cursor;1307 let telemetry = lease.telemetry.snapshot();1308 let observed_at = Instant::now();1309 let mut candidate = controller.clone();1310 let evaluation = candidate.evaluate(&report, telemetry, observed_at)?;1311 let mut decision = evaluation.decision;1312 decision.applied =1313 commit_coordinator.bitrate_is_applied(decision.target_kbps);1314 let mut resolution_candidate = resolution_controller.clone();1315 let resolution_decision = resolution_candidate1316 .as_mut()1317 .map(|policy| policy.observe_window(&report, &evaluation, observed_at));1318 let force_keyframe =1319 adaptive_recovery_keyframe_required(&report, &decision);1320 let plan = AdaptiveCommitPlan {1321 decision,1322 telemetry,1323 resolution_decision,1324 force_keyframe,1325 };13261327 // One fixed-width decision is written for every evaluated1328 // report before actuator work is submitted. Encoder1329 // acknowledgements are correlated in bounded host1330 // diagnostics and never gate the Portal-facing cadence.1331 write_adaptive_decision(1332 send,1333 remote,1334 lease.session_id,1335 encoder_actuation_available,1336 &plan,1337 )1338 .await?;13391340 // Evaluation state advances at feedback cadence. The1341 // actuator independently retains the exact requested,1342 // committed, and latest desired targets.1343 controller = candidate;1344 resolution_controller = resolution_candidate;1345 if let Some(immediate) = commit_coordinator.submit(plan) {1346 log_adaptive_acknowledgement(1347 remote,1348 lease.session_id,1349 encoder_actuation_available,1350 &commit_coordinator,1351 &immediate,1352 );1353 }1354 }1355 _ = &mut session_changed => {1356 if !sessions.is_active(remote, lease.session_id) {1357 break Ok(());1358 }1359 }1360 _ = &mut connection_closed => {1361 break Ok(());1362 }1363 }1364 if reader_finished1365 && !receiver_open1366 && !commit_coordinator.has_pending()1367 && commit_coordinator.queued.is_none()1368 {1369 break Ok(());1370 }1371 }1372 }1373 .await;13741375 if !reader_finished {1376 reader_task.abort();1377 let _ = reader_task.await;1378 }1379 commit_coordinator.abort_and_drain().await;1380 result1381}13821383async fn write_adaptive_decision<W>(1384 send: &mut W,1385 remote: EndpointId,1386 session_id: u64,1387 encoder_actuation_available: bool,1388 plan: &AdaptiveCommitPlan,1389) -> Result<()>1390where1391 W: tokio::io::AsyncWrite + Unpin,1392{1393 let decision = plan.decision;1394 let telemetry = plan.telemetry;1395 let resolution_decision = plan.resolution_decision;1396 tokio::time::timeout(1397 FEEDBACK_WRITE_TIMEOUT,1398 write_adaptive_bitrate_decision_v1(send, &decision),1399 )1400 .await1401 .context("timed out writing adaptive bitrate decision")??;1402 debug!(1403 %remote,1404 session_id,1405 report_id = decision.report_id,1406 decision_id = decision.decision_id,1407 target_kbps = decision.target_kbps,1408 ?decision.state,1409 reasons = decision.reasons.bits(),1410 applied = decision.applied,1411 resolution_width = resolution_decision.map(|change| change.target.width),1412 resolution_height = resolution_decision.map(|change| change.target.height),1413 resolution_trigger = ?resolution_decision.map(|change| change.trigger),1414 path_rtt_micros = telemetry.selected_path_rtt_micros,1415 path_lost_packets = telemetry.selected_path_lost_packets,1416 path_congestion_events = telemetry.selected_path_congestion_events,1417 scheduler_cancellations = telemetry.scheduler_cancellations,1418 send_failures = telemetry.send_failures,1419 mode = if encoder_actuation_available { "active" } else { "shadow" },1420 "adaptive bitrate decision"1421 );1422 Ok(())1423}14241425fn log_adaptive_acknowledgement(1426 remote: EndpointId,1427 session_id: u64,1428 encoder_actuation_available: bool,1429 coordinator: &AdaptiveCommitCoordinator,1430 acknowledgement: &AdaptiveCommitAcknowledgement,1431) {1432 let decision = acknowledgement.plan.decision;1433 let resolution_decision = acknowledgement.plan.resolution_decision;1434 debug!(1435 %remote,1436 session_id,1437 report_id = decision.report_id,1438 decision_id = decision.decision_id,1439 target_kbps = decision.target_kbps,1440 bitrate_applied = acknowledgement.applied_bitrate_kbps.is_some(),1441 applied_bitrate_kbps = ?acknowledgement.applied_bitrate_kbps,1442 recovery_acknowledged = ?acknowledgement.recovery_acknowledged,1443 resolution_width = resolution_decision.map(|change| change.target.width),1444 resolution_height = resolution_decision.map(|change| change.target.height),1445 resolution_applied = ?acknowledgement.resolution_applied,1446 applied_resolution = ?acknowledgement.applied_resolution,1447 desired_bitrate_kbps = coordinator.desired_bitrate_kbps,1448 requested_bitrate_kbps = coordinator.requested_bitrate_kbps,1449 committed_bitrate_kbps = coordinator.committed_bitrate_kbps,1450 desired_resolution = ?coordinator.desired_resolution,1451 requested_resolution = ?coordinator.requested_resolution,1452 committed_resolution = ?coordinator.committed_resolution,1453 mode = if encoder_actuation_available { "active" } else { "shadow" },1454 "adaptive actuator acknowledgement"1455 );1456}14571458async fn run_adaptive_commit(1459 sessions: Arc<SessionRegistry>,1460 remote: EndpointId,1461 session_id: u64,1462 plan: AdaptiveCommitPlan,1463 bitrate_proposal: Option<AdaptiveEncoderProposal>,1464 resolution_proposal: Option<ResolutionEncoderProposal>,1465) -> AdaptiveCommitAcknowledgement {1466 let mut acknowledgement = AdaptiveCommitAcknowledgement::immediate(plan);1467 if let Some(proposal) = bitrate_proposal {1468 match commit_adaptive_encoder_proposal(&sessions, remote, session_id, &proposal).await {1469 AdaptiveEncoderCommit::GenerationEnded => {1470 acknowledgement.generation_ended = true;1471 return acknowledgement;1472 }1473 AdaptiveEncoderCommit::NotApplied(error) => {1474 warn!(1475 %remote,1476 session_id,1477 %error,1478 "adaptive encoder application failed; retaining committed bitrate"1479 );1480 }1481 AdaptiveEncoderCommit::Applied => {1482 acknowledgement.plan.decision.applied = true;1483 acknowledgement.applied_bitrate_kbps = Some(proposal.target_kbps);1484 if proposal.force_keyframe_revision.is_some() {1485 match await_adaptive_recovery_keyframe(&sessions, remote, session_id, &proposal)1486 .await1487 {1488 None => {1489 acknowledgement.generation_ended = true;1490 return acknowledgement;1491 }1492 Some(Ok(())) => acknowledgement.recovery_acknowledged = Some(true),1493 Some(Err(error)) => {1494 acknowledgement.recovery_acknowledged = Some(false);1495 warn!(1496 %remote,1497 session_id,1498 %error,1499 "adaptive bitrate applied but forced-IDR recovery was not acknowledged"1500 );1501 }1502 }1503 }1504 }1505 }1506 }15071508 if let Some(proposal) = resolution_proposal {1509 match commit_resolution_encoder_proposal(&sessions, remote, session_id, &proposal).await {1510 AdaptiveEncoderCommit::GenerationEnded => {1511 acknowledgement.generation_ended = true;1512 return acknowledgement;1513 }1514 AdaptiveEncoderCommit::NotApplied(error) => {1515 acknowledgement.resolution_applied = Some(false);1516 warn!(1517 %remote,1518 session_id,1519 target_width = proposal.target.width,1520 target_height = proposal.target.height,1521 %error,1522 "adaptive resolution application failed; retaining committed resolution"1523 );1524 }1525 AdaptiveEncoderCommit::Applied => {1526 acknowledgement.resolution_applied = Some(true);1527 acknowledgement.applied_resolution = Some(proposal.target);1528 }1529 }1530 }1531 acknowledgement1532}15331534async fn forward_media_feedback_reports<R>(1535 mut reader: R,1536 sender: tokio::sync::watch::Sender<CumulativeFeedbackIngress>,1537) -> Result<()>1538where1539 R: tokio::io::AsyncRead + Unpin,1540{1541 let mut last_report_id = None;1542 let mut next_read_at = tokio::time::Instant::now();1543 loop {1544 tokio::time::sleep_until(next_read_at).await;1545 let Some(report) = read_media_feedback_report_v1(&mut reader).await? else {1546 return Ok(());1547 };1548 ensure!(1549 last_report_id.is_none_or(|last| report.report_id > last),1550 "feedback report IDs must increase monotonically"1551 );1552 last_report_id = Some(report.report_id);1553 sender.send_modify(|ingress| ingress.observe(report));1554 if sender.is_closed() {1555 return Ok(());1556 }1557 next_read_at = tokio::time::Instant::now() + FEEDBACK_MIN_READ_INTERVAL;1558 }1559}15601561fn adaptive_bitrate_ceiling_kbps(config: &HostConfig) -> Result<u32> {1562 let configured = match config.source {1563 VideoSource::TestPattern => SHADOW_BITRATE_CEILING_KBPS,1564 VideoSource::GamescopePipewire => {1565 let gamescope = config1566 .gamescope_pipewire1567 .as_ref()1568 .context("Gamescope feedback requires gamescope_pipewire configuration")?;1569 match gamescope.rate_control {1570 VaapiRateControl::Cbr => gamescope1571 .bitrate_kbps1572 .context("adaptive bitrate shadow mode requires a Gamescope CBR bitrate")?,1573 VaapiRateControl::Cqp => SHADOW_BITRATE_CEILING_KBPS,1574 }1575 }1576 };1577 let ceiling = quantize_down(configured);1578 ensure!(1579 ceiling >= ADAPTIVE_BITRATE_FLOOR_KBPS,1580 "adaptive bitrate ceiling is below the 1000 kbps floor"1581 );1582 Ok(ceiling)1583}15841585fn adaptive_bitrate_actuation_enabled(config: &HostConfig) -> bool {1586 config.source == VideoSource::GamescopePipewire1587 && config.gamescope_pipewire.as_ref().is_some_and(|gamescope| {1588 gamescope.encoder_backend == GamescopeEncoderBackend::InProcessGstreamer1589 && gamescope.rate_control == VaapiRateControl::Cbr1590 })1591}15921593fn adaptive_recovery_keyframe_required(1594 report: &MediaFeedbackReportV1,1595 decision: &AdaptiveBitrateDecisionV1,1596) -> bool {1597 decision.state != AdaptiveBitrateStateV1::Hold1598 && report.flags.contains(MediaFeedbackFlags::RESYNC_ACTIVE)1599}160016011602enum AdaptiveEncoderCommit {1603 GenerationEnded,1604 NotApplied(anyhow::Error),1605 Applied,1606}16071608async fn await_while_session_active<F, T>(1609 sessions: &SessionRegistry,1610 remote: EndpointId,1611 session_id: u64,1612 future: F,1613) -> Option<T>1614where1615 F: Future<Output = T>,1616{1617 tokio::pin!(future);1618 loop {1619 let session_changed = sessions.session_changed.notified();1620 tokio::pin!(session_changed);1621 session_changed.as_mut().enable();1622 if !sessions.is_active(remote, session_id) {1623 return None;1624 }1625 tokio::select! {1626 result = &mut future => return sessions.is_active(remote, session_id).then_some(result),1627 _ = &mut session_changed => {}1628 }1629 }1630}16311632async fn commit_adaptive_encoder_proposal(1633 sessions: &SessionRegistry,1634 remote: EndpointId,1635 session_id: u64,1636 proposal: &AdaptiveEncoderProposal,1637) -> AdaptiveEncoderCommit {1638 let bitrate_result = await_while_session_active(1639 sessions,1640 remote,1641 session_id,1642 proposal1643 .control1644 .wait_for_bitrate_applied(proposal.bitrate_revision, ENCODER_CONTROL_COMMIT_TIMEOUT),1645 )1646 .await;1647 let Some(bitrate_result) = bitrate_result else {1648 return AdaptiveEncoderCommit::GenerationEnded;1649 };1650 let status = match bitrate_result {1651 Ok(status) => status,1652 Err(error) => return AdaptiveEncoderCommit::NotApplied(error),1653 };1654 if status.applied_bitrate_revision != Some(proposal.bitrate_revision)1655 || status.applied_bitrate_kbps != Some(proposal.target_kbps)1656 {1657 return AdaptiveEncoderCommit::NotApplied(anyhow::anyhow!(1658 "encoder bitrate commit mismatch: revision {:?}, readback {:?}, expected revision {} at {} kbps",1659 status.applied_bitrate_revision,1660 status.applied_bitrate_kbps,1661 proposal.bitrate_revision,1662 proposal.target_kbps1663 ));1664 }16651666 AdaptiveEncoderCommit::Applied1667}16681669async fn commit_resolution_encoder_proposal(1670 sessions: &SessionRegistry,1671 remote: EndpointId,1672 session_id: u64,1673 proposal: &ResolutionEncoderProposal,1674) -> AdaptiveEncoderCommit {1675 let result = await_while_session_active(1676 sessions,1677 remote,1678 session_id,1679 proposal.control.wait_for_resolution_applied(1680 proposal.revision,1681 proposal.target.width,1682 proposal.target.height,1683 ENCODER_CONTROL_COMMIT_TIMEOUT,1684 ),1685 )1686 .await;1687 let Some(result) = result else {1688 return AdaptiveEncoderCommit::GenerationEnded;1689 };1690 match result {1691 Ok(status)1692 if status.applied_resolution_revision == Some(proposal.revision)1693 && status.applied_width == Some(proposal.target.width)1694 && status.applied_height == Some(proposal.target.height) =>1695 {1696 AdaptiveEncoderCommit::Applied1697 }1698 Ok(status) => AdaptiveEncoderCommit::NotApplied(anyhow::anyhow!(1699 "encoder resolution commit mismatch: revision {:?}, dimensions {:?}x{:?}, expected revision {} at {}x{}",1700 status.applied_resolution_revision,1701 status.applied_width,1702 status.applied_height,1703 proposal.revision,1704 proposal.target.width,1705 proposal.target.height1706 )),1707 Err(error) => AdaptiveEncoderCommit::NotApplied(error),1708 }1709}17101711async fn await_adaptive_recovery_keyframe(1712 sessions: &SessionRegistry,1713 remote: EndpointId,1714 session_id: u64,1715 proposal: &AdaptiveEncoderProposal,1716) -> Option<Result<()>> {1717 let force_keyframe_revision = proposal.force_keyframe_revision?;1718 await_while_session_active(1719 sessions,1720 remote,1721 session_id,1722 proposal.control.wait_for_force_keyframe_acknowledged(1723 force_keyframe_revision,1724 ENCODER_CONTROL_COMMIT_TIMEOUT,1725 ),1726 )1727 .await1728 .map(|result| result.map(|_| ()))1729}173017311732mod tests {1733 use super::super::{endpoint, moq_test_config};1734 use super::*;17351736 fn clean_feedback(report_id: u64) -> MediaFeedbackReportV1 {1737 MediaFeedbackReportV1 {1738 report_id,1739 interval_ms: 1_000,1740 flags: MediaFeedbackFlags::NONE,1741 last_sequence: Some(report_id),1742 transport_dropped_delta: 0,1743 frontend_dropped_delta: 0,1744 decoder_dropped_delta: 0,1745 presenter_dropped_delta: 0,1746 frontend_queue_depth: 0,1747 frontend_queue_capacity: 4,1748 decode_queue_depth: 0,1749 decode_queue_capacity: 4,1750 presenter_queue_depth: 0,1751 presenter_queue_capacity: 2,1752 transport_delivery_p95_ms: Some(10),1753 decode_p95_ms: Some(3),1754 presentation_p95_ms: Some(5),1755 }1756 }17571758 fn controller_decide(1759 controller: &mut ShadowBitrateController,1760 report: &MediaFeedbackReportV1,1761 second: u64,1762 ) -> AdaptiveBitrateDecisionV1 {1763 controller1764 .decide(1765 report,1766 MediaV3TelemetrySnapshot::default(),1767 Instant::now() + Duration::from_secs(second),1768 )1769 .unwrap()1770 }17711772 fn evaluate_commit_plan(1773 controller: &mut ShadowBitrateController,1774 resolution_controller: &mut Option<MotionResolutionPolicy>,1775 report: &MediaFeedbackReportV1,1776 observed_at: Instant,1777 ) -> AdaptiveCommitPlan {1778 let telemetry = MediaV3TelemetrySnapshot::default();1779 let evaluation = controller.evaluate(report, telemetry, observed_at).unwrap();1780 let resolution_decision = resolution_controller1781 .as_mut()1782 .map(|policy| policy.observe_window(report, &evaluation, observed_at));1783 AdaptiveCommitPlan {1784 decision: evaluation.decision,1785 telemetry,1786 resolution_decision,1787 force_keyframe: adaptive_recovery_keyframe_required(report, &evaluation.decision),1788 }1789 }17901791 fn native_dimensions() -> VideoDimensions {1792 VideoDimensions {1793 width: 1_280,1794 height: 800,1795 }1796 }17971798 1799 fn motion_resolution_tier_is_exact_even_and_same_aspect() {1800 let native = native_dimensions();1801 let policy = MotionResolutionPolicy::new(native).unwrap();1802 assert_eq!(policy.target(), native);1803 assert_eq!(1804 policy.reduced,1805 VideoDimensions {1806 width: 960,1807 height: 600,1808 }1809 );1810 assert_eq!(1811 u32::from(native.width) * u32::from(policy.reduced.height),1812 u32::from(native.height) * u32::from(policy.reduced.width)1813 );1814 assert!(policy.reduced.width.is_multiple_of(2));1815 assert!(policy.reduced.height.is_multiple_of(2));18161817 assert!(1818 MotionResolutionPolicy::new(VideoDimensions {1819 width: 1_278,1820 height: 800,1821 })1822 .is_err()1823 );1824 assert!(1825 MotionResolutionPolicy::new(VideoDimensions {1826 width: 80,1827 height: 64,1828 })1829 .is_err()1830 );1831 assert!(motion_resolution_policy(1_920, 1_080).unwrap().is_some());1832 assert!(motion_resolution_policy(1_366, 768).unwrap().is_none());1833 }18341835 1836 fn feedback_resolution_policy_uses_encoder_generation_dimensions() {1837 let full_hd = crate::source::EncoderControlTestHarness::with_dimensions(1_920, 1_080);1838 let policy = motion_resolution_policy_for_encoder(&full_hd.control)1839 .unwrap()1840 .unwrap();1841 assert_eq!(1842 policy.native,1843 VideoDimensions {1844 width: 1_920,1845 height: 1_080,1846 }1847 );18481849 let no_exact_tier = crate::source::EncoderControlTestHarness::with_dimensions(1_366, 768);1850 assert!(1851 motion_resolution_policy_for_encoder(&no_exact_tier.control)1852 .unwrap()1853 .is_none()1854 );1855 }18561857 1858 fn motion_downshifts_immediately_and_recovers_only_after_settle_and_clean_windows() {1859 let now = Instant::now();1860 let mut policy = MotionResolutionPolicy::new(native_dimensions()).unwrap();1861 let down = policy.observe_motion(true, now);1862 assert_eq!(down.tier, ResolutionTier::Reduced);1863 assert_eq!(down.trigger, ResolutionTrigger::Motion);1864 assert!(down.changed);18651866 let still = policy.observe_motion(false, now + Duration::from_secs(1));1867 assert!(!still.changed);1868 for second in [2, 3] {1869 assert!(1870 !policy1871 .observe_feedback(1872 FeedbackSeverity::Clean,1873 true,1874 now + Duration::from_secs(second),1875 )1876 .changed1877 );1878 }1879 let recovered =1880 policy.observe_feedback(FeedbackSeverity::Clean, true, now + Duration::from_secs(4));1881 assert_eq!(recovered.tier, ResolutionTier::Native);1882 assert_eq!(recovered.target, native_dimensions());1883 assert_eq!(recovered.trigger, ResolutionTrigger::Recovery);1884 assert!(recovered.changed);1885 }18861887 1888 fn repeated_still_observations_do_not_postpone_resolution_recovery() {1889 let now = Instant::now();1890 let mut policy = MotionResolutionPolicy::new(native_dimensions()).unwrap();1891 policy.observe_motion(true, now);1892 policy.observe_motion(false, now + Duration::from_secs(1));1893 policy.observe_feedback(FeedbackSeverity::Clean, true, now + Duration::from_secs(2));1894 policy.observe_motion(false, now + Duration::from_millis(2_500));1895 policy.observe_feedback(FeedbackSeverity::Clean, true, now + Duration::from_secs(3));1896 let recovered =1897 policy.observe_feedback(FeedbackSeverity::Clean, true, now + Duration::from_secs(4));1898 assert!(recovered.changed);1899 assert_eq!(recovered.tier, ResolutionTier::Native);1900 }19011902 1903 fn resolution_pressure_requires_fresh_hysteresis_but_severe_is_immediate() {1904 let now = Instant::now();1905 let mut moderate = MotionResolutionPolicy::new(native_dimensions()).unwrap();1906 moderate.observe_motion(false, now);1907 assert!(1908 !moderate1909 .observe_feedback(1910 FeedbackSeverity::Moderate,1911 true,1912 now + Duration::from_secs(1),1913 )1914 .changed1915 );1916 assert!(1917 !moderate1918 .observe_feedback(1919 FeedbackSeverity::Moderate,1920 false,1921 now + Duration::from_secs(2),1922 )1923 .changed1924 );1925 assert!(1926 !moderate1927 .observe_feedback(1928 FeedbackSeverity::Moderate,1929 true,1930 now + Duration::from_secs(3),1931 )1932 .changed1933 );1934 let down = moderate.observe_feedback(1935 FeedbackSeverity::Moderate,1936 true,1937 now + Duration::from_secs(4),1938 );1939 assert!(down.changed);1940 assert_eq!(down.trigger, ResolutionTrigger::Pressure);19411942 let mut severe = MotionResolutionPolicy::new(native_dimensions()).unwrap();1943 let immediate =1944 severe.observe_feedback(FeedbackSeverity::Severe, true, now + Duration::from_secs(1));1945 assert!(immediate.changed);1946 assert_eq!(immediate.tier, ResolutionTier::Reduced);1947 }19481949 1950 fn stale_feedback_cannot_restore_native_resolution() {1951 let now = Instant::now();1952 let mut policy = MotionResolutionPolicy::new(native_dimensions()).unwrap();1953 policy.observe_motion(false, now);1954 policy.observe_feedback(FeedbackSeverity::Severe, true, now + Duration::from_secs(1));1955 for second in [2, 3, 4, 5] {1956 let decision = policy.observe_feedback(1957 FeedbackSeverity::Clean,1958 false,1959 now + Duration::from_secs(second),1960 );1961 assert_eq!(decision.tier, ResolutionTier::Reduced);1962 assert!(!decision.changed);1963 }1964 assert!(policy.pressure_active);19651966 for second in [6, 7] {1967 assert!(1968 !policy1969 .observe_feedback(1970 FeedbackSeverity::Clean,1971 true,1972 now + Duration::from_secs(second),1973 )1974 .changed1975 );1976 }1977 let recovered =1978 policy.observe_feedback(FeedbackSeverity::Clean, true, now + Duration::from_secs(8));1979 assert!(recovered.changed);1980 assert_eq!(recovered.tier, ResolutionTier::Native);1981 }19821983 1984 fn motion_keeps_resolution_reduced_after_pressure_clears() {1985 let now = Instant::now();1986 let mut policy = MotionResolutionPolicy::new(native_dimensions()).unwrap();1987 policy.observe_motion(true, now);1988 policy.observe_feedback(FeedbackSeverity::Severe, true, now + Duration::from_secs(1));1989 for second in 2..=5 {1990 let decision = policy.observe_feedback(1991 FeedbackSeverity::Clean,1992 true,1993 now + Duration::from_secs(second),1994 );1995 assert_eq!(decision.tier, ResolutionTier::Reduced);1996 assert!(!decision.changed);1997 }1998 assert!(!policy.pressure_active);1999 assert!(policy.motion_active);2000 }20012002 2003 fn damage_driven_motion_proxy_seeds_then_downshifts_and_stale_blocks_upscale() {2004 let now = Instant::now();2005 let telemetry = MediaV3TelemetrySnapshot::default();2006 let mut bitrate = ShadowBitrateController::new(12_000, telemetry);2007 let mut resolution = MotionResolutionPolicy::new(native_dimensions()).unwrap();20082009 let mut first = clean_feedback(1);2010 first.last_sequence = Some(100);2011 let first_evaluation = bitrate.evaluate(&first, telemetry, now).unwrap();2012 assert!(!first_evaluation.fresh);2013 let seeded = resolution.observe_window(&first, &first_evaluation, now);2014 assert_eq!(seeded.tier, ResolutionTier::Native);2015 assert!(!seeded.changed);20162017 let mut second = clean_feedback(2);2018 second.last_sequence = Some(110);2019 let second_evaluation = bitrate2020 .evaluate(&second, telemetry, now + Duration::from_secs(1))2021 .unwrap();2022 assert!(second_evaluation.fresh);2023 let down =2024 resolution.observe_window(&second, &second_evaluation, now + Duration::from_secs(1));2025 assert_eq!(down.trigger, ResolutionTrigger::Motion);2026 assert_eq!(down.tier, ResolutionTier::Reduced);2027 assert!(down.changed);20282029 let mut stale = clean_feedback(3);2030 stale.interval_ms = 5_000;2031 stale.last_sequence = Some(110);2032 let stale_evaluation = bitrate2033 .evaluate(&stale, telemetry, now + Duration::from_secs(5))2034 .unwrap();2035 assert!(!stale_evaluation.fresh);2036 let held =2037 resolution.observe_window(&stale, &stale_evaluation, now + Duration::from_secs(5));2038 assert_eq!(held.tier, ResolutionTier::Reduced);2039 assert!(!held.changed);2040 }20412042 2043 fn damage_driven_no_progress_cannot_restore_native_resolution() {2044 let now = Instant::now();2045 let telemetry = MediaV3TelemetrySnapshot::default();2046 let mut bitrate = ShadowBitrateController::new(12_000, telemetry);2047 let mut resolution = MotionResolutionPolicy::new(native_dimensions()).unwrap();20482049 let mut seed = clean_feedback(1);2050 seed.last_sequence = Some(100);2051 let seed_evaluation = bitrate.evaluate(&seed, telemetry, now).unwrap();2052 resolution.observe_window(&seed, &seed_evaluation, now);20532054 let mut pressure = clean_feedback(2);2055 pressure.last_sequence = Some(110);2056 pressure.flags = MediaFeedbackFlags::RESYNC_ACTIVE;2057 let pressure_evaluation = bitrate2058 .evaluate(&pressure, telemetry, now + Duration::from_secs(1))2059 .unwrap();2060 let downshift = resolution.observe_window(2061 &pressure,2062 &pressure_evaluation,2063 now + Duration::from_secs(1),2064 );2065 assert_eq!(downshift.tier, ResolutionTier::Reduced);2066 assert!(downshift.changed);20672068 for second in 2..=4 {2069 let mut no_progress = clean_feedback(second + 1);2070 no_progress.last_sequence = Some(110);2071 let evaluation = bitrate2072 .evaluate(&no_progress, telemetry, now + Duration::from_secs(second))2073 .unwrap();2074 assert_eq!(evaluation.pressure_severity, FeedbackSeverity::Clean);2075 assert!(!evaluation.clean_recovery_evidence);2076 let decision = resolution.observe_window(2077 &no_progress,2078 &evaluation,2079 now + Duration::from_secs(second),2080 );2081 assert_eq!(decision.tier, ResolutionTier::Reduced);2082 assert_eq!(decision.trigger, ResolutionTrigger::Pressure);2083 assert!(!decision.changed);2084 assert_eq!(resolution.clean_windows, 0);2085 }20862087 for (report_id, second, sequence) in [(6, 5, 111), (7, 6, 112), (8, 7, 113)] {2088 let mut low_motion = clean_feedback(report_id);2089 low_motion.last_sequence = Some(sequence);2090 let evaluation = bitrate2091 .evaluate(&low_motion, telemetry, now + Duration::from_secs(second))2092 .unwrap();2093 assert!(evaluation.clean_recovery_evidence);2094 let decision = resolution.observe_window(2095 &low_motion,2096 &evaluation,2097 now + Duration::from_secs(second),2098 );2099 if report_id < 8 {2100 assert_eq!(decision.tier, ResolutionTier::Reduced);2101 assert!(!decision.changed);2102 } else {2103 assert_eq!(decision.tier, ResolutionTier::Native);2104 assert_eq!(decision.trigger, ResolutionTrigger::Recovery);2105 assert!(decision.changed);2106 }2107 }2108 }21092110 2111 fn shadow_controller_immediately_decreases_for_severe_feedback_and_stays_bounded() {2112 let mut controller =2113 ShadowBitrateController::new(12_137, MediaV3TelemetrySnapshot::default());2114 assert_eq!(controller.ceiling_kbps, 12_000);2115 let first = controller_decide(&mut controller, &clean_feedback(1), 0);2116 assert_eq!(first.state, AdaptiveBitrateStateV1::Hold);2117 assert!(2118 first2119 .reasons2120 .contains(AdaptiveBitrateReasonFlagsV1::FEEDBACK_STALE)2121 );2122 for report_id in 2..=21 {2123 let mut report = clean_feedback(report_id);2124 report.flags = MediaFeedbackFlags::RESYNC_ACTIVE;2125 let decision = controller_decide(&mut controller, &report, report_id - 1);2126 assert!(!decision.applied);2127 assert!(decision.target_kbps >= ADAPTIVE_BITRATE_FLOOR_KBPS);2128 assert!(decision.target_kbps <= 12_000);2129 assert_eq!(decision.target_kbps % ADAPTIVE_BITRATE_QUANTUM_KBPS, 0);2130 assert!(2131 decision2132 .reasons2133 .contains(AdaptiveBitrateReasonFlagsV1::RECEIVER_QUEUE)2134 );2135 }2136 assert_eq!(controller.target_kbps, ADAPTIVE_BITRATE_FLOOR_KBPS);2137 }21382139 2140 fn shadow_controller_requires_two_moderate_windows() {2141 let mut controller =2142 ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());2143 controller_decide(&mut controller, &clean_feedback(1), 0);2144 let mut first = clean_feedback(2);2145 first.frontend_queue_depth = 2;2146 assert_eq!(2147 controller_decide(&mut controller, &first, 1).state,2148 AdaptiveBitrateStateV1::Hold2149 );2150 let mut second = clean_feedback(3);2151 second.frontend_queue_depth = 2;2152 let decision = controller_decide(&mut controller, &second, 2);2153 assert_eq!(decision.state, AdaptiveBitrateStateV1::Decrease);2154 assert_eq!(decision.target_kbps, 9_500);2155 }21562157 2158 fn shadow_controller_increases_only_after_clean_cooldown() {2159 let mut controller =2160 ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());2161 controller_decide(&mut controller, &clean_feedback(1), 0);2162 let mut severe = clean_feedback(2);2163 severe.decoder_dropped_delta = 1;2164 assert_eq!(2165 controller_decide(&mut controller, &severe, 1).target_kbps,2166 9_0002167 );2168 for report_id in 3..=11 {2169 let decision =2170 controller_decide(&mut controller, &clean_feedback(report_id), report_id - 1);2171 assert_eq!(decision.state, AdaptiveBitrateStateV1::Hold);2172 assert_eq!(decision.target_kbps, 9_000);2173 }2174 let decision = controller_decide(&mut controller, &clean_feedback(12), 11);2175 assert_eq!(decision.state, AdaptiveBitrateStateV1::Increase);2176 assert_eq!(decision.target_kbps, 9_250);2177 assert!(2178 decision2179 .reasons2180 .contains(AdaptiveBitrateReasonFlagsV1::CLEAN_RECOVERY)2181 );2182 }21832184 2185 fn stale_or_non_monotonic_feedback_cannot_drive_an_increase() {2186 let mut controller =2187 ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());2188 controller_decide(&mut controller, &clean_feedback(1), 0);2189 let mut severe = clean_feedback(2);2190 severe.flags = MediaFeedbackFlags::RESYNC_ACTIVE;2191 controller_decide(&mut controller, &severe, 1);2192 let stale = controller_decide(&mut controller, &clean_feedback(3), 4);2193 assert_eq!(stale.state, AdaptiveBitrateStateV1::Hold);2194 assert_eq!(stale.target_kbps, 9_000);2195 assert!(2196 stale2197 .reasons2198 .contains(AdaptiveBitrateReasonFlagsV1::FEEDBACK_STALE)2199 );2200 assert!(2201 controller2202 .decide(2203 &clean_feedback(2),2204 MediaV3TelemetrySnapshot::default(),2205 Instant::now() + Duration::from_secs(4),2206 )2207 .is_err()2208 );22092210 let now = Instant::now();2211 let mut rapid = ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());2212 rapid2213 .decide(&clean_feedback(1), MediaV3TelemetrySnapshot::default(), now)2214 .unwrap();2215 let too_fast = rapid2216 .decide(2217 &clean_feedback(2),2218 MediaV3TelemetrySnapshot::default(),2219 now + Duration::from_millis(500),2220 )2221 .unwrap();2222 assert_eq!(too_fast.state, AdaptiveBitrateStateV1::Hold);2223 assert!(2224 too_fast2225 .reasons2226 .contains(AdaptiveBitrateReasonFlagsV1::FEEDBACK_STALE)2227 );2228 }22292230 2231 fn full_decode_queue_is_severe_even_at_capacity_two() {2232 let mut controller =2233 ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());2234 controller_decide(&mut controller, &clean_feedback(1), 0);2235 let mut report = clean_feedback(2);2236 report.decode_queue_capacity = 2;2237 report.decode_queue_depth = 2;2238 let decision = controller_decide(&mut controller, &report, 1);2239 assert_eq!(decision.state, AdaptiveBitrateStateV1::Decrease);2240 assert!(2241 decision2242 .reasons2243 .contains(AdaptiveBitrateReasonFlagsV1::DECODE_BACKLOG)2244 );2245 }22462247 2248 fn decision_id_exhaustion_fails_instead_of_repeating() {2249 let mut controller =2250 ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());2251 controller.next_decision_id = u64::MAX;2252 assert!(2253 controller2254 .decide(2255 &clean_feedback(1),2256 MediaV3TelemetrySnapshot::default(),2257 Instant::now(),2258 )2259 .is_err()2260 );2261 assert_eq!(controller.next_decision_id, u64::MAX);2262 }22632264 2265 fn stale_feedback_applies_trusted_loss_and_send_failure_pressure_once() {2266 for telemetry in [2267 MediaV3TelemetrySnapshot {2268 selected_path_lost_packets: 1,2269 ..MediaV3TelemetrySnapshot::default()2270 },2271 MediaV3TelemetrySnapshot {2272 send_failures: 1,2273 ..MediaV3TelemetrySnapshot::default()2274 },2275 ] {2276 let now = Instant::now();2277 let mut controller =2278 ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());2279 let decision = controller2280 .decide(&clean_feedback(1), telemetry, now)2281 .unwrap();2282 assert_eq!(decision.state, AdaptiveBitrateStateV1::Decrease);2283 assert_eq!(decision.target_kbps, 9_000);2284 assert!(2285 decision2286 .reasons2287 .contains(AdaptiveBitrateReasonFlagsV1::FEEDBACK_STALE)2288 );2289 assert!(2290 decision2291 .reasons2292 .contains(AdaptiveBitrateReasonFlagsV1::LOSS_OR_CANCELLATION)2293 );22942295 let repeated = controller2296 .decide(2297 &clean_feedback(2),2298 telemetry,2299 now + Duration::from_millis(500),2300 )2301 .unwrap();2302 assert_eq!(repeated.state, AdaptiveBitrateStateV1::Hold);2303 assert_eq!(repeated.target_kbps, 9_000);2304 }2305 }23062307 2308 fn stale_moderate_cancellation_is_retained_until_hysteresis_reduces() {2309 let now = Instant::now();2310 let telemetry = MediaV3TelemetrySnapshot {2311 scheduler_cancellations: 1,2312 ..MediaV3TelemetrySnapshot::default()2313 };2314 let mut controller =2315 ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());2316 let first = controller2317 .decide(&clean_feedback(1), telemetry, now)2318 .unwrap();2319 assert_eq!(first.state, AdaptiveBitrateStateV1::Hold);2320 assert_eq!(controller.previous_telemetry.scheduler_cancellations, 0);23212322 let second = controller2323 .decide(2324 &clean_feedback(2),2325 telemetry,2326 now + Duration::from_millis(500),2327 )2328 .unwrap();2329 assert_eq!(second.state, AdaptiveBitrateStateV1::Decrease);2330 assert_eq!(second.target_kbps, 9_500);2331 assert_eq!(controller.previous_telemetry.scheduler_cancellations, 1);2332 }23332334 2335 fn clean_recovery_below_five_mbps_advances_by_one_quantum() {2336 let mut controller =2337 ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());2338 controller.target_kbps = 4_000;2339 controller_decide(&mut controller, &clean_feedback(1), 0);2340 controller_decide(&mut controller, &clean_feedback(2), 1);2341 for report_id in 3..=11 {2342 let decision =2343 controller_decide(&mut controller, &clean_feedback(report_id), report_id - 1);2344 assert_eq!(decision.state, AdaptiveBitrateStateV1::Hold);2345 }2346 let decision = controller_decide(&mut controller, &clean_feedback(12), 11);2347 assert_eq!(decision.state, AdaptiveBitrateStateV1::Increase);2348 assert_eq!(decision.target_kbps, 4_250);2349 }23502351 2352 fn recovery_requires_latency_samples_and_sequence_progress_but_pressure_still_reduces() {2353 let mut controller =2354 ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());2355 controller.target_kbps = 9_000;2356 controller_decide(&mut controller, &clean_feedback(1), 0);2357 controller_decide(&mut controller, &clean_feedback(2), 1);23582359 for report_id in 3..=14 {2360 let mut report = clean_feedback(report_id);2361 report.last_sequence = Some(2);2362 let decision = controller_decide(&mut controller, &report, report_id - 1);2363 assert_eq!(decision.state, AdaptiveBitrateStateV1::Hold);2364 assert_eq!(decision.target_kbps, 9_000);2365 }2366 assert_eq!(controller.clean_windows, 0);23672368 for report_id in 15..=26 {2369 let mut report = clean_feedback(report_id);2370 report.decode_p95_ms = None;2371 let decision = controller_decide(&mut controller, &report, report_id - 1);2372 assert_eq!(decision.state, AdaptiveBitrateStateV1::Hold);2373 assert_eq!(decision.target_kbps, 9_000);2374 }2375 assert_eq!(controller.clean_windows, 0);23762377 let mut pressure = clean_feedback(27);2378 pressure.decode_p95_ms = None;2379 pressure.decode_queue_capacity = 2;2380 pressure.decode_queue_depth = 2;2381 let decision = controller_decide(&mut controller, &pressure, 26);2382 assert_eq!(decision.state, AdaptiveBitrateStateV1::Decrease);2383 assert_eq!(decision.target_kbps, 6_750);2384 assert!(2385 decision2386 .reasons2387 .contains(AdaptiveBitrateReasonFlagsV1::DECODE_BACKLOG)2388 );2389 }23902391 fn gamescope_feedback_test_config(rate_control: VaapiRateControl) -> HostConfig {2392 let (bitrate_kbps, quantizer) = match rate_control {2393 VaapiRateControl::Cbr => (Some(12_123), None),2394 VaapiRateControl::Cqp => (None, Some(24)),2395 };2396 let mut config = moq_test_config();2397 config.source = VideoSource::GamescopePipewire;2398 config.width = None;2399 config.height = None;2400 config.gamescope_pipewire = Some(crate::config::GamescopePipewireConfig {2401 node_name: "gamescope".into(),2402 media_class: "Video/Source".into(),2403 match_properties: std::collections::BTreeMap::new(),2404 xwayland_display: None,2405 pw_dump_path: "/usr/bin/pw-dump".into(),2406 gst_launch_path: "/usr/bin/gst-launch-1.0".into(),2407 gst_inspect_path: "/usr/bin/gst-inspect-1.0".into(),2408 encoder_backend: GamescopeEncoderBackend::ExternalGstLaunch,2409 vaapi_encoder: "vah264enc".into(),2410 vaapi_render_node: "/dev/dri/renderD128".into(),2411 rate_control,2412 bitrate_kbps,2413 quantizer,2414 });2415 config.validate().unwrap();2416 config2417 }24182419 2420 fn external_cqp_uses_a_fixed_shadow_bitrate_ceiling() {2421 let config = gamescope_feedback_test_config(VaapiRateControl::Cqp);24222423 assert_eq!(2424 adaptive_bitrate_ceiling_kbps(&config).unwrap(),2425 SHADOW_BITRATE_CEILING_KBPS2426 );2427 assert!(!adaptive_bitrate_actuation_enabled(&config));2428 }24292430 2431 fn gamescope_cbr_keeps_its_quantized_configured_ceiling() {2432 let config = gamescope_feedback_test_config(VaapiRateControl::Cbr);24332434 assert_eq!(adaptive_bitrate_ceiling_kbps(&config).unwrap(), 12_000);2435 }24362437 2438 fn malformed_cbr_without_a_bitrate_still_fails_closed() {2439 let mut config = gamescope_feedback_test_config(VaapiRateControl::Cbr);2440 config.gamescope_pipewire.as_mut().unwrap().bitrate_kbps = None;24412442 assert_eq!(2443 adaptive_bitrate_ceiling_kbps(&config)2444 .unwrap_err()2445 .to_string(),2446 "adaptive bitrate shadow mode requires a Gamescope CBR bitrate"2447 );2448 }24492450 2451 async fn adaptive_encoder_commit_requires_exact_readback_and_tracks_recovery_separately() {2452 let sessions = Arc::new(SessionRegistry::default());2453 let remote = endpoint(1);2454 let media = sessions2455 .claim(remote, [4; 16], InvitationGrants::VIEW)2456 .unwrap();2457 let harness = crate::source::EncoderControlTestHarness::new();2458 sessions2459 .install_encoder_control(remote, media.session_id, Some(harness.control.clone()))2460 .unwrap();24612462 let proposal = sessions2463 .propose_adaptive_encoder_update(remote, media.session_id, 8_000, true)2464 .unwrap()2465 .unwrap();2466 harness.status.send_modify(|status| {2467 status.applied_bitrate_revision = Some(proposal.bitrate_revision);2468 status.applied_bitrate_kbps = Some(8_000);2469 status.requested_force_keyframe_revision = proposal.force_keyframe_revision;2470 status.acknowledged_force_keyframe_revision = proposal.force_keyframe_revision;2471 });2472 assert!(matches!(2473 commit_adaptive_encoder_proposal(&sessions, remote, media.session_id, &proposal).await,2474 AdaptiveEncoderCommit::Applied2475 ));2476 assert!(matches!(2477 await_adaptive_recovery_keyframe(&sessions, remote, media.session_id, &proposal).await,2478 Some(Ok(()))2479 ));24802481 let mismatch = sessions2482 .propose_adaptive_encoder_update(remote, media.session_id, 7_000, false)2483 .unwrap()2484 .unwrap();2485 harness.status.send_modify(|status| {2486 status.applied_bitrate_revision = Some(mismatch.bitrate_revision);2487 status.applied_bitrate_kbps = Some(7_250);2488 });2489 assert!(matches!(2490 commit_adaptive_encoder_proposal(&sessions, remote, media.session_id, &mismatch).await,2491 AdaptiveEncoderCommit::NotApplied(_)2492 ));24932494 let stale_generation = sessions2495 .propose_adaptive_encoder_update(remote, media.session_id, 6_000, false)2496 .unwrap()2497 .unwrap();2498 let session_id = media.session_id;2499 drop(media);2500 assert!(matches!(2501 commit_adaptive_encoder_proposal(&sessions, remote, session_id, &stale_generation)2502 .await,2503 AdaptiveEncoderCommit::GenerationEnded2504 ));2505 }25062507 2508 fn adaptive_recovery_idr_is_only_requested_for_a_bitrate_change_during_resync() {2509 let mut report = clean_feedback(1);2510 let mut decision = AdaptiveBitrateDecisionV1 {2511 decision_id: 1,2512 report_id: 1,2513 target_kbps: 8_000,2514 floor_kbps: 1_000,2515 ceiling_kbps: 12_000,2516 state: AdaptiveBitrateStateV1::Decrease,2517 reasons: AdaptiveBitrateReasonFlagsV1::RECEIVER_QUEUE,2518 applied: false,2519 };2520 assert!(!adaptive_recovery_keyframe_required(&report, &decision));2521 report.flags = MediaFeedbackFlags::RESYNC_ACTIVE;2522 assert!(adaptive_recovery_keyframe_required(&report, &decision));2523 decision.state = AdaptiveBitrateStateV1::Hold;2524 assert!(!adaptive_recovery_keyframe_required(&report, &decision));2525 }25262527 2528 async fn stalled_adaptive_apply_coalesces_feedback_without_losing_pressure() {2529 use tokio::io::AsyncWriteExt as _;25302531 let (sender, mut receiver) =2532 tokio::sync::watch::channel(CumulativeFeedbackIngress::default());2533 let (mut writer, reader) = tokio::io::duplex(512);2534 let reader_task = tokio::spawn(forward_media_feedback_reports(reader, sender.clone()));2535 let mut cursor = FeedbackIngressCursor::default();25362537 sigil_protocol::write_media_feedback_report_v1(&mut writer, &clean_feedback(1))2538 .await2539 .unwrap();2540 receiver.changed().await.unwrap();2541 let (first, next_cursor) = receiver2542 .borrow_and_update()2543 .report_since(cursor)2544 .unwrap()2545 .unwrap();2546 assert_eq!(first.report_id, 1);2547 cursor = next_cursor;25482549 // The consumer is now notionally stalled on an encoder commit. Two2550 // reports overwrite one watch notification, but their interval,2551 // deltas, and resync pressure must remain cumulative.2552 let mut pressured = clean_feedback(2);2553 pressured.flags = MediaFeedbackFlags::RESYNC_ACTIVE;2554 pressured.transport_dropped_delta = 2;2555 pressured.decoder_dropped_delta = 1;2556 pressured.decode_queue_depth = 3;2557 pressured.transport_delivery_p95_ms = Some(180);2558 sigil_protocol::write_media_feedback_report_v1(&mut writer, &pressured)2559 .await2560 .unwrap();25612562 let mut latest = clean_feedback(3);2563 latest.frontend_dropped_delta = 4;2564 latest.presenter_dropped_delta = 3;2565 latest.frontend_queue_depth = 3;2566 latest.decode_queue_depth = 1;2567 latest.transport_delivery_p95_ms = Some(40);2568 latest.decode_p95_ms = Some(9);2569 sigil_protocol::write_media_feedback_report_v1(&mut writer, &latest)2570 .await2571 .unwrap();2572 writer.shutdown().await.unwrap();2573 reader_task.await.unwrap().unwrap();25742575 receiver.changed().await.unwrap();2576 let (aggregate, next_cursor) = receiver2577 .borrow_and_update()2578 .report_since(cursor)2579 .unwrap()2580 .unwrap();2581 assert_eq!(aggregate.report_id, 3);2582 assert_eq!(aggregate.interval_ms, 2_000);2583 assert_eq!(aggregate.transport_dropped_delta, 2);2584 assert_eq!(aggregate.frontend_dropped_delta, 4);2585 assert_eq!(aggregate.decoder_dropped_delta, 1);2586 assert_eq!(aggregate.presenter_dropped_delta, 3);2587 assert!(aggregate.flags.contains(MediaFeedbackFlags::RESYNC_ACTIVE));2588 assert_eq!(aggregate.last_sequence, latest.last_sequence);2589 assert_eq!(aggregate.frontend_queue_depth, latest.frontend_queue_depth);2590 assert_eq!(aggregate.decode_queue_depth, latest.decode_queue_depth);2591 assert_eq!(2592 aggregate.transport_delivery_p95_ms,2593 latest.transport_delivery_p95_ms2594 );2595 assert_eq!(aggregate.decode_p95_ms, latest.decode_p95_ms);2596 cursor = next_cursor;2597 assert!(2598 receiver.borrow().report_since(cursor).unwrap().is_none(),2599 "one cumulative ingress generation must be processed only once"2600 );26012602 for report_id in 4..=9 {2603 sender.send_modify(|ingress| ingress.observe(clean_feedback(report_id)));2604 }2605 receiver.changed().await.unwrap();2606 let (clamped, _) = receiver2607 .borrow_and_update()2608 .report_since(cursor)2609 .unwrap()2610 .unwrap();2611 assert_eq!(clamped.interval_ms, 5_000);2612 let mut controller =2613 ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());2614 let decision = controller2615 .decide(2616 &clamped,2617 MediaV3TelemetrySnapshot::default(),2618 Instant::now(),2619 )2620 .unwrap();2621 assert!(2622 decision2623 .reasons2624 .contains(AdaptiveBitrateReasonFlagsV1::FEEDBACK_STALE),2625 "a clamped aggregate must be stale rather than disguised as one fresh report"2626 );2627 }26282629 2630 async fn queued_adaptive_plan_keeps_latest_ids_and_recovery_barrier() {2631 let sessions = Arc::new(SessionRegistry::default());2632 let remote = endpoint(1);2633 let media = sessions2634 .claim(remote, [3; 16], InvitationGrants::VIEW)2635 .unwrap();2636 let harness = crate::source::EncoderControlTestHarness::new();2637 sessions2638 .install_encoder_control(remote, media.session_id, Some(harness.control.clone()))2639 .unwrap();2640 let mut coordinator = AdaptiveCommitCoordinator::new(2641 Arc::clone(&sessions),2642 remote,2643 media.session_id,2644 true,2645 12_000,2646 Some(native_dimensions()),2647 );2648 let mut controller =2649 ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());2650 let mut resolution_controller = None;2651 let observed_at = Instant::now();26522653 let seed = evaluate_commit_plan(2654 &mut controller,2655 &mut resolution_controller,2656 &clean_feedback(1),2657 observed_at,2658 );2659 assert!(coordinator.submit(seed).is_some());26602661 let mut first_pressure = clean_feedback(2);2662 first_pressure.decoder_dropped_delta = 1;2663 let first = evaluate_commit_plan(2664 &mut controller,2665 &mut resolution_controller,2666 &first_pressure,2667 observed_at + Duration::from_secs(1),2668 );2669 assert!(!first.force_keyframe);2670 assert!(coordinator.submit(first).is_none());26712672 let mut recovery_pressure = clean_feedback(3);2673 recovery_pressure.decoder_dropped_delta = 1;2674 recovery_pressure.flags = MediaFeedbackFlags::RESYNC_ACTIVE;2675 let recovery = evaluate_commit_plan(2676 &mut controller,2677 &mut resolution_controller,2678 &recovery_pressure,2679 observed_at + Duration::from_secs(2),2680 );2681 assert!(recovery.force_keyframe);2682 assert!(coordinator.submit(recovery).is_none());26832684 let latest = evaluate_commit_plan(2685 &mut controller,2686 &mut resolution_controller,2687 &clean_feedback(4),2688 observed_at + Duration::from_secs(3),2689 );2690 assert!(!latest.force_keyframe);2691 assert!(coordinator.submit(latest).is_none());2692 let retained = coordinator.queued.expect("latest plan must be retained");2693 assert_eq!(retained.decision.decision_id, latest.decision.decision_id);2694 assert_eq!(retained.decision.report_id, latest.decision.report_id);2695 assert!(2696 retained.force_keyframe,2697 "a coalesced recovery barrier must survive a newer hold decision"2698 );26992700 coordinator.abort_and_drain().await;2701 drop(media);2702 }27032704 2705 async fn timed_out_actuator_requests_are_overwritten_when_policy_returns_to_committed_state() {2706 let sessions = Arc::new(SessionRegistry::default());2707 let remote = endpoint(1);2708 let media = sessions2709 .claim(remote, [8; 16], InvitationGrants::VIEW)2710 .unwrap();2711 let harness = crate::source::EncoderControlTestHarness::new();2712 sessions2713 .install_encoder_control(remote, media.session_id, Some(harness.control.clone()))2714 .unwrap();2715 let native = native_dimensions();2716 let reduced = native.three_quarter().unwrap();2717 let mut coordinator = AdaptiveCommitCoordinator::new(2718 Arc::clone(&sessions),2719 remote,2720 media.session_id,2721 true,2722 12_000,2723 Some(native),2724 );2725 let timed_out = AdaptiveCommitPlan {2726 decision: AdaptiveBitrateDecisionV1 {2727 decision_id: 1,2728 report_id: 1,2729 target_kbps: 9_000,2730 floor_kbps: 1_000,2731 ceiling_kbps: 12_000,2732 state: AdaptiveBitrateStateV1::Decrease,2733 reasons: AdaptiveBitrateReasonFlagsV1::RECEIVER_QUEUE,2734 applied: false,2735 },2736 telemetry: MediaV3TelemetrySnapshot::default(),2737 resolution_decision: Some(MotionResolutionDecision {2738 tier: ResolutionTier::Reduced,2739 target: reduced,2740 trigger: ResolutionTrigger::Pressure,2741 changed: true,2742 }),2743 force_keyframe: false,2744 };2745 assert!(coordinator.submit(timed_out).is_none());2746 let stale_bitrate = harness2747 .requested_bitrate()2748 .expect("the first bitrate request must reach the actuator");2749 let stale_resolution = harness2750 .requested_resolution()2751 .expect("the first resolution request must reach the actuator");2752 assert_eq!(stale_bitrate.1, 9_000);2753 assert_eq!(2754 (stale_resolution.1, stale_resolution.2),2755 (reduced.width, reduced.height)2756 );27572758 // Feedback evaluation recovers all the way to the previously2759 // committed state while the old request is still awaiting readback.2760 let recovered = AdaptiveCommitPlan {2761 decision: AdaptiveBitrateDecisionV1 {2762 decision_id: 2,2763 report_id: 2,2764 target_kbps: 12_000,2765 floor_kbps: 1_000,2766 ceiling_kbps: 12_000,2767 state: AdaptiveBitrateStateV1::Increase,2768 reasons: AdaptiveBitrateReasonFlagsV1::CLEAN_RECOVERY,2769 applied: false,2770 },2771 telemetry: MediaV3TelemetrySnapshot::default(),2772 resolution_decision: Some(MotionResolutionDecision {2773 tier: ResolutionTier::Native,2774 target: native,2775 trigger: ResolutionTrigger::Recovery,2776 changed: true,2777 }),2778 force_keyframe: false,2779 };2780 assert!(coordinator.submit(recovered).is_none());27812782 // The first task times out its bitrate wait and then its resolution2783 // wait. Neither timeout changes the last acknowledged state.2784 tokio::task::yield_now().await;2785 tokio::time::advance(ENCODER_CONTROL_COMMIT_TIMEOUT).await;2786 tokio::task::yield_now().await;2787 tokio::time::advance(ENCODER_CONTROL_COMMIT_TIMEOUT).await;2788 tokio::task::yield_now().await;2789 let result = coordinator.acknowledgements.join_next().await;2790 let acknowledgement = coordinator.complete(result).unwrap();2791 assert!(!acknowledgement.plan.decision.applied);2792 assert_eq!(acknowledgement.resolution_applied, Some(false));2793 assert_eq!(coordinator.committed_bitrate_kbps, 12_000);2794 assert_eq!(coordinator.committed_resolution, Some(native));2795 assert_eq!(coordinator.requested_bitrate_kbps, 9_000);2796 assert_eq!(coordinator.requested_resolution, Some(reduced));27972798 // Starting the recovered plan must overwrite the stale persistent2799 // actuator watch even though its targets equal the old committed2800 // values. Otherwise the timed-out request would apply on resume.2801 assert!(coordinator.start_queued().is_none());2802 assert!(coordinator.has_pending());2803 let corrected_bitrate = harness2804 .requested_bitrate()2805 .expect("recovered bitrate must overwrite stale actuator intent");2806 let corrected_resolution = harness2807 .requested_resolution()2808 .expect("recovered resolution must overwrite stale actuator intent");2809 assert!(corrected_bitrate.0 > stale_bitrate.0);2810 assert!(corrected_resolution.0 > stale_resolution.0);2811 assert_eq!(corrected_bitrate.1, 12_000);2812 assert_eq!(2813 (corrected_resolution.1, corrected_resolution.2),2814 (native.width, native.height)2815 );2816 assert_eq!(coordinator.requested_bitrate_kbps, 12_000);2817 assert_eq!(coordinator.requested_resolution, Some(native));28182819 coordinator.abort_and_drain().await;2820 drop(media);2821 }28222823 2824 async fn feedback_loop_writes_every_decision_while_combined_commit_is_hung() {2825 let sessions = Arc::new(SessionRegistry::default());2826 let remote = endpoint(1);2827 let nonce = [9; 16];2828 let media = sessions2829 .claim(remote, nonce, InvitationGrants::VIEW)2830 .unwrap();2831 let harness = crate::source::EncoderControlTestHarness::new();2832 sessions2833 .install_encoder_control(remote, media.session_id, Some(harness.control.clone()))2834 .unwrap();2835 let feedback = sessions.claim_feedback(remote, nonce).unwrap();2836 let session_id = feedback.session_id;28372838 let (mut report_writer, report_reader) = tokio::io::duplex(4_096);2839 let (feedback_sender, mut feedback_receiver) =2840 tokio::sync::watch::channel(CumulativeFeedbackIngress::default());2841 let reader_task = tokio::spawn(forward_media_feedback_reports(2842 report_reader,2843 feedback_sender,2844 ));2845 let (mut decision_writer, mut decision_reader) = tokio::io::duplex(4_096);2846 let loop_sessions = Arc::clone(&sessions);2847 let loop_task = tokio::spawn(async move {2848 run_media_feedback_loop(2849 &mut decision_writer,2850 std::future::pending(),2851 remote,2852 &loop_sessions,2853 &feedback,2854 true,2855 12_000,2856 &mut feedback_receiver,2857 reader_task,2858 )2859 .await2860 });28612862 let mut observed = Vec::new();2863 for report_id in 1..=5 {2864 if report_id > 1 {2865 tokio::time::sleep(Duration::from_millis(800)).await;2866 }2867 let mut report = clean_feedback(report_id);2868 if report_id >= 2 {2869 report.flags = MediaFeedbackFlags::RESYNC_ACTIVE;2870 report.decoder_dropped_delta = 4;2871 }2872 sigil_protocol::write_media_feedback_report_v1(&mut report_writer, &report)2873 .await2874 .unwrap();2875 let decision = tokio::time::timeout(2876 Duration::from_secs(1),2877 sigil_protocol::read_adaptive_bitrate_decision_v1(&mut decision_reader),2878 )2879 .await2880 .expect("each evaluated report must be written before the hung commit timeout")2881 .unwrap()2882 .expect("feedback loop closed before writing its decision");2883 observed.push((decision.decision_id, decision.report_id));2884 }2885 assert_eq!(2886 observed,2887 (1..=5).map(|id| (id, id)).collect::<Vec<_>>(),2888 "slow bitrate and resolution readbacks must not coalesce Portal decisions"2889 );2890 assert!(2891 harness.requested_bitrate().is_some(),2892 "the adversarial run must actually have pending bitrate work"2893 );2894 assert!(2895 harness.requested_resolution().is_some(),2896 "the adversarial run must actually have pending resolution work"2897 );28982899 drop(media);2900 drop(report_writer);2901 tokio::time::timeout(Duration::from_secs(1), loop_task)2902 .await2903 .expect("session teardown must abort the hung commit promptly")2904 .unwrap()2905 .unwrap();2906 assert!(!sessions.is_active(remote, session_id));2907 }29082909 2910 async fn slow_adaptive_commit_does_not_pause_feedback_evaluation() {2911 let sessions = Arc::new(SessionRegistry::default());2912 let remote = endpoint(1);2913 let media = sessions2914 .claim(remote, [4; 16], InvitationGrants::VIEW)2915 .unwrap();2916 let harness = crate::source::EncoderControlTestHarness::new();2917 sessions2918 .install_encoder_control(remote, media.session_id, Some(harness.control.clone()))2919 .unwrap();2920 let mut coordinator = AdaptiveCommitCoordinator::new(2921 Arc::clone(&sessions),2922 remote,2923 media.session_id,2924 true,2925 12_000,2926 Some(native_dimensions()),2927 );2928 let mut controller =2929 ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());2930 let mut resolution_controller = None;2931 let observed_at = Instant::now();29322933 let seed = evaluate_commit_plan(2934 &mut controller,2935 &mut resolution_controller,2936 &clean_feedback(1),2937 observed_at,2938 );2939 assert_eq!(2940 coordinator.submit(seed).unwrap().plan.decision.decision_id,2941 12942 );29432944 let mut severe = clean_feedback(2);2945 severe.decoder_dropped_delta = 1;2946 let first = evaluate_commit_plan(2947 &mut controller,2948 &mut resolution_controller,2949 &severe,2950 observed_at + Duration::from_secs(1),2951 );2952 assert_eq!(first.decision.state, AdaptiveBitrateStateV1::Decrease);2953 assert!(coordinator.submit(first).is_none());2954 assert!(coordinator.has_pending());29552956 tokio::time::advance(Duration::from_secs(1)).await;2957 let evaluation_tick = tokio::time::Instant::now();2958 let mut newer_severe = clean_feedback(3);2959 newer_severe.decoder_dropped_delta = 1;2960 let queued = evaluate_commit_plan(2961 &mut controller,2962 &mut resolution_controller,2963 &newer_severe,2964 observed_at + Duration::from_secs(2),2965 );2966 assert_eq!(queued.decision.decision_id, 3);2967 assert_eq!(queued.decision.state, AdaptiveBitrateStateV1::Decrease);2968 assert!(coordinator.submit(queued).is_none());2969 assert_eq!(tokio::time::Instant::now(), evaluation_tick);2970 assert_eq!(coordinator.acknowledgements.len(), 1);2971 assert_eq!(2972 coordinator.queued.unwrap().decision.decision_id,2973 queued.decision.decision_id2974 );29752976 // Revision one is acknowledged only after another feedback window was2977 // evaluated. The decision retains that exact readback result.2978 harness.status.send_modify(|status| {2979 status.applied_bitrate_revision = Some(1);2980 status.applied_bitrate_kbps = Some(first.decision.target_kbps);2981 });2982 let commit_result = coordinator.acknowledgements.join_next().await;2983 let completed = coordinator.complete(commit_result).unwrap();2984 assert_eq!(completed.plan.decision.decision_id, 2);2985 assert!(completed.plan.decision.applied);2986 assert_eq!(2987 completed.applied_bitrate_kbps,2988 Some(first.decision.target_kbps)2989 );29902991 assert!(coordinator.start_queued().is_none());2992 assert!(coordinator.has_pending());2993 assert_eq!(coordinator.acknowledgements.len(), 1);29942995 let session_id = media.session_id;2996 drop(media);2997 assert!(!sessions.is_active(remote, session_id));2998 let session_end_result = coordinator.acknowledgements.join_next().await;2999 let session_end = coordinator.complete(session_end_result).unwrap();3000 assert!(session_end.generation_ended);3001 coordinator.abort_and_drain().await;3002 assert!(!coordinator.has_pending());3003 assert!(coordinator.queued.is_none());3004 assert!(coordinator.acknowledgements.is_empty());3005 }30063007 3008 async fn timing_out_commits_keep_uniform_hysteresis_and_bounded_work() {3009 let sessions = Arc::new(SessionRegistry::default());3010 let remote = endpoint(1);3011 let media = sessions3012 .claim(remote, [5; 16], InvitationGrants::VIEW)3013 .unwrap();3014 let harness = crate::source::EncoderControlTestHarness::new();3015 sessions3016 .install_encoder_control(remote, media.session_id, Some(harness.control.clone()))3017 .unwrap();3018 let mut coordinator = AdaptiveCommitCoordinator::new(3019 Arc::clone(&sessions),3020 remote,3021 media.session_id,3022 true,3023 12_000,3024 Some(native_dimensions()),3025 );3026 let mut controller =3027 ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());3028 let mut resolution_controller =3029 Some(MotionResolutionPolicy::new(native_dimensions()).unwrap());3030 let observed_at = Instant::now();3031 let virtual_started_at = tokio::time::Instant::now();3032 let mut evaluation_ticks = Vec::new();3033 let mut decision_ids = Vec::new();3034 let mut states = Vec::new();3035 let mut completed = Vec::new();30363037 for report_id in 1..=9 {3038 evaluation_ticks.push(tokio::time::Instant::now());3039 let mut report = clean_feedback(report_id);3040 if report_id > 1 {3041 report.frontend_queue_depth = 2;3042 }3043 let plan = evaluate_commit_plan(3044 &mut controller,3045 &mut resolution_controller,3046 &report,3047 observed_at + Duration::from_secs(report_id - 1),3048 );3049 decision_ids.push(plan.decision.decision_id);3050 states.push(plan.decision.state);3051 if let Some(immediate) = coordinator.submit(plan) {3052 completed.push(immediate);3053 }3054 assert!(3055 coordinator.acknowledgements.len() <= 1,3056 "the coordinator must own at most one commit task"3057 );3058 assert!(3059 coordinator.queued.iter().count() <= 1,3060 "the coordinator must retain only the latest plan"3061 );30623063 tokio::time::advance(Duration::from_secs(1)).await;3064 tokio::task::yield_now().await;3065 if let Some(result) = coordinator.acknowledgements.try_join_next() {3066 completed.push(coordinator.complete(Some(result)).unwrap());3067 if let Some(immediate) = coordinator.start_queued() {3068 completed.push(immediate);3069 }3070 }3071 }30723073 assert_eq!(3074 decision_ids,3075 (1..=9).collect::<Vec<_>>(),3076 "coalescing must not pause or reuse decision IDs"3077 );3078 assert_eq!(3079 states,3080 [3081 AdaptiveBitrateStateV1::Hold,3082 AdaptiveBitrateStateV1::Hold,3083 AdaptiveBitrateStateV1::Decrease,3084 AdaptiveBitrateStateV1::Hold,3085 AdaptiveBitrateStateV1::Decrease,3086 AdaptiveBitrateStateV1::Hold,3087 AdaptiveBitrateStateV1::Decrease,3088 AdaptiveBitrateStateV1::Hold,3089 AdaptiveBitrateStateV1::Decrease,3090 ],3091 "moderate-pressure hysteresis must advance once per feedback window"3092 );3093 for (index, tick) in evaluation_ticks.into_iter().enumerate() {3094 assert_eq!(3095 tick.duration_since(virtual_started_at),3096 Duration::from_secs(index as u64),3097 "feedback evaluation cadence must remain uniform during commit timeouts"3098 );3099 }3100 assert!(3101 completed.iter().any(|acknowledgement| {3102 acknowledgement.plan.decision.decision_id == 33103 && !acknowledgement.plan.decision.applied3104 && acknowledgement.resolution_applied == Some(false)3105 }),3106 "the first bitrate and resolution timeout must retain exact negative acknowledgement"3107 );3108 assert!(coordinator.has_pending());3109 assert_eq!(coordinator.acknowledgements.len(), 1);3110 assert!(coordinator.queued.is_some());31113112 coordinator.abort_and_drain().await;3113 assert!(!coordinator.has_pending());3114 assert!(coordinator.queued.is_none());3115 assert!(coordinator.acknowledgements.is_empty());31163117 let session_id = media.session_id;3118 drop(media);3119 assert!(!sessions.is_active(remote, session_id));3120 }3121}