From aa5e0e1e69cec7fd66673e35ee05b5be2c8279fe Mon Sep 17 00:00:00 2001 From: Tyson George Date: Thu, 24 Sep 2026 08:53:46 -0400 Subject: [PATCH] fix: notify dialog state only for transitions that are applied DialogInner::transition sent the new state to subscribers before deciding whether to apply it, so ignored transitions were still notified: a second Terminated when two teardown paths race (a local BYE completing while the peer's BYE is handled), and a WaitAck on an already confirmed dialog. The late-update check added in 0.7.0 reads the state under a lock it then releases, so a concurrent transition can still slip a notification past it. Decide under the state lock, update the state, then send the notification while still holding the lock, so every notification describes a transition that happened and notifications follow the order of state changes. Updates arriving after Terminated, including the event-only variants (Updated/Notify/Info/Options/Refer), stay dropped as in 0.7.0; while the dialog is live, event-only variants are sent as before. Adds an end-to-end test over UDP through DialogLayer: the peer sends an INFO, then a BYE before the application answers it; once the INFO is answered, no Confirmed may follow the Terminated notification. --- src/dialog/dialog.rs | 43 ++- src/dialog/tests/mod.rs | 1 + src/dialog/tests/test_dialog_states.rs | 270 ++++++++++++++++++ .../tests/test_state_after_terminated.rs | 184 ++++++++++++ 4 files changed, 472 insertions(+), 26 deletions(-) create mode 100644 src/dialog/tests/test_state_after_terminated.rs diff --git a/src/dialog/dialog.rs b/src/dialog/dialog.rs index 9f62fc2a..91f7553e 100644 --- a/src/dialog/dialog.rs +++ b/src/dialog/dialog.rs @@ -1546,18 +1546,17 @@ impl DialogInner { } } pub(super) fn transition(&self, state: DialogState) -> Result<()> { - // Try to send state update, but don't fail if channel is closed - // Late state updates after termination (e.g. CANCEL's 200 arriving - // after the INVITE's 487) are no longer broadcast and do not change - // the lifecycle state — observers already saw Terminated. - let terminated_now = matches!(&*self.state.lock(), DialogState::Terminated(..)); - if terminated_now && !matches!(state, DialogState::Terminated(..)) { - debug!(target = ?state, "dialog already terminated, ignoring late transition"); + // Decide under the state lock and notify only transitions that are + // applied, while still holding it, so notifications follow the order + // in which the state changed. + let mut old_state = self.state.lock(); + // Late updates after termination (e.g. CANCEL's 200 arriving after + // the INVITE's 487, or a second Terminated from a racing teardown) + // are neither applied nor broadcast: observers already saw Terminated. + if let DialogState::Terminated(id, _) = &*old_state { + debug!(id = %id, target = %state, "dialog already terminated, ignoring late transition"); return Ok(()); } - - self.state_sender.send(state.clone()).ok(); - // In-dialog request events do not change the established lifecycle state. match state { DialogState::Updated(_, _, _) @@ -1565,28 +1564,20 @@ impl DialogInner { | DialogState::Info(_, _, _) | DialogState::Options(_, _, _) | DialogState::Refer(_, _, _) => { + // Try to send state update, but don't fail if channel is closed + self.state_sender.send(state).ok(); return Ok(()); } _ => {} } - let mut old_state = self.state.lock(); - match (&*old_state, &state) { - (DialogState::Terminated(id, _), _) => { - warn!( - id = %id, - target = %state, - "dialog already terminated, ignoring transition" - ); - return Ok(()); - } - (DialogState::Confirmed(_, _), DialogState::WaitAck(_, _)) => { - warn!(target = %state, "dialog already confirmed, ignoring transition"); - return Ok(()); - } - _ => {} + if let (DialogState::Confirmed(_, _), DialogState::WaitAck(_, _)) = (&*old_state, &state) { + warn!(target = %state, "dialog already confirmed, ignoring transition"); + return Ok(()); } debug!(from = %old_state, to = %state, "transitioning state"); - *old_state = state; + *old_state = state.clone(); + // Try to send state update, but don't fail if channel is closed + self.state_sender.send(state).ok(); Ok(()) } diff --git a/src/dialog/tests/mod.rs b/src/dialog/tests/mod.rs index 52a59d7a..56e56a23 100644 --- a/src/dialog/tests/mod.rs +++ b/src/dialog/tests/mod.rs @@ -12,5 +12,6 @@ mod test_proxy_headers; mod test_refer; mod test_server_dialog; mod test_session_id; +mod test_state_after_terminated; mod test_sub_pub; mod test_uas_ack_timeout; diff --git a/src/dialog/tests/test_dialog_states.rs b/src/dialog/tests/test_dialog_states.rs index ad5a90b0..f41c639d 100644 --- a/src/dialog/tests/test_dialog_states.rs +++ b/src/dialog/tests/test_dialog_states.rs @@ -509,3 +509,273 @@ async fn test_dialog_id_creation() -> crate::Result<()> { Ok(()) } + +/// Name of a state variant, for asserting on notification sequences. +fn state_name(state: &DialogState) -> &'static str { + match state { + DialogState::Calling(_) => "Calling", + DialogState::Trying(_) => "Trying", + DialogState::Early(_, _) => "Early", + DialogState::WaitAck(_, _) => "WaitAck", + DialogState::Confirmed(_, _) => "Confirmed", + DialogState::Updated(_, _, _) => "Updated", + DialogState::Publish(_, _, _) => "Publish", + DialogState::Notify(_, _, _) => "Notify", + DialogState::Info(_, _, _) => "Info", + DialogState::Options(_, _, _) => "Options", + DialogState::Refer(_, _, _) => "Refer", + DialogState::Message(_, _, _) => "Message", + DialogState::Terminated(_, _) => "Terminated", + } +} + +/// Drain every notification currently queued on a state receiver. +fn drain_states( + receiver: &mut tokio::sync::mpsc::UnboundedReceiver, +) -> Vec<&'static str> { + let mut names = Vec::new(); + while let Ok(state) = receiver.try_recv() { + names.push(state_name(&state)); + } + names +} + +#[tokio::test] +async fn test_dialog_lifecycle_notifies_each_state_once() -> crate::Result<()> { + let endpoint = create_test_endpoint().await?; + let (state_sender, mut state_receiver) = unbounded_channel(); + let dialog_id = DialogId { + call_id: "test-call-id-lifecycle".to_string(), + local_tag: "alice-tag-456".to_string(), + remote_tag: "bob-tag-789".to_string(), + }; + let invite_req = create_invite_request("alice-tag-456", "", "test-call-id-lifecycle"); + let (tu_sender, _tu_receiver) = unbounded_channel(); + let dialog_inner = DialogInner::new( + TransactionRole::Client, + dialog_id.clone(), + invite_req, + endpoint.inner.clone(), + state_sender, + None, + Some(crate::sip::Uri::try_from( + "sip:alice@alice.example.com:5060", + )?), + tu_sender, + )?; + + dialog_inner.transition(DialogState::Calling(dialog_id.clone()))?; + dialog_inner.transition(DialogState::Trying(dialog_id.clone()))?; + let ringing = create_response( + StatusCode::Ringing, + "alice-tag-456", + "bob-tag-789", + "test-call-id-lifecycle", + ); + dialog_inner.transition(DialogState::Early(dialog_id.clone(), ringing))?; + let ok = create_response( + StatusCode::OK, + "alice-tag-456", + "bob-tag-789", + "test-call-id-lifecycle", + ); + dialog_inner.transition(DialogState::Confirmed(dialog_id.clone(), ok))?; + dialog_inner.transition(DialogState::Terminated( + dialog_id.clone(), + TerminatedReason::UacBye, + ))?; + + assert_eq!( + drain_states(&mut state_receiver), + vec!["Calling", "Trying", "Early", "Confirmed", "Terminated"] + ); + Ok(()) +} + +#[tokio::test] +async fn test_no_state_notification_after_terminated() -> crate::Result<()> { + let endpoint = create_test_endpoint().await?; + let (state_sender, mut state_receiver) = unbounded_channel(); + let dialog_id = DialogId { + call_id: "test-call-id-double-term".to_string(), + local_tag: "alice-tag-456".to_string(), + remote_tag: "bob-tag-789".to_string(), + }; + let invite_req = + create_invite_request("alice-tag-456", "bob-tag-789", "test-call-id-double-term"); + let (tu_sender, _tu_receiver) = unbounded_channel(); + let dialog_inner = DialogInner::new( + TransactionRole::Client, + dialog_id.clone(), + invite_req, + endpoint.inner.clone(), + state_sender, + None, + Some(crate::sip::Uri::try_from( + "sip:alice@alice.example.com:5060", + )?), + tu_sender, + )?; + + dialog_inner.transition(DialogState::Confirmed( + dialog_id.clone(), + Response::default(), + ))?; + // Two teardown paths racing: our BYE completes, and the peer's BYE is + // handled as well. + dialog_inner.transition(DialogState::Terminated( + dialog_id.clone(), + TerminatedReason::UacBye, + ))?; + dialog_inner.transition(DialogState::Terminated( + dialog_id.clone(), + TerminatedReason::UasBye, + ))?; + // A late state change after termination must not be applied or notified. + dialog_inner.transition(DialogState::Confirmed( + dialog_id.clone(), + Response::default(), + ))?; + + assert_eq!( + drain_states(&mut state_receiver), + vec!["Confirmed", "Terminated"], + "subscribers must see exactly one Terminated and nothing after it" + ); + assert!(matches!( + &*dialog_inner.state.lock(), + DialogState::Terminated(_, TerminatedReason::UacBye) + )); + Ok(()) +} + +#[tokio::test] +async fn test_ignored_waitack_after_confirmed_is_not_notified() -> crate::Result<()> { + let endpoint = create_test_endpoint().await?; + let (state_sender, mut state_receiver) = unbounded_channel(); + let dialog_id = DialogId { + call_id: "test-call-id-waitack".to_string(), + local_tag: "bob-tag-789".to_string(), + remote_tag: "alice-tag-456".to_string(), + }; + let invite_req = create_invite_request("alice-tag-456", "", "test-call-id-waitack"); + let (tu_sender, _tu_receiver) = unbounded_channel(); + let dialog_inner = DialogInner::new( + TransactionRole::Server, + dialog_id.clone(), + invite_req, + endpoint.inner.clone(), + state_sender, + None, + None, + tu_sender, + )?; + + dialog_inner.transition(DialogState::Confirmed( + dialog_id.clone(), + Response::default(), + ))?; + // e.g. a second accept() after the ACK already confirmed the dialog. + dialog_inner.transition(DialogState::WaitAck(dialog_id.clone(), Response::default()))?; + + assert_eq!(drain_states(&mut state_receiver), vec!["Confirmed"]); + assert!(dialog_inner.is_confirmed()); + Ok(()) +} + +#[tokio::test] +async fn test_info_answered_after_bye_does_not_notify_confirmed() -> crate::Result<()> { + use crate::dialog::server_dialog::ServerInviteDialog; + use crate::transaction::{key::TransactionKey, transaction::Transaction}; + use std::sync::Arc; + + let endpoint = create_test_endpoint().await?; + let (state_sender, mut state_receiver) = unbounded_channel(); + let dialog_id = DialogId { + call_id: "test-call-id-info-after-bye".to_string(), + local_tag: "bob-tag-789".to_string(), + remote_tag: "alice-tag-456".to_string(), + }; + let invite_req = create_invite_request("alice-tag-456", "", "test-call-id-info-after-bye"); + let (tu_sender, _tu_receiver) = unbounded_channel(); + let dialog_inner = DialogInner::new( + TransactionRole::Server, + dialog_id.clone(), + invite_req, + endpoint.inner.clone(), + state_sender, + None, + None, + tu_sender, + )?; + dialog_inner.transition(DialogState::Confirmed( + dialog_id.clone(), + Response::default(), + ))?; + let dialog = ServerInviteDialog { + inner: Arc::new(dialog_inner), + }; + + let in_dialog_request = |method: crate::sip::Method, cseq: &str, branch: &str| Request { + method, + uri: crate::sip::Uri::try_from("sip:bob@127.0.0.1:5060").unwrap(), + headers: vec![ + Via::new(format!("SIP/2.0/UDP 127.0.0.1:5060;branch={}", branch)).into(), + CSeq::new(cseq).into(), + From::new("Alice ;tag=alice-tag-456").into(), + To::new("Bob ;tag=bob-tag-789").into(), + CallId::new("test-call-id-info-after-bye").into(), + MaxForwards::new("70").into(), + ] + .into(), + version: crate::sip::Version::V2, + body: vec![], + }; + let server_tx = |req: Request| -> crate::Result { + let key = TransactionKey::from_request(&req, TransactionRole::Server)?; + Ok(Transaction::new_server( + key, + req, + endpoint.inner.clone(), + None, + )) + }; + + // An INFO arrives while confirmed; the application holds its handle. + let mut info_tx = server_tx(in_dialog_request( + crate::sip::Method::Info, + "2 INFO", + "z9hG4bK-info", + ))?; + let mut info_dialog = dialog.clone(); + let info_task = tokio::spawn(async move { info_dialog.handle(&mut info_tx).await }); + + assert_eq!(drain_states(&mut state_receiver), vec!["Confirmed"]); + let handle = match state_receiver.recv().await { + Some(DialogState::Info(_, _, handle)) => handle, + other => panic!("expected Info, got {:?}", other.as_ref().map(state_name)), + }; + + // Before the INFO is answered, the peer hangs up. + let mut bye_tx = server_tx(in_dialog_request( + crate::sip::Method::Bye, + "3 BYE", + "z9hG4bK-bye", + ))?; + dialog.clone().handle(&mut bye_tx).await.ok(); + + // The application answers the INFO after the dialog has terminated. + handle.reply(StatusCode::OK).await.ok(); + info_task.await.expect("info task panicked").ok(); + + assert_eq!( + drain_states(&mut state_receiver), + vec!["Terminated"], + "no Confirmed may be notified after Terminated" + ); + assert!(matches!( + &*dialog.inner.state.lock(), + DialogState::Terminated(_, TerminatedReason::UacBye) + )); + Ok(()) +} diff --git a/src/dialog/tests/test_state_after_terminated.rs b/src/dialog/tests/test_state_after_terminated.rs new file mode 100644 index 00000000..70373b63 --- /dev/null +++ b/src/dialog/tests/test_state_after_terminated.rs @@ -0,0 +1,184 @@ +//! Once a dialog has terminated (RFC 3261 §15: a BYE ends the dialog), the +//! state channel must not report it as anything else. A mid-dialog request +//! that the application answers after the peer's BYE must not produce a +//! `Confirmed` notification after `Terminated`. +use crate::dialog::{ + dialog::{Dialog, DialogState, DialogStateReceiver}, + dialog_layer::DialogLayer, + invite_dialog::InviteDialog, +}; +use crate::sip::{prelude::HeadersExt, Method, Response, SipMessage, StatusCode}; +use crate::transport::{udp::UdpConnection, TransportLayer}; +use crate::EndpointBuilder; +use std::net::SocketAddr; +use std::sync::Arc; +use std::time::Duration; +use tokio::net::UdpSocket; +use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver}; +use tokio_util::sync::CancellationToken; + +const CALL_ID: &str = "state-after-terminated-test"; +const FROM_TAG: &str = "uac-tag"; + +/// A raw UAC peer. +struct Peer { + socket: UdpSocket, + uas: SocketAddr, +} + +impl Peer { + async fn send_request(&self, method: Method, cseq: u32, to_tag: Option<&str>) { + let addr = self.socket.local_addr().unwrap(); + let to = match to_tag { + Some(tag) => format!(";tag={tag}", self.uas), + None => format!("", self.uas), + }; + let msg = format!( + "{method} sip:bob@{uas} SIP/2.0\r\n\ + Via: SIP/2.0/UDP {addr};branch=z9hG4bK-{method}-{cseq}\r\n\ + Max-Forwards: 70\r\n\ + From: ;tag={FROM_TAG}\r\n\ + To: {to}\r\n\ + Call-ID: {CALL_ID}\r\n\ + CSeq: {cseq} {method}\r\n\ + Contact: \r\n\ + Content-Length: 0\r\n\r\n", + uas = self.uas, + ); + self.socket.send_to(msg.as_bytes(), self.uas).await.unwrap(); + } + + /// Wait for the final response to the request with `cseq`. + async fn recv_final(&self, cseq: u32) -> Response { + let mut buf = vec![0u8; 4096]; + loop { + let (len, _) = + tokio::time::timeout(Duration::from_secs(2), self.socket.recv_from(&mut buf)) + .await + .expect("timeout waiting for a final response") + .unwrap(); + let text = std::str::from_utf8(&buf[..len]).unwrap(); + if let Ok(SipMessage::Response(resp)) = SipMessage::try_from(text) { + let seq = resp.cseq_header().unwrap().seq().unwrap(); + if seq == cseq && resp.status_code.code() >= 200 { + return resp; + } + } + } + } +} + +/// A UAS endpoint running the usual incoming-transaction loop. +async fn setup( + token: &CancellationToken, +) -> crate::Result<(DialogStateReceiver, UnboundedReceiver, Peer)> { + let transport_layer = TransportLayer::new(token.child_token()); + let udp = UdpConnection::create_connection( + "127.0.0.1:0".parse().unwrap(), + None, + Some(token.child_token()), + ) + .await?; + let uas: SocketAddr = udp.get_addr().get_socketaddr()?; + transport_layer.add_transport(udp.into()); + let endpoint = EndpointBuilder::new() + .with_user_agent("rsipstack-test") + .with_transport_layer(transport_layer) + .with_cancel_token(token.child_token()) + .build(); + let dialog_layer = Arc::new(DialogLayer::new(endpoint.inner.clone())); + let mut incoming = endpoint.incoming_transactions()?; + let endpoint_inner = endpoint.inner.clone(); + tokio::spawn(async move { + let _ = endpoint_inner.serve().await; + }); + + let (state_sender, states) = unbounded_channel(); + let (dialog_sender, dialogs) = unbounded_channel(); + tokio::spawn(async move { + while let Some(mut tx) = incoming.recv().await { + let has_to_tag = tx.original.to_header().unwrap().tag().unwrap().is_some(); + let dialog = if has_to_tag { + dialog_layer.match_dialog(&tx) + } else if tx.original.method == Method::Invite { + let dialog = dialog_layer + .get_or_create_server_invite(&tx, state_sender.clone(), None, None) + .expect("server dialog"); + dialog_sender.send(dialog.clone()).unwrap(); + Some(Dialog::Invite(dialog)) + } else { + None + }; + if let Some(mut dialog) = dialog { + tokio::spawn(async move { + let _ = dialog.handle(&mut tx).await; + }); + } + } + }); + + let peer = Peer { + socket: UdpSocket::bind("127.0.0.1:0").await?, + uas, + }; + Ok((states, dialogs, peer)) +} + +async fn next_state(rx: &mut DialogStateReceiver) -> DialogState { + tokio::time::timeout(Duration::from_secs(2), rx.recv()) + .await + .expect("timeout waiting for dialog state") + .expect("state channel closed") +} + +#[tokio::test] +async fn test_info_answered_after_peer_bye_is_not_notified_as_confirmed() -> crate::Result<()> { + let token = CancellationToken::new(); + let (mut states, mut dialogs, peer) = setup(&token).await?; + + // Establish the call: INVITE, 200, ACK. + peer.send_request(Method::Invite, 1, None).await; + let dialog = tokio::time::timeout(Duration::from_secs(2), dialogs.recv()) + .await + .expect("timeout waiting for the server dialog") + .unwrap(); + dialog.accept(None, None)?; + let ok = peer.recv_final(1).await; + let to_tag = ok.to_header()?.tag()?.expect("To tag").value().to_string(); + peer.send_request(Method::Ack, 1, Some(&to_tag)).await; + while !matches!(next_state(&mut states).await, DialogState::Confirmed(..)) {} + + // An INFO arrives; the application takes a moment to answer it. + peer.send_request(Method::Info, 2, Some(&to_tag)).await; + let info = loop { + if let DialogState::Info(_, _, handle) = next_state(&mut states).await { + break handle; + } + }; + + // Meanwhile the peer hangs up. + peer.send_request(Method::Bye, 3, Some(&to_tag)).await; + assert_eq!(peer.recv_final(3).await.status_code, StatusCode::OK); + assert!( + matches!(next_state(&mut states).await, DialogState::Terminated(..)), + "the peer's BYE must terminate the dialog" + ); + + // Now the application answers the INFO. + info.reply(StatusCode::OK).await.ok(); + assert_eq!(peer.recv_final(2).await.status_code, StatusCode::OK); + tokio::time::sleep(Duration::from_millis(200)).await; + + let mut after = Vec::new(); + while let Ok(state) = states.try_recv() { + after.push(state.to_string()); + } + assert!( + after.is_empty(), + "nothing may be notified after Terminated, got {after:?}" + ); + assert!(dialog.state().is_terminated()); + + token.cancel(); + Ok(()) +}