fix(voip): prevent stale audio-to-video upgrades

Cargo.toml

@@ -221,7 +221,7 @@ hmac = { workspace = true }
 libc = "0.2"
 metrics-exporter-prometheus = "0.18"
 sha2 = { workspace = true }
-tokio = { workspace = true, features = ["full"] }
+tokio = { workspace = true, features = ["full", "test-util"] }
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 uuid = { workspace = true, features = ["v4"] }
 wacore = { workspace = true, features = ["test-util"] }

examples/voip-cli/src/main.rs

@@ -55,7 +55,7 @@ use whatsapp_rust::voip::audio::{WaOpusDecoder, WaOpusEncoder};
 use whatsapp_rust::voip::session::{MediaPipeline, MediaPipelineParams};
 #[cfg(feature = "voip-opus")]
 use whatsapp_rust::voip::{AudioCodec, EncodedAudioFrame};
-use whatsapp_rust::voip::{AudioFormat, CallHandle, VideoState};
+use whatsapp_rust::voip::{AudioFormat, CallHandle, VideoState, VideoUpgradeToken};
 
 mod video;
 use video::VideoOpts;
@@ -977,7 +977,7 @@ struct CallState {
 #[derive(Clone, Copy, PartialEq)]
 enum VideoUi {
     /// The peer asked for video (`UpgradeRequestV2`); `v` accepts it.
-    PendingPeerRequest,
+    PendingPeerRequest(VideoUpgradeToken),
     /// Our video plane is up; `v` downgrades.
     Active,
 }
@@ -1080,12 +1080,14 @@ impl EventHandler for CallObserver {
                     let audio = self.audio;
                     // Only answer with video when we're allowed to AND the offer is actually a video
                     // call; advertising `<video>` on an audio offer would be wrong.
-                    let video = self.video && *is_video;
+                    let initial_video = self.video && *is_video;
+                    let auto_video = self.video;
                     let state = self.state.clone();
                     let cid = call_id.clone();
                     begin_call_startup(&state, &cid);
                     tokio::spawn(async move {
-                        if let Err(e) = respond_to_offer(&client, &call, accept, video, audio).await
+                        if let Err(e) =
+                            respond_to_offer(&client, &call, accept, initial_video, audio).await
                         {
                             error!("call signaling failed: {e}");
                             complete_call_startup(&state, &cid);
@@ -1095,7 +1097,9 @@ impl EventHandler for CallObserver {
                             complete_call_startup(&state, &cid);
                             return;
                         }
-                        match start_media(&client, &call, video, audio, &state).await {
+                        match start_media(&client, &call, initial_video, auto_video, audio, &state)
+                            .await
+                        {
                             Ok(handle) => register_handle(&state, cid.clone(), handle).await,
                             Err(e) => {
                                 let peer_ended = peer_terminated_during_startup(&state, &cid);
@@ -1141,15 +1145,16 @@ impl EventHandler for CallObserver {
 async fn start_media(
     client: &Arc<Client>,
     call: &IncomingCall,
-    video: bool,
+    initial_video: bool,
+    auto_video: bool,
     audio: AudioMode,
     state: &Arc<Mutex<CallState>>,
 ) -> Result<Arc<CallHandle>> {
     let mic = spawn_mic()?;
     let speaker = spawn_speaker()?;
     let event_speaker = speaker.clone();
     info!("🔌 connecting media via client.voip().accept(..)…");
-    let video_endpoints = if video {
+    let video_endpoints = if initial_video {
         let opts = VideoOpts::from_env().await?;
         let cid = call.action.call_id();
         Some((
@@ -1162,7 +1167,7 @@ async fn start_media(
     if peer_terminated_during_startup(state, call.action.call_id()) {
         bail!("peer ended the call during media preparation");
     }
-    send_final_accept(client, call, video, audio).await?;
+    send_final_accept(client, call, initial_video, audio).await?;
     let voip = client.voip();
     let mut builder = match audio {
         #[cfg(feature = "voip-mlow")]
@@ -1185,13 +1190,13 @@ async fn start_media(
         "🎙  {} media flow live for call {} — speak into the mic.{}",
         audio.name(),
         handle.call_id(),
-        if video { " 🎥 video on." } else { "" }
+        if initial_video { " 🎥 video on." } else { "" }
     );
     let handle = Arc::new(handle);
-    if video {
+    if initial_video {
         mark_video(state, handle.call_id(), Some(VideoUi::Active));
     }
-    spawn_call_event_listener(handle.clone(), event_speaker, video, state.clone());
+    spawn_call_event_listener(handle.clone(), event_speaker, auto_video, state.clone());
     Ok(handle)
 }
 
@@ -1258,6 +1263,29 @@ fn mark_video(state: &Arc<Mutex<CallState>>, call_id: &str, ui: Option<VideoUi>)
     }
 }
 
+fn clear_pending_peer_video(state: &Arc<Mutex<CallState>>, call_id: &str) {
+    let mut st = lock_call_state(state);
+    if matches!(
+        st.video_ui.get(call_id),
+        Some(VideoUi::PendingPeerRequest(_))
+    ) {
+        st.video_ui.remove(call_id);
+    }
+}
+
+fn activate_peer_video_request(
+    state: &Arc<Mutex<CallState>>,
+    call_id: &str,
+    request: VideoUpgradeToken,
+) {
+    let mut st = lock_call_state(state);
+    if let Some(ui) = st.video_ui.get_mut(call_id)
+        && *ui == VideoUi::PendingPeerRequest(request)
+    {
+        *ui = VideoUi::Active;
+    }
+}
+
 /// Surface call diagnostics and drive the video upgrade handshake.
 fn spawn_call_event_listener(
     handle: Arc<CallHandle>,
@@ -1380,25 +1408,51 @@ fn spawn_call_event_listener(
                         "🎥 relay-send backpressure: dropped {video_access_units} complete video AUs / {packets} packets"
                     );
                 }
-                CallEvent::VideoStateChanged { state: vs, .. } => match vs {
+                CallEvent::VideoStateChanged {
+                    state: vs,
+                    upgrade_token,
+                    ..
+                } => match vs {
                     VideoState::UpgradeRequest | VideoState::UpgradeRequestV2 => {
+                        let Some(request) = upgrade_token else {
+                            info!("🎥 concurrent video upgrade resolved automatically");
+                            continue;
+                        };
+                        mark_video(
+                            &state,
+                            handle.call_id(),
+                            Some(VideoUi::PendingPeerRequest(request)),
+                        );
                         if auto_video {
                             info!("🎥 peer asked for video — auto-accepting (--video)");
-                            if let Err(e) = accept_peer_video(&handle).await {
-                                warn!("accepting peer video failed: {e}");
-                            } else {
-                                mark_video(&state, handle.call_id(), Some(VideoUi::Active));
-                            }
+                            let handle = handle.clone();
+                            let state = state.clone();
+                            tokio::spawn(async move {
+                                if let Err(e) = accept_peer_video(&handle, request).await {
+                                    warn!("accepting peer video failed: {e}");
+                                } else {
+                                    activate_peer_video_request(&state, handle.call_id(), request);
+                                }
+                            });
                         } else {
                             info!("🎥 peer asked for video — press `v` + Enter to accept");
-                            mark_video(&state, handle.call_id(), Some(VideoUi::PendingPeerRequest));
                         }
                     }
                     VideoState::UpgradeAccept | VideoState::Enabled => {
                         info!("🎥 peer video {vs:?} — inbound video should start flowing");
                     }
-                    VideoState::Stopped | VideoState::Disabled => {
+                    VideoState::Stopped => {
                         info!("🎥 peer stopped its video");
+                        clear_pending_peer_video(&state, handle.call_id());
+                    }
+                    VideoState::Disabled
+                    | VideoState::UpgradeReject
+                    | VideoState::UpgradeRejectByTimeout
+                    | VideoState::UpgradeCancel
+                    | VideoState::UpgradeCancelByTimeout
+                    | VideoState::Error => {
+                        info!("🎥 video upgrade ended ({vs:?})");
+                        mark_video(&state, handle.call_id(), None);
                     }
                     other => info!("🎥 peer video state: {other:?}"),
                 },
@@ -1410,12 +1464,12 @@ fn spawn_call_event_listener(
 }
 
 /// Fresh ffmpeg/ffplay endpoints for a mid-call upgrade/accept on `handle`.
-async fn accept_peer_video(handle: &CallHandle) -> Result<()> {
+async fn accept_peer_video(handle: &CallHandle, request: VideoUpgradeToken) -> Result<()> {
     let opts = VideoOpts::from_env().await?;
     let src = video::spawn_video_source(&opts).await?;
     let sink = video::spawn_video_sink(&opts, handle.call_id()).await?;
     handle
-        .accept_video(src, sink)
+        .accept_video(request, src, sink)
         .await
         .map_err(|e| anyhow!("accept_video: {e}"))
 }
@@ -1580,12 +1634,12 @@ fn spawn_stdin_ui(client: Arc<Client>, state: Arc<Mutex<CallState>>) {
                             mark_video(&state, &cid, None);
                         }
                     }
-                    Some(VideoUi::PendingPeerRequest) => {
+                    Some(VideoUi::PendingPeerRequest(request)) => {
                         info!("🎥 accepting the peer's video request");
-                        if let Err(e) = accept_peer_video(&handle).await {
+                        if let Err(e) = accept_peer_video(&handle, request).await {
                             warn!("accept_video failed: {e}");
                         } else {
-                            mark_video(&state, &cid, Some(VideoUi::Active));
+                            activate_peer_video_request(&state, &cid, request);
                         }
                     }
                     None => {

src/client/voip.rs

@@ -68,6 +68,10 @@ pub enum CallError {
     #[cfg(feature = "voip-runtime")]
     #[error("media offer error: {0}")]
     Media(&'static str),
+    /// The peer cancelled or replaced the upgrade before its video source became ready.
+    #[cfg(feature = "voip-runtime")]
+    #[error("video upgrade request is no longer current")]
+    VideoUpgradeExpired,
     /// `call(peer)` resolved zero devices for the peer (nothing to address an offer to).
     #[cfg(feature = "voip-runtime")]
     #[error("peer has no resolvable devices")]