From d2037bad22eeb012a9fb225c0f1e9e7cd401aee8 Mon Sep 17 00:00:00 2001 From: Tyson George Date: Sun, 11 Oct 2026 17:39:03 -0400 Subject: [PATCH 1/3] fix(dialog): end the INVITE of an aborted do_invite_async task do_invite_async drove the INVITE in a spawned task with no drop guard. Aborting that task before the final response dropped the transaction and left the early dialog registered in the layer: no CANCEL, a 2xx was neither ACKed nor ended with a BYE, and the dialog stayed in Calling or Early. The task now holds the same DialogGuardForUnconfirmed as do_invite, so an aborted task ends the INVITE like a dropped do_invite future: CANCEL at once in Trying or Early, on the first provisional in Calling (RFC 3261 section 9.1), ACK + BYE for a 2xx, and the layer entry is removed. --- src/dialog/invitation.rs | 21 +++++++- src/dialog/tests/test_cancel_2xx_race.rs | 62 ++++++++++++++++++++++++ 2 files changed, 81 insertions(+), 2 deletions(-) diff --git a/src/dialog/invitation.rs b/src/dialog/invitation.rs index 944adeaa..5b4f946d 100644 --- a/src/dialog/invitation.rs +++ b/src/dialog/invitation.rs @@ -776,7 +776,7 @@ impl DialogLayer { opt: InviteOption, state_sender: DialogStateSender, ) -> Result<(InviteDialog, crate::platform::JoinHandle)> { - 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) @@ -789,7 +789,24 @@ impl DialogLayer { // 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; + // As in do_invite: a task aborted before the final response + // still ends the INVITE (CANCEL, or ACK + BYE for a 2xx). + let mut guard = DialogGuardForUnconfirmed { + dialog_layer_inner: &inner, + id: &id0, + invite_tx: Some(tx), + dialog: Some(dialog_clone.clone()), + }; + 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()); diff --git a/src/dialog/tests/test_cancel_2xx_race.rs b/src/dialog/tests/test_cancel_2xx_race.rs index b57f5030..2521da87 100644 --- a/src/dialog/tests/test_cancel_2xx_race.rs +++ b/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). From d1ebf289970f208746decc6482f7ec0cee46638a Mon Sep 17 00:00:00 2001 From: yeoleobun Date: Mon, 12 Oct 2026 20:46:15 +0800 Subject: [PATCH 2/3] fix(dialog): preserve independent CANCEL cleanup after abort Create the INVITE cleanup guard before spawning so an abort before the first poll removes the early dialog without sending a request. Run the existing CANCEL operation independently until its response or transaction timeout, rather than dropping it after two seconds or an INVITE final response. Keep waiting for the INVITE final response and ACK/BYE cleanup. Add regressions for immediate abort and CANCEL retransmission beyond the old deadline and after an INVITE final response. All three fail on the original PR; 426 library tests and the platform-embassy compile check pass with this fix. Refs #204, #205. --- src/dialog/invitation.rs | 116 ++++++++--------------- src/dialog/tests/test_cancel_2xx_race.rs | 69 ++++++++++++++ 2 files changed, 106 insertions(+), 79 deletions(-) diff --git a/src/dialog/invitation.rs b/src/dialog/invitation.rs index 5b4f946d..055a9177 100644 --- a/src/dialog/invitation.rs +++ b/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, + /// Whether the owning future started processing the INVITE. + started: bool, /// The INVITE's dialog, `None` once `process_invite` returned. dialog: Option, } -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()), }; @@ -787,16 +742,19 @@ 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 { - // As in do_invite: a task aborted before the final response - // still ends the INVITE (CANCEL, or ACK + BYE for a 2xx). - let mut guard = DialogGuardForUnconfirmed { - dialog_layer_inner: &inner, - id: &id0, - invite_tx: Some(tx), - dialog: Some(dialog_clone.clone()), - }; + guard.started = true; let invite_tx = guard .invite_tx .as_mut() diff --git a/src/dialog/tests/test_cancel_2xx_race.rs b/src/dialog/tests/test_cancel_2xx_race.rs index 2521da87..bdb8397f 100644 --- a/src/dialog/tests/test_cancel_2xx_race.rs +++ b/src/dialog/tests/test_cancel_2xx_race.rs @@ -861,3 +861,72 @@ 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 +} From ebc03c435602311b474c6b9af3daccccd5a0889d Mon Sep 17 00:00:00 2001 From: yeoleobun Date: Mon, 12 Oct 2026 20:50:59 +0800 Subject: [PATCH 3/3] style: apply CI formatting for cancellation tests --- src/dialog/tests/test_cancel_2xx_race.rs | 35 +++++++++++++++++++----- 1 file changed, 28 insertions(+), 7 deletions(-) diff --git a/src/dialog/tests/test_cancel_2xx_race.rs b/src/dialog/tests/test_cancel_2xx_race.rs index bdb8397f..e0396b36 100644 --- a/src/dialog/tests/test_cancel_2xx_race.rs +++ b/src/dialog/tests/test_cancel_2xx_race.rs @@ -863,18 +863,30 @@ async fn test_cancel_in_calling_without_response_times_out() -> crate::Result<() } #[tokio::test] -async fn test_aborted_invite_before_first_poll_removes_dialog_without_sending() -> crate::Result<()> { +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 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))); + 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()); + assert!( + tokio::time::timeout(Duration::from_millis(100), peer.recv_from(&mut buf)) + .await + .is_err() + ); token.cancel(); Ok(()) } @@ -883,13 +895,20 @@ async fn test_aborted_invite_before_first_poll_removes_dialog_without_sending() // 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 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; + 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; @@ -915,7 +934,9 @@ async fn run_cleanup_cancel_retransmission(final_first: bool) -> crate::Result<( // 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()); + }) + .await + .is_err()); assert!(dialog_layer.is_empty()); token.cancel(); Ok(())