Use paused virtual time to prove slow and timing-out encoder commits do not disturb feedback cadence, decision IDs, or moderate-pressure hysteresis. Assert exact acknowledgements, bounded coordinator state, and teardown draining.
- commit
- 57984d8d05c6c0b69e0bc1d95f7b8bff9d5be16e
- parent
- d2440180f2a0340fbf1f0962bd21c4c036ba13c6
- author
- Dami <dami.etoile@gmail.com>
| M | crates/ | 1 | + |
| M | crates/ | 232 | ++++++++++++++++++++++++++++++ |
@@ -56,3 +56,4 @@ x11rb = { version = "0.13.2", default-features = false }
5656 5757 [dev-dependencies]5858 tempfile = "3"59+tokio = { version = "1", features = ["test-util"] }@@ -1621,6 +1621,25 @@ mod tests {
16211621 .unwrap()16221622 }16231623 1624+ fn evaluate_commit_plan(1625+ controller: &mut ShadowBitrateController,1626+ resolution_controller: &mut Option<MotionResolutionPolicy>,1627+ report: &MediaFeedbackReportV1,1628+ observed_at: Instant,1629+ ) -> AdaptiveCommitPlan {1630+ let telemetry = MediaV3TelemetrySnapshot::default();1631+ let evaluation = controller.evaluate(report, telemetry, observed_at).unwrap();1632+ let resolution_decision = resolution_controller1633+ .as_mut()1634+ .map(|policy| policy.observe_window(report, &evaluation, observed_at));1635+ AdaptiveCommitPlan {1636+ decision: evaluation.decision,1637+ telemetry,1638+ resolution_decision,1639+ force_keyframe: adaptive_recovery_keyframe_required(report, &evaluation.decision),1640+ }1641+ }1642+16241643 fn native_dimensions() -> VideoDimensions {16251644 VideoDimensions {16261645 width: 1_280,@@ -2488,4 +2507,217 @@ mod tests {
24882507 "a clamped aggregate must be stale rather than disguised as one fresh report"24892508 );24902509 }2510+2511+ 2512+ async fn slow_adaptive_commit_does_not_pause_feedback_evaluation() {2513+ let sessions = Arc::new(SessionRegistry::default());2514+ let remote = endpoint(1);2515+ let media = sessions2516+ .claim(remote, [4; 16], InvitationGrants::VIEW)2517+ .unwrap();2518+ let harness = crate::source::EncoderControlTestHarness::new();2519+ sessions2520+ .install_encoder_control(remote, media.session_id, Some(harness.control.clone()))2521+ .unwrap();2522+ let mut coordinator = AdaptiveCommitCoordinator::new(2523+ Arc::clone(&sessions),2524+ remote,2525+ media.session_id,2526+ true,2527+ 12_000,2528+ Some(native_dimensions()),2529+ );2530+ let mut controller =2531+ ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());2532+ let mut resolution_controller = None;2533+ let observed_at = Instant::now();2534+2535+ let seed = evaluate_commit_plan(2536+ &mut controller,2537+ &mut resolution_controller,2538+ &clean_feedback(1),2539+ observed_at,2540+ );2541+ assert_eq!(2542+ coordinator.submit(seed).unwrap().plan.decision.decision_id,2543+ 12544+ );2545+2546+ let mut severe = clean_feedback(2);2547+ severe.decoder_dropped_delta = 1;2548+ let first = evaluate_commit_plan(2549+ &mut controller,2550+ &mut resolution_controller,2551+ &severe,2552+ observed_at + Duration::from_secs(1),2553+ );2554+ assert_eq!(first.decision.state, AdaptiveBitrateStateV1::Decrease);2555+ assert!(coordinator.submit(first).is_none());2556+ assert!(coordinator.has_pending());2557+2558+ tokio::time::advance(Duration::from_secs(1)).await;2559+ let evaluation_tick = tokio::time::Instant::now();2560+ let mut newer_severe = clean_feedback(3);2561+ newer_severe.decoder_dropped_delta = 1;2562+ let queued = evaluate_commit_plan(2563+ &mut controller,2564+ &mut resolution_controller,2565+ &newer_severe,2566+ observed_at + Duration::from_secs(2),2567+ );2568+ assert_eq!(queued.decision.decision_id, 3);2569+ assert_eq!(queued.decision.state, AdaptiveBitrateStateV1::Decrease);2570+ assert!(coordinator.submit(queued).is_none());2571+ assert_eq!(tokio::time::Instant::now(), evaluation_tick);2572+ assert_eq!(coordinator.acknowledgements.len(), 1);2573+ assert_eq!(2574+ coordinator.queued.unwrap().decision.decision_id,2575+ queued.decision.decision_id2576+ );2577+2578+ // Revision one is acknowledged only after another feedback window was2579+ // evaluated. The decision retains that exact readback result.2580+ harness.status.send_modify(|status| {2581+ status.applied_bitrate_revision = Some(1);2582+ status.applied_bitrate_kbps = Some(first.decision.target_kbps);2583+ });2584+ let commit_result = coordinator.acknowledgements.join_next().await;2585+ let completed = coordinator.complete(commit_result).unwrap();2586+ assert_eq!(completed.plan.decision.decision_id, 2);2587+ assert!(completed.plan.decision.applied);2588+ assert_eq!(2589+ completed.applied_bitrate_kbps,2590+ Some(first.decision.target_kbps)2591+ );2592+2593+ assert!(coordinator.start_queued().is_none());2594+ assert!(coordinator.has_pending());2595+ assert_eq!(coordinator.acknowledgements.len(), 1);2596+2597+ let session_id = media.session_id;2598+ drop(media);2599+ assert!(!sessions.is_active(remote, session_id));2600+ let session_end_result = coordinator.acknowledgements.join_next().await;2601+ let session_end = coordinator.complete(session_end_result).unwrap();2602+ assert!(session_end.generation_ended);2603+ coordinator.abort_and_drain().await;2604+ assert!(!coordinator.has_pending());2605+ assert!(coordinator.queued.is_none());2606+ assert!(coordinator.acknowledgements.is_empty());2607+ }2608+2609+ 2610+ async fn timing_out_commits_keep_uniform_hysteresis_and_bounded_work() {2611+ let sessions = Arc::new(SessionRegistry::default());2612+ let remote = endpoint(1);2613+ let media = sessions2614+ .claim(remote, [5; 16], InvitationGrants::VIEW)2615+ .unwrap();2616+ let harness = crate::source::EncoderControlTestHarness::new();2617+ sessions2618+ .install_encoder_control(remote, media.session_id, Some(harness.control.clone()))2619+ .unwrap();2620+ let mut coordinator = AdaptiveCommitCoordinator::new(2621+ Arc::clone(&sessions),2622+ remote,2623+ media.session_id,2624+ true,2625+ 12_000,2626+ Some(native_dimensions()),2627+ );2628+ let mut controller =2629+ ShadowBitrateController::new(12_000, MediaV3TelemetrySnapshot::default());2630+ let mut resolution_controller =2631+ Some(MotionResolutionPolicy::new(native_dimensions()).unwrap());2632+ let observed_at = Instant::now();2633+ let virtual_started_at = tokio::time::Instant::now();2634+ let mut evaluation_ticks = Vec::new();2635+ let mut decision_ids = Vec::new();2636+ let mut states = Vec::new();2637+ let mut completed = Vec::new();2638+2639+ for report_id in 1..=9 {2640+ evaluation_ticks.push(tokio::time::Instant::now());2641+ let mut report = clean_feedback(report_id);2642+ if report_id > 1 {2643+ report.frontend_queue_depth = 2;2644+ }2645+ let plan = evaluate_commit_plan(2646+ &mut controller,2647+ &mut resolution_controller,2648+ &report,2649+ observed_at + Duration::from_secs(report_id - 1),2650+ );2651+ decision_ids.push(plan.decision.decision_id);2652+ states.push(plan.decision.state);2653+ if let Some(immediate) = coordinator.submit(plan) {2654+ completed.push(immediate);2655+ }2656+ assert!(2657+ coordinator.acknowledgements.len() <= 1,2658+ "the coordinator must own at most one commit task"2659+ );2660+ assert!(2661+ coordinator.queued.iter().count() <= 1,2662+ "the coordinator must retain only the latest plan"2663+ );2664+2665+ tokio::time::advance(Duration::from_secs(1)).await;2666+ tokio::task::yield_now().await;2667+ if let Some(result) = coordinator.acknowledgements.try_join_next() {2668+ completed.push(coordinator.complete(Some(result)).unwrap());2669+ if let Some(immediate) = coordinator.start_queued() {2670+ completed.push(immediate);2671+ }2672+ }2673+ }2674+2675+ assert_eq!(2676+ decision_ids,2677+ (1..=9).collect::<Vec<_>>(),2678+ "coalescing must not pause or reuse decision IDs"2679+ );2680+ assert_eq!(2681+ states,2682+ [2683+ AdaptiveBitrateStateV1::Hold,2684+ AdaptiveBitrateStateV1::Hold,2685+ AdaptiveBitrateStateV1::Decrease,2686+ AdaptiveBitrateStateV1::Hold,2687+ AdaptiveBitrateStateV1::Decrease,2688+ AdaptiveBitrateStateV1::Hold,2689+ AdaptiveBitrateStateV1::Decrease,2690+ AdaptiveBitrateStateV1::Hold,2691+ AdaptiveBitrateStateV1::Decrease,2692+ ],2693+ "moderate-pressure hysteresis must advance once per feedback window"2694+ );2695+ for (index, tick) in evaluation_ticks.into_iter().enumerate() {2696+ assert_eq!(2697+ tick.duration_since(virtual_started_at),2698+ Duration::from_secs(index as u64),2699+ "feedback evaluation cadence must remain uniform during commit timeouts"2700+ );2701+ }2702+ assert!(2703+ completed.iter().any(|acknowledgement| {2704+ acknowledgement.plan.decision.decision_id == 32705+ && !acknowledgement.plan.decision.applied2706+ && acknowledgement.resolution_applied == Some(false)2707+ }),2708+ "the first bitrate and resolution timeout must retain exact negative acknowledgement"2709+ );2710+ assert!(coordinator.has_pending());2711+ assert_eq!(coordinator.acknowledgements.len(), 1);2712+ assert!(coordinator.queued.is_some());2713+2714+ coordinator.abort_and_drain().await;2715+ assert!(!coordinator.has_pending());2716+ assert!(coordinator.queued.is_none());2717+ assert!(coordinator.acknowledgements.is_empty());2718+2719+ let session_id = media.session_id;2720+ drop(media);2721+ assert!(!sessions.is_active(remote, session_id));2722+ }24912723 }