Route explicit non-Join recovery through a bounded coalescing forced-IDR coordinator while preserving Join replay and external encoder fallback. Add recovery latency evidence and transport regression coverage.
- commit
- f36cd090117752cd0e41b042a0d4f547cb345e1e
- parent
- ef404d700d7d66be5be86a27a2a91a8c9ff31e75
- author
- FelineStateMachine <dami.etoile@gmail.com>
| M | README.md | 8 | +- |
| M | crates/ | 16 | + |
| M | crates/ | 350 | ++++++++++++++++++++++++++++++- |
| M | crates/ | 63 | +++++ |
| M | scripts/ | 8 | +- |
@@ -115,10 +115,10 @@ not authentication.
115115 5. **Upstream Iroh/MoQ transport implemented; hardware proof pending:** an116116 authenticated control lease gates a session-scoped upstream MoQ broadcast;117117 bounded native groups cancel stale GOPs and recover through explicit118- keyframe requests. The external encoder responds at its next natural119- configured IDR. An opt-in in-process encoder-control foundation exists, but120- its Bazzite/package acceptance and induced-loss relay proof remain in issue121- #7.118+ keyframe requests. The opt-in in-process encoder turns recovery requests119+ into coalesced forced IDRs without blocking media publication; the external120+ encoder responds at its next natural configured IDR. Bazzite/package121+ acceptance and induced-loss relay proof remain in issue #7.122122 6. **Host hardware-proven; physical client controller pending:** the virtual123123 Xbox-style controller negotiated over Iroh and produced the expected124124 button, stick, trigger, D-pad, and neutral-release events on the Bazzite@@ -1159,6 +1159,8 @@ async fn main() -> Result<()> {
11591159 let mut object_sequence_v3 = MediaObjectSequenceV3::new();11601160 let mut keyframe_request_sent = false;11611161 let mut keyframe_recovery_verified = false;1162+ let mut keyframe_request_sent_at: Option<Instant> = None;1163+ let mut keyframe_recovery_micros = None;11621164 let mut keyframe_request_group_id = None;11631165 let mut keyframe_request_last_sequence = None;11641166 @@ -1204,6 +1206,9 @@ async fn main() -> Result<()> {
12041206 .is_some_and(|prior| frame.header.sequence > prior),12051207 "MoQ keyframe request recovery did not advance media sequence"12061208 );1209+ keyframe_recovery_micros = keyframe_request_sent_at.map(|sent_at| {1210+ u64::try_from(sent_at.elapsed().as_micros()).unwrap_or(u64::MAX)1211+ });12071212 keyframe_recovery_verified = true;12081213 }12091214 break (@@ -1329,6 +1334,11 @@ async fn main() -> Result<()> {
13291334 }),13301335 "keyframe request recovery did not advance media sequence"13311336 );1337+ keyframe_recovery_micros =1338+ keyframe_request_sent_at.map(|sent_at| {1339+ u64::try_from(sent_at.elapsed().as_micros())1340+ .unwrap_or(u64::MAX)1341+ });13321342 keyframe_recovery_verified = true;13331343 }13341344 break (object.into(), recovery);@@ -1398,6 +1408,7 @@ async fn main() -> Result<()> {
13981408 .context("timed out writing v3 keyframe request")??;13991409 keyframe_request_group_id = frame.v3_group_id;14001410 keyframe_request_last_sequence = Some(frame.sequence);1411+ keyframe_request_sent_at = Some(Instant::now());14011412 keyframe_request_sent = true;14021413 match media_transport {14031414 MediaTransport::UpstreamMoq => {@@ -1561,8 +1572,13 @@ async fn main() -> Result<()> {
15611572 );15621573 if args.keyframe_smoke {15631574 println!("keyframe_request_id={}", args.keyframe_request_id);1575+ println!(1576+ "keyframe_recovery_micros={}",1577+ keyframe_recovery_micros.context("keyframe recovery latency was not measured")?1578+ );15641579 } else {15651580 println!("keyframe_request_id=not-requested");1581+ println!("keyframe_recovery_micros=not-requested");15661582 }15671583 match accepted_fps {15681584 Some(fps) => println!("accepted_fps={fps:.3}"),@@ -979,6 +979,11 @@ struct MediaV3Telemetry {
979979 selected_path_rtt_micros: AtomicU64,980980 selected_path_lost_packets: AtomicU64,981981 selected_path_congestion_events: AtomicU64,982+ keyframe_control_requests: AtomicU64,983+ encoder_force_requests: AtomicU64,984+ encoder_force_acknowledgements: AtomicU64,985+ encoder_force_failures: AtomicU64,986+ last_encoder_force_ack_micros: AtomicU64,982987 }983988 984989 @@ -1005,6 +1010,178 @@ struct ResolutionEncoderProposal {
10051010 revision: u64,10061011 }10071012 1013+1014+enum ForcedIdrDisposition {1015+ JoinReplay,1016+ Unavailable,1017+ Requested { revision: u64 },1018+ Coalesced { revision: u64 },1019+ Failed { error: String },1020+}1021+1022+struct ForcedIdrAcknowledgement {1023+ requested_revision: u64,1024+ elapsed: Duration,1025+ result: Result<crate::source::EncoderControlStatus>,1026+}1027+1028+struct ForcedIdrCoordinator {1029+ control: Option<EncoderControl>,1030+ pending_revision: Option<u64>,1031+ acknowledgements: tokio::task::JoinSet<ForcedIdrAcknowledgement>,1032+ telemetry: Arc<MediaV3Telemetry>,1033+}1034+1035+impl ForcedIdrCoordinator {1036+ fn new(control: Option<EncoderControl>, telemetry: Arc<MediaV3Telemetry>) -> Self {1037+ Self {1038+ control,1039+ pending_revision: None,1040+ acknowledgements: tokio::task::JoinSet::new(),1041+ telemetry,1042+ }1043+ }1044+1045+ fn request(&mut self, reason: KeyframeRequestReasonV3) -> ForcedIdrDisposition {1046+ self.telemetry1047+ .keyframe_control_requests1048+ .fetch_add(1, Ordering::Relaxed);1049+ if reason == KeyframeRequestReasonV3::Join {1050+ return ForcedIdrDisposition::JoinReplay;1051+ }1052+ if let Some(revision) = self.pending_revision {1053+ return ForcedIdrDisposition::Coalesced { revision };1054+ }1055+ let Some(control) = self.control.clone() else {1056+ return ForcedIdrDisposition::Unavailable;1057+ };1058+ let revision = match control.request_force_keyframe() {1059+ Ok(revision) => revision,1060+ Err(error) => {1061+ self.telemetry1062+ .encoder_force_failures1063+ .fetch_add(1, Ordering::Relaxed);1064+ return ForcedIdrDisposition::Failed {1065+ error: error.to_string(),1066+ };1067+ }1068+ };1069+ self.pending_revision = Some(revision);1070+ self.telemetry1071+ .encoder_force_requests1072+ .fetch_add(1, Ordering::Relaxed);1073+ self.acknowledgements.spawn(async move {1074+ let started_at = Instant::now();1075+ let result = control1076+ .wait_for_recovery_keyframe_acknowledged(revision, ENCODER_CONTROL_COMMIT_TIMEOUT)1077+ .await;1078+ ForcedIdrAcknowledgement {1079+ requested_revision: revision,1080+ elapsed: started_at.elapsed(),1081+ result,1082+ }1083+ });1084+ ForcedIdrDisposition::Requested { revision }1085+ }1086+1087+ fn complete(1088+ &mut self,1089+ result: Option<Result<ForcedIdrAcknowledgement, tokio::task::JoinError>>,1090+ remote: EndpointId,1091+ transport: &'static str,1092+ ) {1093+ let pending_revision = self.pending_revision.take();1094+ match result {1095+ Some(Ok(acknowledgement)) => match acknowledgement.result {1096+ Ok(status) => {1097+ let elapsed_micros =1098+ u64::try_from(acknowledgement.elapsed.as_micros()).unwrap_or(u64::MAX);1099+ self.telemetry1100+ .encoder_force_acknowledgements1101+ .fetch_add(1, Ordering::Relaxed);1102+ self.telemetry1103+ .last_encoder_force_ack_micros1104+ .store(elapsed_micros, Ordering::Relaxed);1105+ debug!(1106+ %remote,1107+ transport,1108+ requested_revision = acknowledgement.requested_revision,1109+ acknowledged_revision = ?status.acknowledged_force_keyframe_revision,1110+ elapsed_micros,1111+ "forced-IDR recovery acknowledged"1112+ );1113+ }1114+ Err(error) => {1115+ self.telemetry1116+ .encoder_force_failures1117+ .fetch_add(1, Ordering::Relaxed);1118+ warn!(1119+ %remote,1120+ transport,1121+ requested_revision = acknowledgement.requested_revision,1122+ %error,1123+ "forced-IDR recovery was not acknowledged; retaining natural-IDR fallback"1124+ );1125+ }1126+ },1127+ Some(Err(error)) => {1128+ self.telemetry1129+ .encoder_force_failures1130+ .fetch_add(1, Ordering::Relaxed);1131+ warn!(1132+ %remote,1133+ transport,1134+ ?pending_revision,1135+ %error,1136+ "forced-IDR acknowledgement task failed; retaining natural-IDR fallback"1137+ );1138+ }1139+ None => {1140+ self.telemetry1141+ .encoder_force_failures1142+ .fetch_add(1, Ordering::Relaxed);1143+ warn!(1144+ %remote,1145+ transport,1146+ ?pending_revision,1147+ "forced-IDR acknowledgement task ended without a result"1148+ );1149+ }1150+ }1151+ }1152+1153+ async fn abort_and_drain(&mut self, remote: EndpointId, transport: &'static str) {1154+ self.pending_revision = None;1155+ self.acknowledgements.abort_all();1156+ while self.acknowledgements.join_next().await.is_some() {}1157+ debug!(1158+ %remote,1159+ transport,1160+ keyframe_control_requests = self1161+ .telemetry1162+ .keyframe_control_requests1163+ .load(Ordering::Relaxed),1164+ encoder_force_requests = self1165+ .telemetry1166+ .encoder_force_requests1167+ .load(Ordering::Relaxed),1168+ encoder_force_acknowledgements = self1169+ .telemetry1170+ .encoder_force_acknowledgements1171+ .load(Ordering::Relaxed),1172+ encoder_force_failures = self1173+ .telemetry1174+ .encoder_force_failures1175+ .load(Ordering::Relaxed),1176+ last_encoder_force_ack_micros = self1177+ .telemetry1178+ .last_encoder_force_ack_micros1179+ .load(Ordering::Relaxed),1180+ "forced-IDR recovery session summary"1181+ );1182+ }1183+}1184+10081185 impl MediaV3Telemetry {10091186 fn snapshot(&self) -> MediaV3TelemetrySnapshot {10101187 MediaV3TelemetrySnapshot {@@ -2118,6 +2295,22 @@ fn apply_media_v3_keyframe_request(
21182295 (true, cancel_sequences)21192296 }21202297 2298+fn apply_moq_keyframe_request(2299+ publisher: &mut MoqGroupPublisher,2300+ replay_cursor: &mut MediaReplayCursor,2301+ through_sequence: Option<u64>,2302+ reason: KeyframeRequestReasonV3,2303+) -> Option<u64> {2304+ // The bounded current group is already the late joiner's decodable replay.2305+ // Aborting it on Join can strand a static source until its next natural IDR.2306+ if reason == KeyframeRequestReasonV3::Join {2307+ return None;2308+ }2309+ let cancelled_group = publisher.request_keyframe();2310+ replay_cursor.enter_resync_through(through_sequence);2311+ cancelled_group2312+}2313+21212314 fn apply_media_v3_send_failure(21222315 scheduler: &mut MediaV3Scheduler,21232316 group_cursor: &mut MediaV3GroupCursor,@@ -2687,7 +2880,6 @@ async fn serve_control_moq(
26872880 };26882881 let source_task = SourceTaskGuard::new(source_task);26892882 sessions.install_encoder_control(remote, lease.session_id, encoder_control.clone())?;2690- let _encoder_control = encoder_control;26912883 26922884 let mut broadcast = Broadcast::new().produce();26932885 let track = broadcast@@ -2748,6 +2940,7 @@ async fn serve_control_moq(
27482940 closed,27492941 track,27502942 &mut broadcast,2943+ encoder_control,27512944 Arc::clone(&lease.media_v3_telemetry),27522945 )27532946 .await;@@ -2774,6 +2967,7 @@ async fn run_control_moq_session(
27742967 mut moq_closed: tokio::sync::oneshot::Receiver<()>,27752968 track: TrackProducer,27762969 broadcast: &mut BroadcastProducer,2970+ encoder_control: Option<EncoderControl>,27772971 telemetry: Arc<MediaV3Telemetry>,27782972 ) -> Result<()> {27792973 let maximum_replay_age = maximum_media_replay_age(config.framerate);@@ -2786,6 +2980,7 @@ async fn run_control_moq_session(
27862980 ));27872981 let mut control_task_finished = false;27882982 let mut control_receiver_open = true;2983+ let mut forced_idr = ForcedIdrCoordinator::new(encoder_control, Arc::clone(&telemetry));27892984 27902985 let result = async {27912986 loop {@@ -2818,13 +3013,27 @@ async fn run_control_moq_session(
28183013 .as_ref()28193014 .and_then(|gop| gop.frames.last())28203015 .map(|frame| frame.sequence);2821- let cancelled_group = publisher.request_keyframe();3016+ let cancelled_group = apply_moq_keyframe_request(3017+ &mut publisher,3018+ &mut replay_cursor,3019+ through_sequence,3020+ request.reason,3021+ );28223022 if cancelled_group.is_some() {28233023 telemetry28243024 .scheduler_cancellations28253025 .fetch_add(1, Ordering::Relaxed);28263026 }2827- replay_cursor.enter_resync_through(through_sequence);3027+ let forced_idr_disposition = forced_idr.request(request.reason);3028+ if let ForcedIdrDisposition::Failed { error } = &forced_idr_disposition {3029+ warn!(3030+ %remote,3031+ request_id = request.request_id,3032+ ?request.reason,3033+ %error,3034+ "forced-IDR request failed; retaining natural-IDR fallback"3035+ );3036+ }28283037 debug!(28293038 %remote,28303039 request_id = request.request_id,@@ -2832,9 +3041,15 @@ async fn run_control_moq_session(
28323041 advisory_last_sequence = ?request.last_sequence,28333042 coalesced = cancelled_group.is_none(),28343043 ?cancelled_group,3044+ ?forced_idr_disposition,28353045 "accepted MoQ keyframe request"28363046 );28373047 }3048+ acknowledgement = forced_idr.acknowledgements.join_next(),3049+ if forced_idr.pending_revision.is_some() =>3050+ {3051+ forced_idr.complete(acknowledgement, remote, "iroh-moq");3052+ }28383053 reason = connection.closed() => {28393054 debug!(%remote, ?reason, "MoQ control connection closed");28403055 return Ok(());@@ -2932,6 +3147,7 @@ async fn run_control_moq_session(
29323147 }29333148 .await;29343149 3150+ forced_idr.abort_and_drain(remote, "iroh-moq").await;29353151 publisher.abort();29363152 let _ = broadcast.abort(MoqError::Cancel);29373153 if !control_task_finished {@@ -3791,7 +4007,6 @@ async fn serve_media_v3(
37914007 };37924008 let source_task = SourceTaskGuard::new(source_task);37934009 sessions.install_encoder_control(remote, lease.session_id, encoder_control.clone())?;3794- let _encoder_control = encoder_control;37954010 37964011 let mut media_hello = HostHello::accepted(37974012 lease.session_id,@@ -3812,6 +4027,7 @@ async fn serve_media_v3(
38124027 &mut current_gop_receiver,38134028 recv,38144029 remote,4030+ encoder_control,38154031 Arc::clone(&lease.media_v3_telemetry),38164032 )38174033 .await;@@ -4036,6 +4252,7 @@ async fn run_media_v3_session(
40364252 current_gop_receiver: &mut tokio::sync::watch::Receiver<Option<EncodedGop>>,40374253 control_recv: iroh::endpoint::RecvStream,40384254 remote: EndpointId,4255+ encoder_control: Option<EncoderControl>,40394256 telemetry: Arc<MediaV3Telemetry>,40404257 ) -> Result<()> {40414258 let maximum_replay_age = maximum_media_replay_age(config.framerate);@@ -4051,6 +4268,7 @@ async fn run_media_v3_session(
40514268 ));40524269 let mut control_task_finished = false;40534270 let mut control_receiver_open = true;4271+ let mut forced_idr = ForcedIdrCoordinator::new(encoder_control, Arc::clone(&telemetry));40544272 40554273 let result = loop {40564274 telemetry.record_selected_path(connection);@@ -4101,6 +4319,16 @@ async fn run_media_v3_session(
41014319 );41024320 send_tasks.abort_all();41034321 }4322+ let forced_idr_disposition = forced_idr.request(request.reason);4323+ if let ForcedIdrDisposition::Failed { error } = &forced_idr_disposition {4324+ warn!(4325+ %remote,4326+ request_id = request.request_id,4327+ ?request.reason,4328+ %error,4329+ "forced-IDR request failed; retaining natural-IDR fallback"4330+ );4331+ }41044332 debug!(41054333 %remote,41064334 request_id = request.request_id,@@ -4108,9 +4336,15 @@ async fn run_media_v3_session(
41084336 advisory_last_sequence = ?request.last_sequence,41094337 coalesced = !transitioned,41104338 ?cancel_sequences,4339+ ?forced_idr_disposition,41114340 "accepted media v3 keyframe request"41124341 );41134342 }4343+ acknowledgement = forced_idr.acknowledgements.join_next(),4344+ if forced_idr.pending_revision.is_some() =>4345+ {4346+ forced_idr.complete(acknowledgement, remote, "grouped-v3");4347+ }41144348 closed = connection.closed() => {41154349 debug!(%remote, ?closed, "media v3 connection closed");41164350 break Ok(());@@ -4347,6 +4581,7 @@ async fn run_media_v3_session(
43474581 }43484582 };43494583 4584+ forced_idr.abort_and_drain(remote, "grouped-v3").await;43504585 send_tasks.abort_all();43514586 while send_tasks.join_next().await.is_some() {}43524587 if !control_task_finished {@@ -5676,6 +5911,53 @@ mod tests {
56765911 assert!(frame.header.flags.contains(FrameFlags::DISCONTINUITY));56775912 }56785913 5914+ 5915+ async fn upstream_moq_late_join_preserves_active_static_group() {5916+ let track_info = Track {5917+ name: MOQ_VIDEO_H264_TRACK.to_owned(),5918+ priority: MOQ_VIDEO_TRACK_PRIORITY,5919+ };5920+ let mut broadcast = Broadcast::new().produce();5921+ let track = broadcast.create_track(track_info.clone()).unwrap();5922+ let mut consumer = broadcast.consume().subscribe_track(&track_info).unwrap();5923+ let mut publisher = MoqGroupPublisher::new(track);5924+ let mut replay_cursor = MediaReplayCursor::default();5925+ let config = moq_test_config();5926+ let keyframe = media_v3_encoded_frame(10, true, true, 1);5927+ let first_delta = media_v3_encoded_frame(11, false, false, 1);5928+ let next_delta = media_v3_encoded_frame(12, false, false, 1);5929+5930+ publisher.publish(&config, &keyframe, false).unwrap();5931+ replay_cursor.commit_sent(&keyframe);5932+ let mut active_group = consumer.recv_group().await.unwrap().unwrap();5933+ assert!(active_group.read_frame().await.unwrap().is_some());5934+5935+ publisher.publish(&config, &first_delta, false).unwrap();5936+ replay_cursor.commit_sent(&first_delta);5937+ assert!(active_group.read_frame().await.unwrap().is_some());5938+5939+ assert_eq!(5940+ apply_moq_keyframe_request(5941+ &mut publisher,5942+ &mut replay_cursor,5943+ Some(first_delta.sequence),5944+ KeyframeRequestReasonV3::Join,5945+ ),5946+ None5947+ );5948+ assert_eq!(replay_cursor.last_sequence, Some(first_delta.sequence));5949+ assert!(!replay_cursor.waiting_for_keyframe);5950+ assert_eq!(5951+ publisher.publish(&config, &next_delta, false).unwrap(),5952+ MoqGroupDecision::Published {5953+ group_id: 0,5954+ frame_id: 2,5955+ cancelled_previous: false,5956+ }5957+ );5958+ assert!(active_group.read_frame().await.unwrap().is_some());5959+ }5960+56795961 56805962 async fn upstream_moq_resync_aborts_current_group_and_waits_for_configured_idr() {56815963 let track_info = Track {@@ -5712,6 +5994,66 @@ mod tests {
57125994 );57135995 }57145996 5997+ 5998+ async fn forced_idr_recovery_is_one_slot_join_safe_and_rearms_after_ack() {5999+ let telemetry = Arc::new(MediaV3Telemetry::default());6000+ let harness = crate::source::EncoderControlTestHarness::new();6001+ let mut coordinator =6002+ ForcedIdrCoordinator::new(Some(harness.control.clone()), Arc::clone(&telemetry));6003+6004+ assert_eq!(6005+ coordinator.request(KeyframeRequestReasonV3::Join),6006+ ForcedIdrDisposition::JoinReplay6007+ );6008+ assert_eq!(harness.requested_force_keyframe_revision(), None);6009+6010+ let requested_revision = match coordinator.request(KeyframeRequestReasonV3::DecoderReset) {6011+ ForcedIdrDisposition::Requested { revision } => revision,6012+ disposition => panic!("unexpected forced-IDR disposition: {disposition:?}"),6013+ };6014+ assert_eq!(6015+ harness.requested_force_keyframe_revision(),6016+ Some(requested_revision)6017+ );6018+ assert_eq!(6019+ coordinator.request(KeyframeRequestReasonV3::TransportGap),6020+ ForcedIdrDisposition::Coalesced {6021+ revision: requested_revision6022+ }6023+ );6024+ assert_eq!(telemetry.encoder_force_requests.load(Ordering::Relaxed), 1);6025+6026+ let newer_revision = harness.control.request_force_keyframe().unwrap();6027+ harness.status.send_modify(|status| {6028+ status.requested_force_keyframe_revision = Some(newer_revision);6029+ status.acknowledged_force_keyframe_revision = Some(newer_revision);6030+ });6031+ let acknowledgement = coordinator.acknowledgements.join_next().await;6032+ coordinator.complete(acknowledgement, endpoint(1), "test");6033+ assert_eq!(coordinator.pending_revision, None);6034+ assert_eq!(6035+ telemetry6036+ .encoder_force_acknowledgements6037+ .load(Ordering::Relaxed),6038+ 16039+ );6040+ assert_eq!(telemetry.encoder_force_failures.load(Ordering::Relaxed), 0);6041+6042+ assert!(matches!(6043+ coordinator.request(KeyframeRequestReasonV3::DecoderReset),6044+ ForcedIdrDisposition::Requested { revision } if revision > newer_revision6045+ ));6046+ coordinator.abort_and_drain(endpoint(1), "test").await;6047+6048+ let fallback_telemetry = Arc::new(MediaV3Telemetry::default());6049+ let mut fallback = ForcedIdrCoordinator::new(None, fallback_telemetry);6050+ assert_eq!(6051+ fallback.request(KeyframeRequestReasonV3::DecoderReset),6052+ ForcedIdrDisposition::Unavailable6053+ );6054+ assert!(fallback.acknowledgements.is_empty());6055+ }6056+57156057 57166058 async fn upstream_moq_group_counts_envelope_bytes_before_upstream_cache_eviction() {57176059 let track_info = Track {@@ -225,6 +225,36 @@ impl EncoderControl {
225225 .with_context(|| format!("timed out waiting for force-keyframe revision {revision}"))?226226 }227227 228+ /// Wait for recovery to reach any configured IDR at or after `revision`.229+ ///230+ /// Recovery requests are deliberately coalesced with later encoder231+ /// controls. Unlike transactional adaptive updates, a newer configured IDR232+ /// still satisfies the decoder barrier established by an older request.233+ pub async fn wait_for_recovery_keyframe_acknowledged(234+ &self,235+ revision: u64,236+ timeout: Duration,237+ ) -> Result<EncoderControlStatus> {238+ let mut status = self.status.clone();239+ let wait = async {240+ loop {241+ let snapshot = *status.borrow_and_update();242+ if snapshot243+ .acknowledged_force_keyframe_revision244+ .is_some_and(|acknowledged| acknowledged >= revision)245+ {246+ return Ok(snapshot);247+ }248+ status.changed().await.context(249+ "encoder control status closed before recovery keyframe was acknowledged",250+ )?;251+ }252+ };253+ tokio::time::timeout(timeout, wait).await.with_context(|| {254+ format!("timed out waiting for recovery keyframe revision {revision}")255+ })?256+ }257+228258 /// Wait until the exact target dimensions emerge from the encoder on an229259 /// independently decodable access unit. A property write or caps event is230260 /// not an application acknowledgement.@@ -292,6 +322,10 @@ impl EncoderControlTestHarness {
292322 _desired: desired,293323 }294324 }325+326+ pub(crate) fn requested_force_keyframe_revision(&self) -> Option<u64> {327+ self._desired.borrow().force_keyframe_revision328+ }295329 }296330 297331 @@ -2206,6 +2240,35 @@ mod tests {
22062240 );22072241 }22082242 2243+ 2244+ async fn recovery_waiter_accepts_a_newer_configured_idr_revision() {2245+ let (control, _desired, status) = EncoderControl::new();2246+ let recovery_revision = control.request_force_keyframe().unwrap();2247+ let newer_revision = control.request_force_keyframe().unwrap();2248+ assert!(newer_revision > recovery_revision);2249+ status.send_modify(|status| {2250+ status.requested_force_keyframe_revision = Some(newer_revision);2251+ status.acknowledged_force_keyframe_revision = Some(newer_revision);2252+ });2253+2254+ let acknowledged = control2255+ .wait_for_recovery_keyframe_acknowledged(recovery_revision, Duration::from_millis(10))2256+ .await2257+ .unwrap();2258+ assert_eq!(2259+ acknowledged.acknowledged_force_keyframe_revision,2260+ Some(newer_revision)2261+ );2262+ assert!(2263+ control2264+ .wait_for_force_keyframe_acknowledged(recovery_revision, Duration::from_millis(10),)2265+ .await2266+ .unwrap_err()2267+ .to_string()2268+ .contains("superseded")2269+ );2270+ }2271+22092272 22102273 22112274 @@ -416,6 +416,12 @@ validate_probe_log() {
416416 grep -Fxq 'frame_sequence=monotonic' "$log_path" || return 1417417 grep -Fxq "keyframe_recovery=${keyframe_recovery}" "$log_path" || return 1418418 grep -Fxq "keyframe_request_id=${expected_request_id}" "$log_path" || return 1419+ if [[ "$expected_request_id" == not-requested ]]; then420+ grep -Fxq 'keyframe_recovery_micros=not-requested' "$log_path" || return 1421+ else422+ awk -F= '$1 == "keyframe_recovery_micros" && $2 ~ /^[0-9]+$/ { found=1 } END { exit !found }' \423+ "$log_path" || return 1424+ fi419425 awk -F= '$1 == "media_objects_dropped" && $2 ~ /^[0-9]+$/ { found=1 } END { exit !found }' \420426 "$log_path" || return 1421427 awk -F= '$1 == "media_objects_late" && $2 ~ /^[0-9]+$/ { found=1 } END { exit !found }' \@@ -623,7 +629,7 @@ printf 'probe_binary=%s\n' "$probe_bin"
623629 printf 'probe_sha256=%s\n' "$(sha256_file "$probe_bin")"624630 printf 'node_id=%s\n' "$node_id"625631 printf 'primary_frames=%s\n' "$primary_frames"626-grep -E '^(transport|control_alpn|transport_alpn|first_configured_idr|frame_sequence|group_sequence|keyframe_recovery|keyframe_request_id|keyframes|sequence_gaps|media_objects_dropped|media_objects_late|moq_catalog|moq_group_capacity|moq_cancelled_groups|moq_group_gaps|moq_unrecovered_group_gaps|moq_maximum_group_objects|moq_maximum_group_bytes|moq_historical_suffix_frames|recovery_barrier|input_ack_micros|path_mode|path_rtt_ms)=' "$primary_log"632+grep -E '^(transport|control_alpn|transport_alpn|first_configured_idr|frame_sequence|group_sequence|keyframe_recovery|keyframe_recovery_micros|keyframe_request_id|keyframes|sequence_gaps|media_objects_dropped|media_objects_late|moq_catalog|moq_group_capacity|moq_cancelled_groups|moq_group_gaps|moq_unrecovered_group_gaps|moq_maximum_group_objects|moq_maximum_group_bytes|moq_historical_suffix_frames|recovery_barrier|input_ack_micros|path_mode|path_rtt_ms)=' "$primary_log"627633 printf 'keyframe_request_correlation=%s\n' \628634 "$([[ "$media_v1" == false && "$media_v2" == false ]] && printf unique || printf not-requested)"629635 printf 'active_client_rejection=ok\n'