Keep a queued forced-IDR requirement while replacing its decision and report identity with the latest bounded plan. Remove the obsolete pre-coordinator rollback helper and cover the recovery coalescing edge.
- commit
- f409f48e271ab9fd85ebc4988d9ada2ddfd67ee4
- parent
- 57984d8d05c6c0b69e0bc1d95f7b8bff9d5be16e
- author
- Dami <dami.etoile@gmail.com>
| M | crates/ | 123 | ++++++++++++++++++++---------- |
@@ -816,16 +816,6 @@ impl ShadowBitrateController {
816816 trusted_host_pressure,817817 )818818 }819-820- 821- fn commit_observation_from(&mut self, candidate: &Self) {822- self.next_decision_id = candidate.next_decision_id;823- self.last_report_id = candidate.last_report_id;824- self.last_report_received_at = candidate.last_report_received_at;825- self.previous_telemetry = candidate.previous_telemetry;826- self.baseline_rtt_micros = candidate.baseline_rtt_micros;827- self.last_sequence = candidate.last_sequence;828- }829819 }830820 831821 fn add_feedback_signal(@@ -928,8 +918,14 @@ impl AdaptiveCommitCoordinator {
928918 self.pending929919 }930920 931- fn submit(&mut self, plan: AdaptiveCommitPlan) -> Option<AdaptiveCommitAcknowledgement> {921+ fn submit(&mut self, mut plan: AdaptiveCommitPlan) -> Option<AdaptiveCommitAcknowledgement> {932922 if self.pending {923+ if let Some(queued) = self.queued {924+ // The latest decision replaces older diagnostics, but a925+ // recovery barrier requested by any coalesced desired update926+ // must survive until that desired state is actuated.927+ plan.force_keyframe |= queued.force_keyframe;928+ }933929 self.queued = Some(plan);934930 return None;935931 }@@ -2356,36 +2352,6 @@ mod tests {
23562352 ));23572353 }23582354 2359- 2360- fn failed_adaptive_application_commits_observation_but_not_target() {2361- let mut committed =2362- ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());2363- controller_decide(&mut committed, &clean_feedback(1), 0);2364- let old_target = committed.target_kbps;2365- let old_decision_id = committed.next_decision_id;2366- let mut severe = clean_feedback(2);2367- severe.flags = MediaFeedbackFlags::RESYNC_ACTIVE;2368-2369- let mut candidate = committed.clone();2370- let decision = controller_decide(&mut candidate, &severe, 1);2371- assert_eq!(decision.state, AdaptiveBitrateStateV1::Decrease);2372- assert!(candidate.target_kbps < old_target);2373- committed.commit_observation_from(&candidate);2374-2375- assert_eq!(committed.target_kbps, old_target);2376- assert_eq!(committed.last_report_id, Some(2));2377- assert_eq!(committed.next_decision_id, old_decision_id + 1);2378- assert!(2379- committed2380- .decide(2381- &clean_feedback(3),2382- MediaV3TelemetrySnapshot::default(),2383- Instant::now() + Duration::from_secs(2)2384- )2385- .is_ok()2386- );2387- }2388-23892355 23902356 fn adaptive_recovery_idr_is_only_requested_for_a_bitrate_change_during_resync() {23912357 let mut report = clean_feedback(1);@@ -2508,6 +2474,81 @@ mod tests {
25082474 );25092475 }25102476 2477+ 2478+ async fn queued_adaptive_plan_keeps_latest_ids_and_recovery_barrier() {2479+ let sessions = Arc::new(SessionRegistry::default());2480+ let remote = endpoint(1);2481+ let media = sessions2482+ .claim(remote, [3; 16], InvitationGrants::VIEW)2483+ .unwrap();2484+ let harness = crate::source::EncoderControlTestHarness::new();2485+ sessions2486+ .install_encoder_control(remote, media.session_id, Some(harness.control.clone()))2487+ .unwrap();2488+ let mut coordinator = AdaptiveCommitCoordinator::new(2489+ Arc::clone(&sessions),2490+ remote,2491+ media.session_id,2492+ true,2493+ 12_000,2494+ Some(native_dimensions()),2495+ );2496+ let mut controller =2497+ ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());2498+ let mut resolution_controller = None;2499+ let observed_at = Instant::now();2500+2501+ let seed = evaluate_commit_plan(2502+ &mut controller,2503+ &mut resolution_controller,2504+ &clean_feedback(1),2505+ observed_at,2506+ );2507+ assert!(coordinator.submit(seed).is_some());2508+2509+ let mut first_pressure = clean_feedback(2);2510+ first_pressure.decoder_dropped_delta = 1;2511+ let first = evaluate_commit_plan(2512+ &mut controller,2513+ &mut resolution_controller,2514+ &first_pressure,2515+ observed_at + Duration::from_secs(1),2516+ );2517+ assert!(!first.force_keyframe);2518+ assert!(coordinator.submit(first).is_none());2519+2520+ let mut recovery_pressure = clean_feedback(3);2521+ recovery_pressure.decoder_dropped_delta = 1;2522+ recovery_pressure.flags = MediaFeedbackFlags::RESYNC_ACTIVE;2523+ let recovery = evaluate_commit_plan(2524+ &mut controller,2525+ &mut resolution_controller,2526+ &recovery_pressure,2527+ observed_at + Duration::from_secs(2),2528+ );2529+ assert!(recovery.force_keyframe);2530+ assert!(coordinator.submit(recovery).is_none());2531+2532+ let latest = evaluate_commit_plan(2533+ &mut controller,2534+ &mut resolution_controller,2535+ &clean_feedback(4),2536+ observed_at + Duration::from_secs(3),2537+ );2538+ assert!(!latest.force_keyframe);2539+ assert!(coordinator.submit(latest).is_none());2540+ let retained = coordinator.queued.expect("latest plan must be retained");2541+ assert_eq!(retained.decision.decision_id, latest.decision.decision_id);2542+ assert_eq!(retained.decision.report_id, latest.decision.report_id);2543+ assert!(2544+ retained.force_keyframe,2545+ "a coalesced recovery barrier must survive a newer hold decision"2546+ );2547+2548+ coordinator.abort_and_drain().await;2549+ drop(media);2550+ }2551+25112552 25122553 async fn slow_adaptive_commit_does_not_pause_feedback_evaluation() {25132554 let sessions = Arc::new(SessionRegistry::default());