From 35746a8fdeddca3b6adf64cb0e531eb8059f1485 Mon Sep 17 00:00:00 2001 From: yeoleobun Date: Mon, 12 Jan 2026 16:12:51 +0800 Subject: [PATCH] cancel wait for 487 to invite --- src/dialog/authenticate.rs | 6 +- src/dialog/client_dialog.rs | 4 +- src/dialog/dialog.rs | 4 +- src/dialog/invitation.rs | 60 +++- src/dialog/registration.rs | 2 +- src/dialog/tests/test_authenticate.rs | 2 +- src/dialog/tests/test_client_dialog.rs | 390 +++++++++++++++++++++++++ 7 files changed, 456 insertions(+), 12 deletions(-) diff --git a/src/dialog/authenticate.rs b/src/dialog/authenticate.rs index 8ec6dcdb..a4cb2bda 100644 --- a/src/dialog/authenticate.rs +++ b/src/dialog/authenticate.rs @@ -124,7 +124,7 @@ pub struct Credential { /// // This is typically called automatically by dialog methods /// let new_tx = handle_client_authenticate( /// new_seq, -/// original_tx, +/// &original_tx, /// auth_challenge_response, /// &credential /// ).await?; @@ -159,7 +159,7 @@ pub struct Credential { /// StatusCode::Unauthorized | StatusCode::ProxyAuthenticationRequired => { /// // Handle authentication challenge /// let auth_tx = handle_client_authenticate( -/// new_seq, tx, resp, &credential +/// new_seq, &tx, resp, &credential /// ).await?; /// /// // Send authenticated request @@ -186,7 +186,7 @@ pub struct Credential { /// This function handles SIP authentication challenges and creates authenticated requests. pub async fn handle_client_authenticate( new_seq: u32, - tx: Transaction, + tx: &Transaction, resp: Response, cred: &Credential, ) -> Result { diff --git a/src/dialog/client_dialog.rs b/src/dialog/client_dialog.rs index 20b3aa10..5c1ee201 100644 --- a/src/dialog/client_dialog.rs +++ b/src/dialog/client_dialog.rs @@ -623,7 +623,7 @@ impl ClientInviteDialog { pub async fn process_invite( &self, - mut tx: Transaction, + tx: &mut Transaction, ) -> Result<(DialogId, Option)> { self.inner.transition(DialogState::Calling(self.id()))?; let mut auth_sent = false; @@ -662,7 +662,7 @@ impl ClientInviteDialog { } auth_sent = true; if let Some(credential) = &self.inner.credential { - tx = handle_client_authenticate( + *tx = handle_client_authenticate( self.inner.increment_local_seq(), tx, resp, diff --git a/src/dialog/dialog.rs b/src/dialog/dialog.rs index 366c3ab5..e630f3ac 100644 --- a/src/dialog/dialog.rs +++ b/src/dialog/dialog.rs @@ -596,7 +596,7 @@ impl DialogInner { auth_sent = true; if let Some(cred) = &self.credential { let new_seq = self.increment_local_seq(); - tx = handle_client_authenticate(new_seq, tx, resp, cred).await?; + tx = handle_client_authenticate(new_seq, &tx, resp, cred).await?; tx.send().await?; continue; } else { @@ -921,7 +921,7 @@ impl DialogInner { rsip::Method::Cancel => self.get_local_seq(), _ => self.increment_local_seq(), }; - tx = handle_client_authenticate(new_seq, tx, resp, cred).await?; + tx = handle_client_authenticate(new_seq, &tx, resp, cred).await?; tx.send().await?; continue; } else { diff --git a/src/dialog/invitation.rs b/src/dialog/invitation.rs index 92e3fa0b..b582fca4 100644 --- a/src/dialog/invitation.rs +++ b/src/dialog/invitation.rs @@ -5,7 +5,11 @@ use super::{ dialog_layer::DialogLayer, }; use crate::{ - dialog::{dialog::Dialog, dialog_layer::DialogLayerInnerRef, DialogId}, + dialog::{ + dialog::{Dialog, DialogState, TerminatedReason}, + dialog_layer::DialogLayerInnerRef, + DialogId, + }, transaction::{ key::{TransactionKey, TransactionRole}, make_tag, @@ -17,7 +21,7 @@ use crate::{ use futures::FutureExt; use rsip::{ prelude::{HeadersExt, ToTypedHeader}, - Request, Response, + Request, Response, SipMessage, StatusCodeKind, }; use std::sync::Arc; use tracing::{debug, info, warn}; @@ -171,6 +175,7 @@ impl Drop for DialogGuard { pub(super) struct DialogGuardForUnconfirmed<'a> { pub dialog_layer_inner: &'a DialogLayerInnerRef, pub id: &'a DialogId, + invite_tx: Option, } impl<'a> Drop for DialogGuardForUnconfirmed<'a> { @@ -180,7 +185,50 @@ impl<'a> Drop for DialogGuardForUnconfirmed<'a> { Ok(mut dialogs) => match dialogs.remove(&self.id.to_string()) { Some(dlg) => { info!(%self.id, "unconfirmed dialog dropped, cancelling it"); + let invite_tx = self.invite_tx.take(); let _ = tokio::spawn(async move { + if let Dialog::ClientInvite(ref client_dialog) = dlg { + if client_dialog.inner.can_cancel() { + if let Err(e) = client_dialog.cancel().await { + warn!(id = %client_dialog.id(), "dialog cancel failed: {}", e); + return; + } + + if let Some(mut invite_tx) = invite_tx { + let duration = tokio::time::Duration::from_secs(2); + let timeout = tokio::time::sleep(duration); + tokio::pin!(timeout); + loop { + tokio::select! { + _ = &mut timeout => break, + msg = invite_tx.receive() => { + if let Some(msg) = msg{ + if let SipMessage::Response(resp) = msg { + if resp.status_code.kind() != StatusCodeKind::Provisional { + debug!( + id = %client_dialog.id(), + status = %resp.status_code, + "received final response" + ); + break; + } + } + }else{ + break; + } + } + } + } + } + let _ = client_dialog.inner.transition(DialogState::Terminated( + client_dialog.id(), + TerminatedReason::UacCancel, + )); + tracing::info!(id = %client_dialog.id(), "dialog terminated"); + return; + } + } + if let Err(e) = dlg.hangup().await { info!(id=%dlg.id(), "failed to hangup unconfirmed dialog: {}", e); } @@ -426,11 +474,17 @@ impl DialogLayer { .ok(); info!(%id, "client invite dialog created"); - let _guard = DialogGuardForUnconfirmed { + let mut guard = DialogGuardForUnconfirmed { dialog_layer_inner: &self.inner, id: &id, + invite_tx: Some(tx), }; + let tx = guard + .invite_tx + .as_mut() + .expect("transcation should be avaible"); + let r = dialog.process_invite(tx).boxed().await; self.inner .dialogs diff --git a/src/dialog/registration.rs b/src/dialog/registration.rs index 887ba49e..466a8982 100644 --- a/src/dialog/registration.rs +++ b/src/dialog/registration.rs @@ -448,7 +448,7 @@ impl Registration { // Handle authentication with the existing transaction // The contact will be updated in the next registration cycle if needed - tx = handle_client_authenticate(self.last_seq, tx, resp, cred).await?; + tx = handle_client_authenticate(self.last_seq, &tx, resp, cred).await?; tx.send().await?; auth_sent = true; diff --git a/src/dialog/tests/test_authenticate.rs b/src/dialog/tests/test_authenticate.rs index 892a143f..8ced6b82 100644 --- a/src/dialog/tests/test_authenticate.rs +++ b/src/dialog/tests/test_authenticate.rs @@ -106,7 +106,7 @@ async fn test_authenticate_via_header_branch_update() -> crate::Result<()> { }; // Call handle_client_authenticate - let new_tx = handle_client_authenticate(2, tx, resp, &cred).await?; + let new_tx = handle_client_authenticate(2, &tx, resp, &cred).await?; // Verify the new request has updated Via header let new_via = new_tx diff --git a/src/dialog/tests/test_client_dialog.rs b/src/dialog/tests/test_client_dialog.rs index dd2b6b19..365b2bb5 100644 --- a/src/dialog/tests/test_client_dialog.rs +++ b/src/dialog/tests/test_client_dialog.rs @@ -1009,3 +1009,393 @@ async fn test_ack_sent_to_websocket_channel_via_locator() -> crate::Result<()> { Ok(()) } + +/// Test that dropping an invitation correctly cancels the INVITE +/// and waiting for the final response and send ACK. +/// +/// This test simulates: +/// 1. UAC sends INVITE +/// 2. UAS sends 100 Trying (dialog in Early state) +/// 3. UAC drops the invite future (triggers CANCEL) +/// 4. UAS responds with 200 OK to CANCEL and 487 to INVITE +/// 5. Verify the drop completes correctly +#[tokio::test] +async fn test_drop_unconfirmed_dialog_with_487_response() -> crate::Result<()> { + use crate::dialog::{dialog_layer::DialogLayer, invitation::InviteOption}; + // Start a UDP socket to simulate UAS + let uas_socket = tokio::net::UdpSocket::bind("127.0.0.1:0").await?; + let uas_port = uas_socket.local_addr()?.port(); + + let endpoint = create_test_endpoint().await?; + + // Setup outbound transport for client + let udp = UdpConnection::create_connection( + "127.0.0.1:0".parse().unwrap(), + None, + Some( + endpoint + .inner + .transport_layer + .inner + .cancel_token + .child_token(), + ), + ) + .await?; + let uac_port = udp.get_addr().addr.port.map(|p| u16::from(p)).unwrap_or(0); + endpoint.inner.transport_layer.add_transport(udp.into()); + endpoint.inner.transport_layer.serve_listens().await?; + + let endpoint_inner = endpoint.inner.clone(); + tokio::spawn(async move { + let _ = endpoint_inner.serve().await; + }); + + let dialog_layer = Arc::new(DialogLayer::new(endpoint.inner.clone())); + + let invite_option = InviteOption { + caller: Uri::try_from("sip:alice@example.com")?, + callee: Uri::try_from(format!("sip:bob@127.0.0.1:{}", uas_port).as_str())?, + contact: Uri::try_from(format!("sip:alice@127.0.0.1:{}", uac_port).as_str())?, + ..Default::default() + }; + + let (state_sender, mut state_receiver) = unbounded_channel(); + + // Start the invite in a task that we will abort to trigger drop + let dialog_layer_clone = dialog_layer.clone(); + let invite_handle = tokio::spawn(async move { + dialog_layer_clone + .do_invite(invite_option, state_sender) + .await + }); + + let mut buf = [0u8; 4096]; + + // Receive the INVITE request + let (len, uac_addr) = tokio::time::timeout( + std::time::Duration::from_secs(2), + uas_socket.recv_from(&mut buf), + ) + .await + .expect("timeout") + .expect("recv failed"); + + let invite_msg = std::str::from_utf8(&buf[..len]).unwrap(); + let invite_req: Request = rsip::SipMessage::try_from(invite_msg)?.try_into()?; + assert_eq!(invite_req.method, rsip::Method::Invite); + + // Send 100 Trying to put dialog in Early state + let trying_resp = format!( + "SIP/2.0 100 Trying\r\n\ + Via: {}\r\n\ + From: {}\r\n\ + To: {}\r\n\ + Call-ID: {}\r\n\ + CSeq: {}\r\n\ + Content-Length: 0\r\n\r\n", + invite_req.via_header()?.value(), + invite_req.from_header()?.value(), + invite_req.to_header()?.value(), + invite_req.call_id_header()?.value(), + invite_req.cseq_header()?.value(), + ); + uas_socket.send_to(trying_resp.as_bytes(), uac_addr).await?; + + // Wait for Trying state + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + + // Abort the invite handle to trigger drop + invite_handle.abort(); + + // Small delay for drop to start + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + + // Receive the CANCEL request + let (len, _) = tokio::time::timeout( + std::time::Duration::from_secs(2), + uas_socket.recv_from(&mut buf), + ) + .await + .expect("timeout receiving CANCEL") + .expect("recv failed"); + + let cancel_msg = std::str::from_utf8(&buf[..len]).unwrap(); + let cancel_req: Request = rsip::SipMessage::try_from(cancel_msg)?.try_into()?; + assert_eq!(cancel_req.method, rsip::Method::Cancel); + + // Send 200 OK to CANCEL + let cancel_ok_resp = format!( + "SIP/2.0 200 OK\r\n\ + Via: {}\r\n\ + From: {}\r\n\ + To: {}\r\n\ + Call-ID: {}\r\n\ + CSeq: {}\r\n\ + Content-Length: 0\r\n\r\n", + cancel_req.via_header()?.value(), + cancel_req.from_header()?.value(), + cancel_req.to_header()?.value(), + cancel_req.call_id_header()?.value(), + cancel_req.cseq_header()?.value(), + ); + uas_socket + .send_to(cancel_ok_resp.as_bytes(), uac_addr) + .await?; + + // Send 487 Request Terminated to INVITE + let invite_487_resp = format!( + "SIP/2.0 487 Request Terminated\r\n\ + Via: {}\r\n\ + From: {}\r\n\ + To: {};tag=uas-tag-487\r\n\ + Call-ID: {}\r\n\ + CSeq: {}\r\n\ + Content-Length: 0\r\n\r\n", + invite_req.via_header()?.value(), + invite_req.from_header()?.value(), + invite_req.to_header()?.value(), + invite_req.call_id_header()?.value(), + invite_req.cseq_header()?.value(), + ); + uas_socket + .send_to(invite_487_resp.as_bytes(), uac_addr) + .await?; + + // Receive ACK for 487 (may need to skip INVITE retransmissions) + let ack_req = loop { + let (len, _) = tokio::time::timeout( + std::time::Duration::from_secs(2), + uas_socket.recv_from(&mut buf), + ) + .await + .expect("timeout receiving ACK for 487") + .expect("recv failed"); + + let msg = std::str::from_utf8(&buf[..len]).unwrap(); + let req: Request = rsip::SipMessage::try_from(msg)?.try_into()?; + if req.method == rsip::Method::Ack { + break req; + } + // Skip INVITE retransmissions + }; + + // Verify ACK was received and conforms to RFC 3261 + assert_eq!(ack_req.method, rsip::Method::Ack, "Expected ACK for 487"); + + // ACK must have same Call-ID as INVITE + assert_eq!( + ack_req.call_id_header()?.value().to_string(), + invite_req.call_id_header()?.value().to_string(), + "ACK Call-ID must match INVITE" + ); + + // ACK must have same From header as INVITE + assert_eq!( + ack_req.from_header()?.value().to_string(), + invite_req.from_header()?.value().to_string(), + "ACK From header must match INVITE" + ); + + // ACK CSeq number must match INVITE (method will be ACK) + assert_eq!( + ack_req.cseq_header()?.seq()?, + invite_req.cseq_header()?.seq()?, + "ACK CSeq number must match INVITE" + ); + + // ACK must have same Request-URI as original INVITE + assert_eq!( + ack_req.uri.to_string(), + invite_req.uri.to_string(), + "ACK Request-URI must match INVITE" + ); + + // Verify no dialog left in the layer after drop completes + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + + // Check that state receiver got terminated state or is closed + // (dialog was removed from layer) + state_receiver.close(); + + Ok(()) +} + +/// Test that dropping an unconfirmed dialog completes even when the UAS +/// only responds to CANCEL with 200 OK but never sends a final response to INVITE. +/// +/// This test simulates a misbehaving UAS that doesn't send 487 after CANCEL. +/// The drop should still complete without hanging. +/// +/// This test has an internal 3-second timeout - if the drop hangs, the test will fail. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn test_drop_unconfirmed_dialog_without_final_response() -> crate::Result<()> { + // Wrap entire test in timeout to fail fast if drop hangs + tokio::time::timeout(std::time::Duration::from_secs(3), async { + test_drop_unconfirmed_dialog_without_final_response_impl().await + }) + .await + .expect("Test timed out - drop handler is likely hanging")?; + Ok(()) +} + +async fn test_drop_unconfirmed_dialog_without_final_response_impl() -> crate::Result<()> { + use crate::dialog::dialog::DialogState; + use crate::dialog::{dialog_layer::DialogLayer, invitation::InviteOption}; + + // Start a UDP socket to simulate UAS + let uas_socket = tokio::net::UdpSocket::bind("127.0.0.1:0").await?; + let uas_port = uas_socket.local_addr()?.port(); + + let endpoint = create_test_endpoint().await?; + + // Setup outbound transport for client + let udp = UdpConnection::create_connection( + "127.0.0.1:0".parse().unwrap(), + None, + Some( + endpoint + .inner + .transport_layer + .inner + .cancel_token + .child_token(), + ), + ) + .await?; + let uac_port = udp.get_addr().addr.port.map(|p| u16::from(p)).unwrap_or(0); + endpoint.inner.transport_layer.add_transport(udp.into()); + endpoint.inner.transport_layer.serve_listens().await?; + + let endpoint_inner = endpoint.inner.clone(); + tokio::spawn(async move { + let _ = endpoint_inner.serve().await; + }); + + let dialog_layer = Arc::new(DialogLayer::new(endpoint.inner.clone())); + + let invite_option = InviteOption { + caller: Uri::try_from("sip:alice@example.com")?, + callee: Uri::try_from(format!("sip:bob@127.0.0.1:{}", uas_port).as_str())?, + contact: Uri::try_from(format!("sip:alice@127.0.0.1:{}", uac_port).as_str())?, + ..Default::default() + }; + + let (state_sender, mut state_receiver) = unbounded_channel(); + + // Start the invite in a task that we will abort to trigger drop + let dialog_layer_clone = dialog_layer.clone(); + let invite_handle = tokio::spawn(async move { + dialog_layer_clone + .do_invite(invite_option, state_sender) + .await + }); + + let mut buf = [0u8; 4096]; + + // Receive the INVITE request + let (len, uac_addr) = tokio::time::timeout( + std::time::Duration::from_secs(2), + uas_socket.recv_from(&mut buf), + ) + .await + .expect("timeout") + .expect("recv failed"); + + let invite_msg = std::str::from_utf8(&buf[..len]).unwrap(); + let invite_req: Request = rsip::SipMessage::try_from(invite_msg)?.try_into()?; + assert_eq!(invite_req.method, rsip::Method::Invite); + + // Send 100 Trying to put dialog in Trying state + let trying_resp = format!( + "SIP/2.0 100 Trying\r\n\ + Via: {}\r\n\ + From: {}\r\n\ + To: {}\r\n\ + Call-ID: {}\r\n\ + CSeq: {}\r\n\ + Content-Length: 0\r\n\r\n", + invite_req.via_header()?.value(), + invite_req.from_header()?.value(), + invite_req.to_header()?.value(), + invite_req.call_id_header()?.value(), + invite_req.cseq_header()?.value(), + ); + uas_socket.send_to(trying_resp.as_bytes(), uac_addr).await?; + + // Wait for Trying state + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + + // Abort the invite handle to trigger drop + invite_handle.abort(); + + // Small delay for drop to start + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + + // Receive the CANCEL request + let (len, _) = tokio::time::timeout( + std::time::Duration::from_secs(2), + uas_socket.recv_from(&mut buf), + ) + .await + .expect("timeout receiving CANCEL") + .expect("recv failed"); + + let cancel_msg = std::str::from_utf8(&buf[..len]).unwrap(); + let cancel_req: Request = rsip::SipMessage::try_from(cancel_msg)?.try_into()?; + assert_eq!(cancel_req.method, rsip::Method::Cancel); + + // Send 200 OK to CANCEL only - deliberately don't send 487 to INVITE + let cancel_ok_resp = format!( + "SIP/2.0 200 OK\r\n\ + Via: {}\r\n\ + From: {}\r\n\ + To: {}\r\n\ + Call-ID: {}\r\n\ + CSeq: {}\r\n\ + Content-Length: 0\r\n\r\n", + cancel_req.via_header()?.value(), + cancel_req.from_header()?.value(), + cancel_req.to_header()?.value(), + cancel_req.call_id_header()?.value(), + cancel_req.cseq_header()?.value(), + ); + uas_socket + .send_to(cancel_ok_resp.as_bytes(), uac_addr) + .await?; + + // Wait for the drop handler to complete (with its internal 500ms timeout) + // The drop should transition the dialog to Terminated state + let terminated_received = tokio::time::timeout(std::time::Duration::from_secs(2), async { + while let Some(state) = state_receiver.recv().await { + if let DialogState::Terminated(_, reason) = state { + return Some(reason); + } + } + None + }) + .await; + + // Assert that dialog was properly terminated + match terminated_received { + Ok(Some(reason)) => { + assert!( + matches!(reason, TerminatedReason::UacCancel), + "Expected UacCancel termination reason, got {:?}", + reason + ); + } + Ok(None) => { + // Channel closed without Terminated state - acceptable if dialog was removed + } + Err(_) => { + // Timeout waiting for state - also acceptable since the drop may have completed + // without sending state (e.g., if state_sender was already dropped) + } + } + + // If we reach here without the test timeout (3s), the drop completed successfully + // The drop mechanism properly handles the case where no 487 is received + + Ok(()) +}