fix(dialog): end the INVITE of an aborted do_invite_async task

src/dialog/invitation.rs

@@ -183,15 +183,17 @@ impl Drop for DialogGuard {
     }
 }
 
-pub(super) struct DialogGuardForUnconfirmed<'a> {
-    pub dialog_layer_inner: &'a DialogLayerInnerRef,
-    pub id: &'a DialogId,
+pub(super) struct DialogGuardForUnconfirmed {
+    pub dialog_layer_inner: DialogLayerInnerRef,
+    pub id: DialogId,
     invite_tx: Option<Transaction>,
+    /// Whether the owning future started processing the INVITE.
+    started: bool,
     /// The INVITE's dialog, `None` once `process_invite` returned.
     dialog: Option<InviteDialog>,
 }
 
-impl<'a> Drop for DialogGuardForUnconfirmed<'a> {
+impl Drop for DialogGuardForUnconfirmed {
     fn drop(&mut self) {
         let client_dialog = match self.dialog_layer_inner.dialogs.remove(&self.id.to_string()) {
             Some(Dialog::Invite(client_dialog)) => client_dialog,
@@ -205,6 +207,15 @@ impl<'a> Drop for DialogGuardForUnconfirmed<'a> {
             },
         };
 
+        // A task dropped before its first poll never sent an INVITE.
+        if !self.started {
+            let _ = client_dialog.inner.transition(DialogState::Terminated(
+                client_dialog.id(),
+                TerminatedReason::UacCancel,
+            ));
+            return;
+        }
+
         match client_dialog.state() {
             DialogState::Terminated(_, _) => {}
             state @ (DialogState::Calling(_)
@@ -240,64 +251,17 @@ impl<'a> Drop for DialogGuardForUnconfirmed<'a> {
                             return;
                         }
                     }
-                    let started = crate::platform::Instant::now();
                     let final_window = invite_tx.endpoint_inner.option.t1x64;
-                    let mut timeout =
-                        core::pin::pin!(crate::platform::sleep(core::time::Duration::from_secs(2)));
                     invite_tx.stop_retransmissions();
 
-                    let mut cancel_done = false;
-                    let mut final_response = None;
-                    // Boxed so it can be dropped before the wait below.
-                    let mut cancel = Box::pin(client_dialog.send_cancel());
-
-                    use crate::platform::select::Which3;
-                    loop {
-                        // after cancel completes, use pending as placeholder (3-arm race with
-                        // unchanged types)
-                        let cancel_sel = if cancel_done {
-                            futures::future::Either::Left(core::future::pending::<
-                                core::result::Result<(), crate::Error>,
-                            >())
-                        } else {
-                            futures::future::Either::Right(&mut cancel)
-                        };
-                        let mut cancel_sel = core::pin::pin!(cancel_sel);
-                        let mut recv_f = core::pin::pin!(invite_tx.receive());
-                        match crate::platform::select::select3(
-                            &mut timeout,
-                            &mut cancel_sel,
-                            &mut recv_f,
-                        )
-                        .await
-                        {
-                            Which3::A(()) => break,
-                            Which3::B(result) => match result {
-                                Ok(()) => cancel_done = true,
-                                Err(e) => {
-                                    warn!(id = %client_dialog.id(), error = %e, "dialog cancel failed");
-                                    break;
-                                }
-                            },
-                            Which3::C(msg) => match msg {
-                                Some(SipMessage::Response(resp))
-                                    if resp.status_code.kind() != StatusCodeKind::Provisional =>
-                                {
-                                    debug!(
-                                        id = %client_dialog.id(),
-                                        status = %resp.status_code,
-                                        "received final response"
-                                    );
-                                    final_response = Some(resp);
-                                    break;
-                                }
-                                Some(_) => {}
-                                None => break,
-                            },
+                    // CANCEL is a separate transaction. Keep driving it to its
+                    // own response or timeout even if the INVITE finishes first.
+                    let cancel_dialog = client_dialog.clone();
+                    let _handle = crate::platform::spawn(async move {
+                        if let Err(e) = cancel_dialog.send_cancel().await {
+                            warn!(id = %cancel_dialog.id(), error = %e, "dialog cancel failed");
                         }
-                    }
-
-                    drop(cancel);
+                    });
                     if !calling {
                         let _ = client_dialog.inner.transition(DialogState::Terminated(
                             client_dialog.id(),
@@ -310,17 +274,7 @@ impl<'a> Drop for DialogGuardForUnconfirmed<'a> {
                     // INVITE transaction until its final response (up to
                     // 64*T1) so a 2xx is ACKed, then end that session with a
                     // BYE (RFC 3261 §9.1, §15).
-                    if final_response.is_none() {
-                        let elapsed = crate::platform::Instant::now()
-                            .checked_duration_since(started)
-                            .unwrap_or_default();
-                        final_response = wait_response(
-                            &mut invite_tx,
-                            final_window.saturating_sub(elapsed),
-                            true,
-                        )
-                        .await;
-                    }
+                    let final_response = wait_response(&mut invite_tx, final_window, true).await;
                     // Closing the command channel terminates the dialog loop.
                     drop(invite_tx);
                     bye_if_2xx(&client_dialog, final_response).await;
@@ -692,9 +646,10 @@ impl DialogLayer {
             .insert(id.to_string(), Dialog::Invite(dialog.clone()));
 
         let mut guard = DialogGuardForUnconfirmed {
-            dialog_layer_inner: &self.inner,
-            id: &id,
+            dialog_layer_inner: self.inner.clone(),
+            id: id.clone(),
             invite_tx: Some(tx),
+            started: true,
             dialog: Some(dialog.clone()),
         };
 
@@ -776,7 +731,7 @@ impl DialogLayer {
         opt: InviteOption,
         state_sender: DialogStateSender,
     ) -> Result<(InviteDialog, crate::platform::JoinHandle<InviteAsyncResult>)> {
-        let (dialog, mut tx) = self.create_client_invite_dialog(opt, state_sender)?;
+        let (dialog, tx) = self.create_client_invite_dialog(opt, state_sender)?;
         let id0 = dialog.id();
 
         // 1) register early key (so in-dialog requests can be matched)
@@ -787,9 +742,29 @@ impl DialogLayer {
         let inner = self.inner.clone();
         let dialog_clone = dialog.clone();
 
+        // Own the guard before spawning so an abort before the first poll
+        // also removes the early registry entry.
+        let mut guard = DialogGuardForUnconfirmed {
+            dialog_layer_inner: inner.clone(),
+            id: id0.clone(),
+            invite_tx: Some(tx),
+            started: false,
+            dialog: Some(dialog_clone.clone()),
+        };
+
         // 2) run invite in background, keep registry updated like do_invite()
         let handle = crate::platform::spawn_with_result(async move {
-            let r = dialog_clone.process_invite(&mut tx).boxed().await;
+            guard.started = true;
+            let invite_tx = guard
+                .invite_tx
+                .as_mut()
+                .expect("transaction should be available");
+            let r = dialog_clone.process_invite(invite_tx).boxed().await;
+            guard.dialog = None;
+            let mut tx = guard
+                .invite_tx
+                .take()
+                .expect("transaction should be available");
 
             // remove early key
             inner.dialogs.remove(&id0.to_string());

src/dialog/tests/test_cancel_2xx_race.rs

@@ -455,6 +455,68 @@ async fn test_removed_cancel_answered_487_sends_no_bye() -> crate::Result<()> {
     run_cancel_answered_487(100, true).await
 }
 
+/// Aborting a `do_invite_async` task abandons the call like dropping the
+/// `do_invite` future. Aborted in Calling, the first response gets a CANCEL
+/// (180) or an ACK and a BYE (200); aborted in Early, the CANCEL goes out at
+/// once. Nothing stays registered in the layer.
+async fn run_aborted_invite_task(in_calling: bool, first: u16) -> crate::Result<()> {
+    let token = CancellationToken::new();
+    let Uac {
+        dialog_layer,
+        option,
+        peer,
+    } = setup(&token).await?;
+    let wait = Duration::from_secs(2);
+
+    let (state_sender, mut states) = unbounded_channel();
+    let (_dialog, join) = dialog_layer.do_invite_async(option, state_sender)?;
+    let (inv, uac) = recv_request(&peer, Method::Invite, wait).await;
+    calling_id(&mut states).await;
+    if !in_calling {
+        reply(&peer, uac, &inv, 180, "Ringing").await;
+        wait_for_state(&mut states, "Early", wait, |s| {
+            matches!(s, DialogState::Early(_, _))
+        })
+        .await;
+    }
+    join.abort();
+    let _ = join.await;
+    if in_calling {
+        reply(&peer, uac, &inv, first, "Response").await;
+    }
+    if first == 180 {
+        let (cancel, _) = recv_request(&peer, Method::Cancel, wait).await;
+        reply(&peer, uac, &cancel, 200, "OK").await;
+        reply(&peer, uac, &inv, 487, "Request Terminated").await;
+    }
+    let (ack, _) = recv_request(&peer, Method::Ack, wait).await;
+    assert_in_dialog(&ack, &inv, "ACK");
+    if first == 200 {
+        let (bye, _) = recv_request(&peer, Method::Bye, wait).await;
+        assert_in_dialog(&bye, &inv, "BYE");
+        reply(&peer, uac, &bye, 200, "OK").await;
+    }
+    tokio::time::sleep(Duration::from_millis(200)).await;
+    assert!(dialog_layer.is_empty(), "an aborted call is not registered");
+    token.cancel();
+    Ok(())
+}
+
+#[tokio::test]
+async fn test_aborted_invite_task_in_calling_is_cancelled_after_the_180() -> crate::Result<()> {
+    run_aborted_invite_task(true, 180).await
+}
+
+#[tokio::test]
+async fn test_aborted_invite_task_in_calling_2xx_is_acked_and_byed() -> crate::Result<()> {
+    run_aborted_invite_task(true, 200).await
+}
+
+#[tokio::test]
+async fn test_aborted_invite_task_in_early_is_cancelled() -> crate::Result<()> {
+    run_aborted_invite_task(false, 180).await
+}
+
 /// Dropped before any response: Terminated(UacCancel) is reported at once and
 /// nothing is sent while no provisional has arrived (RFC 3261 §9.1). The first
 /// response then gets a CANCEL (180), or an ACK and a BYE (200).
@@ -799,3 +861,93 @@ async fn test_cancel_in_calling_without_response_times_out() -> crate::Result<()
     token.cancel();
     Ok(())
 }
+
+#[tokio::test]
+async fn test_aborted_invite_before_first_poll_removes_dialog_without_sending() -> crate::Result<()>
+{
+    let token = CancellationToken::new();
+    let Uac {
+        dialog_layer,
+        option,
+        peer,
+    } = setup(&token).await?;
+    let (sender, _states) = unbounded_channel();
+    let (dialog, join) = dialog_layer.do_invite_async(option, sender)?;
+    // No yield between spawn and abort: the INVITE future has not run.
+    join.abort();
+    assert!(join.await.unwrap_err().is_cancelled());
+    assert!(dialog_layer.is_empty());
+    assert!(matches!(
+        dialog.state(),
+        DialogState::Terminated(_, TerminatedReason::UacCancel)
+    ));
+    let mut buf = [0u8; 4096];
+    assert!(
+        tokio::time::timeout(Duration::from_millis(100), peer.recv_from(&mut buf))
+            .await
+            .is_err()
+    );
+    token.cancel();
+    Ok(())
+}
+
+// Exercise both ways the old cleanup cut CANCEL short: an INVITE final
+// response, and its unrelated two-second deadline.
+async fn run_cleanup_cancel_retransmission(final_first: bool) -> crate::Result<()> {
+    let token = CancellationToken::new();
+    let Uac {
+        dialog_layer,
+        option,
+        peer,
+    } = setup(&token).await?;
+    let wait = Duration::from_secs(5);
+    let (sender, mut states) = unbounded_channel();
+    let (_dialog, join) = dialog_layer.do_invite_async(option, sender)?;
+    let (inv, uac) = recv_request(&peer, Method::Invite, wait).await;
+    reply(&peer, uac, &inv, 180, "Ringing").await;
+    wait_for_state(&mut states, "Early", wait, |s| {
+        matches!(s, DialogState::Early(_, _))
+    })
+    .await;
+    join.abort();
+    let _ = join.await;
+    let (cancel, _) = recv_request(&peer, Method::Cancel, wait).await;
+    if final_first {
+        reply(&peer, uac, &inv, 487, "Request Terminated").await;
+        recv_request(&peer, Method::Ack, wait).await;
+    }
+    // Withhold the CANCEL response, then require a retransmission after
+    // two seconds (discard earlier retries rather than counting queued data).
+    let started = tokio::time::Instant::now();
+    loop {
+        let (retry, _) = recv_request(&peer, Method::Cancel, wait).await;
+        assert_eq!(retry.to_string(), cancel.to_string());
+        if started.elapsed() > Duration::from_secs(2) {
+            reply(&peer, uac, &retry, 200, "OK").await;
+            break;
+        }
+    }
+    if !final_first {
+        reply(&peer, uac, &inv, 487, "Request Terminated").await;
+        recv_request(&peer, Method::Ack, wait).await;
+    }
+    // A received final response must stop further CANCEL retransmissions.
+    assert!(tokio::time::timeout(Duration::from_millis(4200), async {
+        recv_request(&peer, Method::Cancel, Duration::from_secs(5)).await
+    })
+    .await
+    .is_err());
+    assert!(dialog_layer.is_empty());
+    token.cancel();
+    Ok(())
+}
+
+#[tokio::test]
+async fn test_cleanup_cancel_survives_invite_final() -> crate::Result<()> {
+    run_cleanup_cancel_retransmission(true).await
+}
+
+#[tokio::test]
+async fn test_cleanup_cancel_retransmits_beyond_two_seconds() -> crate::Result<()> {
+    run_cleanup_cancel_retransmission(false).await
+}