Evaluate feedback independently from slow encoder commits with a bounded one-active, one-latest coordinator. Preserve exact bitrate, recovery-IDR, and resolution acknowledgement diagnostics while retrying the latest desired targets.
- commit
- d2440180f2a0340fbf1f0962bd21c4c036ba13c6
- parent
- f53de5e47c6a2cadc2e3f951d89281f1ef61adef
- author
- Dami <dami.etoile@gmail.com>
| M | crates/ | 576 | ++++++++++++++++++++---------- |
@@ -260,15 +260,6 @@ impl MotionResolutionPolicy {
260260 }261261 }262262 263- fn commit_observation_from(&mut self, candidate: &Self) {264- self.motion_active = candidate.motion_active;265- self.still_since = candidate.still_since;266- self.pressure_active = candidate.pressure_active;267- self.moderate_windows = candidate.moderate_windows;268- self.clean_windows = candidate.clean_windows;269- self.last_motion_sequence = candidate.last_motion_sequence;270- }271-272263 fn downshift(273264 &mut self,274265 observed_at: Instant,@@ -826,6 +817,7 @@ impl ShadowBitrateController {
826817 )827818 }828819 820+ 829821 fn commit_observation_from(&mut self, candidate: &Self) {830822 self.next_decision_id = candidate.next_decision_id;831823 self.last_report_id = candidate.last_report_id;@@ -860,6 +852,213 @@ fn queue_ratio_at_least(depth: u8, capacity: u8, numerator: u16, denominator: u1
860852 >= u16::from(capacity).saturating_mul(numerator)861853 }862854 855+856+struct AdaptiveCommitPlan {857+ decision: AdaptiveBitrateDecisionV1,858+ telemetry: MediaV3TelemetrySnapshot,859+ resolution_decision: Option<MotionResolutionDecision>,860+ force_keyframe: bool,861+}862+863+864+struct AdaptiveCommitAcknowledgement {865+ plan: AdaptiveCommitPlan,866+ generation_ended: bool,867+ recovery_acknowledged: Option<bool>,868+ resolution_applied: Option<bool>,869+ applied_bitrate_kbps: Option<u32>,870+ applied_resolution: Option<VideoDimensions>,871+}872+873+impl AdaptiveCommitAcknowledgement {874+ fn immediate(plan: AdaptiveCommitPlan) -> Self {875+ Self {876+ plan,877+ generation_ended: false,878+ recovery_acknowledged: None,879+ resolution_applied: None,880+ applied_bitrate_kbps: None,881+ applied_resolution: None,882+ }883+ }884+}885+886+/// Bounded adaptive-control actuator.887+///888+/// Feedback evaluation remains synchronous and deterministic, while this889+/// coordinator owns the slower encoder acknowledgements. At most one commit890+/// task and one latest coalesced plan exist. A queued plan carries the latest891+/// desired bitrate and resolution even when its feedback decision is a hold,892+/// so a slow or failed predecessor cannot strand an uncommitted target.893+struct AdaptiveCommitCoordinator {894+ sessions: Arc<SessionRegistry>,895+ remote: EndpointId,896+ session_id: u64,897+ actuation_enabled: bool,898+ committed_bitrate_kbps: u32,899+ committed_resolution: Option<VideoDimensions>,900+ pending: bool,901+ queued: Option<AdaptiveCommitPlan>,902+ acknowledgements: tokio::task::JoinSet<AdaptiveCommitAcknowledgement>,903+}904+905+impl AdaptiveCommitCoordinator {906+ fn new(907+ sessions: Arc<SessionRegistry>,908+ remote: EndpointId,909+ session_id: u64,910+ actuation_enabled: bool,911+ committed_bitrate_kbps: u32,912+ committed_resolution: Option<VideoDimensions>,913+ ) -> Self {914+ Self {915+ sessions,916+ remote,917+ session_id,918+ actuation_enabled,919+ committed_bitrate_kbps,920+ committed_resolution,921+ pending: false,922+ queued: None,923+ acknowledgements: tokio::task::JoinSet::new(),924+ }925+ }926+927+ fn has_pending(&self) -> bool {928+ self.pending929+ }930+931+ fn submit(&mut self, plan: AdaptiveCommitPlan) -> Option<AdaptiveCommitAcknowledgement> {932+ if self.pending {933+ self.queued = Some(plan);934+ return None;935+ }936+ self.start(plan)937+ }938+939+ fn start_queued(&mut self) -> Option<AdaptiveCommitAcknowledgement> {940+ let plan = self.queued.take()?;941+ self.start(plan)942+ }943+944+ fn start(&mut self, plan: AdaptiveCommitPlan) -> Option<AdaptiveCommitAcknowledgement> {945+ if !self.actuation_enabled {946+ return Some(AdaptiveCommitAcknowledgement::immediate(plan));947+ }948+949+ let desired_resolution = plan.resolution_decision.map(|change| change.target);950+ let changes_resolution =951+ desired_resolution.is_some() && desired_resolution != self.committed_resolution;952+ let bitrate_proposal = if plan.decision.target_kbps != self.committed_bitrate_kbps {953+ match self.sessions.propose_adaptive_encoder_update(954+ self.remote,955+ self.session_id,956+ plan.decision.target_kbps,957+ plan.force_keyframe && !changes_resolution,958+ ) {959+ Ok(Some(proposal)) => Some(proposal),960+ Ok(None) => {961+ warn!(962+ remote = %self.remote,963+ session_id = self.session_id,964+ "active adaptive session lost encoder control; retaining committed bitrate"965+ );966+ None967+ }968+ Err(error) => {969+ warn!(970+ remote = %self.remote,971+ session_id = self.session_id,972+ %error,973+ "adaptive encoder proposal failed; retaining committed bitrate"974+ );975+ None976+ }977+ }978+ } else {979+ None980+ };981+982+ let resolution_proposal = if changes_resolution {983+ let target = desired_resolution.expect("resolution target was checked as present");984+ match self985+ .sessions986+ .propose_resolution_update(self.remote, self.session_id, target)987+ {988+ Ok(Some(proposal)) => Some(proposal),989+ Ok(None) => {990+ warn!(991+ remote = %self.remote,992+ session_id = self.session_id,993+ "active adaptive session lost encoder control; retaining committed resolution"994+ );995+ None996+ }997+ Err(error) => {998+ warn!(999+ remote = %self.remote,1000+ session_id = self.session_id,1001+ %error,1002+ "adaptive resolution proposal failed; retaining committed resolution"1003+ );1004+ None1005+ }1006+ }1007+ } else {1008+ None1009+ };1010+1011+ if bitrate_proposal.is_none() && resolution_proposal.is_none() {1012+ return Some(AdaptiveCommitAcknowledgement::immediate(plan));1013+ }1014+1015+ self.pending = true;1016+ let sessions = Arc::clone(&self.sessions);1017+ let remote = self.remote;1018+ let session_id = self.session_id;1019+ self.acknowledgements.spawn(async move {1020+ run_adaptive_commit(1021+ sessions,1022+ remote,1023+ session_id,1024+ plan,1025+ bitrate_proposal,1026+ resolution_proposal,1027+ )1028+ .await1029+ });1030+ None1031+ }1032+1033+ fn complete(1034+ &mut self,1035+ result: Option<std::result::Result<AdaptiveCommitAcknowledgement, tokio::task::JoinError>>,1036+ ) -> Result<AdaptiveCommitAcknowledgement> {1037+ ensure!(1038+ self.pending,1039+ "adaptive commit completed without pending work"1040+ );1041+ self.pending = false;1042+ let acknowledgement = result1043+ .context("adaptive commit task ended without a result")?1044+ .context("adaptive commit task failed")?;1045+ if let Some(applied) = acknowledgement.applied_bitrate_kbps {1046+ self.committed_bitrate_kbps = applied;1047+ }1048+ if let Some(applied) = acknowledgement.applied_resolution {1049+ self.committed_resolution = Some(applied);1050+ }1051+ Ok(acknowledgement)1052+ }1053+1054+ async fn abort_and_drain(&mut self) {1055+ self.pending = false;1056+ self.queued = None;1057+ self.acknowledgements.abort_all();1058+ while self.acknowledgements.join_next().await.is_some() {}1059+ }1060+}1061+8631062 pub(super) async fn serve_media_feedback(8641063 connection: Connection,8651064 config: &HostConfig,@@ -938,9 +1137,37 @@ pub(super) async fn serve_media_feedback(
9381137 } else {9391138 None9401139 };1140+ let initial_resolution = resolution_controller1141+ .as_ref()1142+ .map(MotionResolutionPolicy::target);1143+ let mut commit_coordinator = AdaptiveCommitCoordinator::new(1144+ Arc::clone(sessions),1145+ remote,1146+ lease.session_id,1147+ encoder_actuation_available,1148+ ceiling_kbps,1149+ initial_resolution,1150+ );9411151 let mut feedback_cursor = FeedbackIngressCursor::default();1152+ let mut ready_acknowledgement: Option<AdaptiveCommitAcknowledgement> = None;9421153 9431154 let result: Result<()> = loop {1155+ if let Some(acknowledgement) = ready_acknowledgement.take() {1156+ if acknowledgement.generation_ended {1157+ break Ok(());1158+ }1159+ write_adaptive_acknowledgement(1160+ &mut send,1161+ remote,1162+ lease.session_id,1163+ encoder_actuation_available,1164+ &acknowledgement,1165+ )1166+ .await?;1167+ ready_acknowledgement = commit_coordinator.start_queued();1168+ continue;1169+ }1170+9441171 let session_changed = sessions.session_changed.notified();9451172 tokio::pin!(session_changed);9461173 session_changed.as_mut().enable();@@ -963,6 +1190,12 @@ pub(super) async fn serve_media_feedback(
9631190 }9641191 }9651192 }1193+ commit_result = commit_coordinator.acknowledgements.join_next(),1194+ if commit_coordinator.has_pending() =>1195+ {1196+ let acknowledgement = commit_coordinator.complete(commit_result)?;1197+ ready_acknowledgement = Some(acknowledgement);1198+ }9661199 changed = feedback_receiver.changed(), if receiver_open => {9671200 if changed.is_err() {9681201 receiver_open = false;@@ -983,198 +1216,25 @@ pub(super) async fn serve_media_feedback(
9831216 let observed_at = Instant::now();9841217 let mut candidate = controller.clone();9851218 let evaluation = candidate.evaluate(&report, telemetry, observed_at)?;986- let mut decision = evaluation.decision;1219+ let decision = evaluation.decision;9871220 let mut resolution_candidate = resolution_controller.clone();9881221 let resolution_decision = resolution_candidate9891222 .as_mut()9901223 .map(|policy| policy.observe_window(&report, &evaluation, observed_at));991- let changes_bitrate = decision.state != AdaptiveBitrateStateV1::Hold;992- let changes_resolution = resolution_decision.is_some_and(|change| change.changed);993- let can_actuate_bitrate = changes_bitrate && encoder_actuation_available;994- let can_actuate_resolution = changes_resolution && encoder_actuation_available;995- let mut recovery_acknowledged = None;996- let bitrate_proposal = if can_actuate_bitrate {997- let force_keyframe =998- adaptive_recovery_keyframe_required(&report, &decision)999- && !changes_resolution;1000- match sessions.propose_adaptive_encoder_update(1001- remote,1002- lease.session_id,1003- decision.target_kbps,1224+ let force_keyframe =1225+ adaptive_recovery_keyframe_required(&report, &decision);1226+ // Evaluation state advances at feedback cadence. The actuator1227+ // independently retains the exact committed target and retries1228+ // the latest desired state after a slow or failed predecessor.1229+ controller = candidate;1230+ resolution_controller = resolution_candidate;1231+ ready_acknowledgement =1232+ commit_coordinator.submit(AdaptiveCommitPlan {1233+ decision,1234+ telemetry,1235+ resolution_decision,10041236 force_keyframe,1005- ) {1006- Ok(Some(proposal)) => Some(proposal),1007- Ok(None) => {1008- warn!(1009- %remote,1010- session_id = lease.session_id,1011- "active adaptive session lost encoder control; retaining committed bitrate"1012- );1013- None1014- }1015- Err(error) => {1016- warn!(1017- %remote,1018- session_id = lease.session_id,1019- %error,1020- "adaptive encoder proposal failed; retaining committed bitrate"1021- );1022- None1023- }1024- }1025- } else {1026- None1027- };1028- let resolution_proposal = if can_actuate_resolution {1029- let target = resolution_decision1030- .expect("resolution change is required for actuation")1031- .target;1032- match sessions.propose_resolution_update(remote, lease.session_id, target) {1033- Ok(Some(proposal)) => Some(proposal),1034- Ok(None) => {1035- warn!(1036- %remote,1037- session_id = lease.session_id,1038- "active adaptive session lost encoder control; retaining committed resolution"1039- );1040- None1041- }1042- Err(error) => {1043- warn!(1044- %remote,1045- session_id = lease.session_id,1046- %error,1047- "adaptive resolution proposal failed; retaining committed resolution"1048- );1049- None1050- }1051- }1052- } else {1053- None1054- };1055-1056- if let Some(proposal) = bitrate_proposal {1057- match commit_adaptive_encoder_proposal(1058- sessions,1059- remote,1060- lease.session_id,1061- &proposal,1062- )1063- .await1064- {1065- AdaptiveEncoderCommit::GenerationEnded => break Ok(()),1066- AdaptiveEncoderCommit::NotApplied(error) => {1067- warn!(1068- %remote,1069- session_id = lease.session_id,1070- %error,1071- "adaptive encoder application failed; retaining committed bitrate"1072- );1073- }1074- AdaptiveEncoderCommit::Applied => {1075- decision.applied = true;1076- controller = candidate.clone();1077- if proposal.force_keyframe_revision.is_some() {1078- match await_adaptive_recovery_keyframe(1079- sessions,1080- remote,1081- lease.session_id,1082- &proposal,1083- )1084- .await1085- {1086- None => break Ok(()),1087- Some(Ok(())) => recovery_acknowledged = Some(true),1088- Some(Err(error)) => {1089- recovery_acknowledged = Some(false);1090- warn!(1091- %remote,1092- session_id = lease.session_id,1093- %error,1094- "adaptive bitrate applied but forced-IDR recovery was not acknowledged"1095- );1096- }1097- }1098- }1099- }1100- }1101- }1102- if can_actuate_bitrate {1103- if !decision.applied {1104- controller.commit_observation_from(&candidate);1105- }1106- } else {1107- // External gst-launch and sources without EncoderControl1108- // remain intentionally shadow-only. Hold decisions do not1109- // need encoder actuation.1110- controller = candidate;1111- }1112-1113- let mut resolution_applied = None;1114- if let Some(proposal) = resolution_proposal {1115- match commit_resolution_encoder_proposal(1116- sessions,1117- remote,1118- lease.session_id,1119- &proposal,1120- )1121- .await1122- {1123- AdaptiveEncoderCommit::GenerationEnded => break Ok(()),1124- AdaptiveEncoderCommit::NotApplied(error) => {1125- resolution_applied = Some(false);1126- warn!(1127- %remote,1128- session_id = lease.session_id,1129- target_width = proposal.target.width,1130- target_height = proposal.target.height,1131- %error,1132- "adaptive resolution application failed; retaining committed resolution"1133- );1134- }1135- AdaptiveEncoderCommit::Applied => {1136- resolution_applied = Some(true);1137- resolution_controller = resolution_candidate.clone();1138- }1139- }1140- }1141- if can_actuate_resolution && resolution_applied != Some(true) {1142- if let (Some(committed), Some(candidate)) =1143- (resolution_controller.as_mut(), resolution_candidate.as_ref())1144- {1145- committed.commit_observation_from(candidate);1146- }1147- } else if !changes_resolution {1148- resolution_controller = resolution_candidate;1149- }1150- tokio::time::timeout(1151- FEEDBACK_WRITE_TIMEOUT,1152- write_adaptive_bitrate_decision_v1(&mut send, &decision),1153- )1154- .await1155- .context("timed out writing adaptive bitrate decision")??;1156- debug!(1157- %remote,1158- session_id = lease.session_id,1159- report_id = decision.report_id,1160- decision_id = decision.decision_id,1161- target_kbps = decision.target_kbps,1162- ?decision.state,1163- reasons = decision.reasons.bits(),1164- applied = decision.applied,1165- ?recovery_acknowledged,1166- resolution_width = resolution_decision.map(|change| change.target.width),1167- resolution_height = resolution_decision.map(|change| change.target.height),1168- resolution_trigger = ?resolution_decision.map(|change| change.trigger),1169- ?resolution_applied,1170- path_rtt_micros = telemetry.selected_path_rtt_micros,1171- path_lost_packets = telemetry.selected_path_lost_packets,1172- path_congestion_events = telemetry.selected_path_congestion_events,1173- scheduler_cancellations = telemetry.scheduler_cancellations,1174- send_failures = telemetry.send_failures,1175- mode = if encoder_actuation_available { "active" } else { "shadow" },1176- "adaptive bitrate decision"1177- );1237+ });11781238 }11791239 _ = &mut session_changed => {11801240 if !sessions.is_active(remote, lease.session_id) {@@ -1186,7 +1246,11 @@ pub(super) async fn serve_media_feedback(
11861246 break Ok(());11871247 }11881248 }1189- if reader_finished && !receiver_open {1249+ if reader_finished1250+ && !receiver_open1251+ && !commit_coordinator.has_pending()1252+ && commit_coordinator.queued.is_none()1253+ {11901254 break Ok(());11911255 }11921256 };@@ -1195,12 +1259,130 @@ pub(super) async fn serve_media_feedback(
11951259 reader_task.abort();11961260 let _ = reader_task.await;11971261 }1262+ commit_coordinator.abort_and_drain().await;11981263 let _ = send.finish();11991264 drop(lease);12001265 info!(%remote, "media feedback client released");12011266 result12021267 }12031268 1269+async fn write_adaptive_acknowledgement(1270+ send: &mut SendStream,1271+ remote: EndpointId,1272+ session_id: u64,1273+ encoder_actuation_available: bool,1274+ acknowledgement: &AdaptiveCommitAcknowledgement,1275+) -> Result<()> {1276+ let decision = acknowledgement.plan.decision;1277+ let telemetry = acknowledgement.plan.telemetry;1278+ let resolution_decision = acknowledgement.plan.resolution_decision;1279+ tokio::time::timeout(1280+ FEEDBACK_WRITE_TIMEOUT,1281+ write_adaptive_bitrate_decision_v1(send, &decision),1282+ )1283+ .await1284+ .context("timed out writing adaptive bitrate decision")??;1285+ debug!(1286+ %remote,1287+ session_id,1288+ report_id = decision.report_id,1289+ decision_id = decision.decision_id,1290+ target_kbps = decision.target_kbps,1291+ ?decision.state,1292+ reasons = decision.reasons.bits(),1293+ applied = decision.applied,1294+ recovery_acknowledged = ?acknowledgement.recovery_acknowledged,1295+ resolution_width = resolution_decision.map(|change| change.target.width),1296+ resolution_height = resolution_decision.map(|change| change.target.height),1297+ resolution_trigger = ?resolution_decision.map(|change| change.trigger),1298+ resolution_applied = ?acknowledgement.resolution_applied,1299+ path_rtt_micros = telemetry.selected_path_rtt_micros,1300+ path_lost_packets = telemetry.selected_path_lost_packets,1301+ path_congestion_events = telemetry.selected_path_congestion_events,1302+ scheduler_cancellations = telemetry.scheduler_cancellations,1303+ send_failures = telemetry.send_failures,1304+ mode = if encoder_actuation_available { "active" } else { "shadow" },1305+ "adaptive bitrate decision"1306+ );1307+ Ok(())1308+}1309+1310+async fn run_adaptive_commit(1311+ sessions: Arc<SessionRegistry>,1312+ remote: EndpointId,1313+ session_id: u64,1314+ plan: AdaptiveCommitPlan,1315+ bitrate_proposal: Option<AdaptiveEncoderProposal>,1316+ resolution_proposal: Option<ResolutionEncoderProposal>,1317+) -> AdaptiveCommitAcknowledgement {1318+ let mut acknowledgement = AdaptiveCommitAcknowledgement::immediate(plan);1319+ if let Some(proposal) = bitrate_proposal {1320+ match commit_adaptive_encoder_proposal(&sessions, remote, session_id, &proposal).await {1321+ AdaptiveEncoderCommit::GenerationEnded => {1322+ acknowledgement.generation_ended = true;1323+ return acknowledgement;1324+ }1325+ AdaptiveEncoderCommit::NotApplied(error) => {1326+ warn!(1327+ %remote,1328+ session_id,1329+ %error,1330+ "adaptive encoder application failed; retaining committed bitrate"1331+ );1332+ }1333+ AdaptiveEncoderCommit::Applied => {1334+ acknowledgement.plan.decision.applied = true;1335+ acknowledgement.applied_bitrate_kbps = Some(proposal.target_kbps);1336+ if proposal.force_keyframe_revision.is_some() {1337+ match await_adaptive_recovery_keyframe(&sessions, remote, session_id, &proposal)1338+ .await1339+ {1340+ None => {1341+ acknowledgement.generation_ended = true;1342+ return acknowledgement;1343+ }1344+ Some(Ok(())) => acknowledgement.recovery_acknowledged = Some(true),1345+ Some(Err(error)) => {1346+ acknowledgement.recovery_acknowledged = Some(false);1347+ warn!(1348+ %remote,1349+ session_id,1350+ %error,1351+ "adaptive bitrate applied but forced-IDR recovery was not acknowledged"1352+ );1353+ }1354+ }1355+ }1356+ }1357+ }1358+ }1359+1360+ if let Some(proposal) = resolution_proposal {1361+ match commit_resolution_encoder_proposal(&sessions, remote, session_id, &proposal).await {1362+ AdaptiveEncoderCommit::GenerationEnded => {1363+ acknowledgement.generation_ended = true;1364+ return acknowledgement;1365+ }1366+ AdaptiveEncoderCommit::NotApplied(error) => {1367+ acknowledgement.resolution_applied = Some(false);1368+ warn!(1369+ %remote,1370+ session_id,1371+ target_width = proposal.target.width,1372+ target_height = proposal.target.height,1373+ %error,1374+ "adaptive resolution application failed; retaining committed resolution"1375+ );1376+ }1377+ AdaptiveEncoderCommit::Applied => {1378+ acknowledgement.resolution_applied = Some(true);1379+ acknowledgement.applied_resolution = Some(proposal.target);1380+ }1381+ }1382+ }1383+ acknowledgement1384+}1385+12041386 async fn forward_media_feedback_reports<R>(12051387 mut reader: R,12061388 sender: tokio::sync::watch::Sender<CumulativeFeedbackIngress>,