From 278c9682187c8a4715ef1ba809111fe409c6e6fe Mon Sep 17 00:00:00 2001 From: Tyson George Date: Fri, 9 Oct 2026 10:17:26 -0400 Subject: [PATCH] fix(transaction): add the 2xx ACK route before sending the 2xx A server INVITE added the route a 2xx ACK takes (dialog and CSeq) only when it entered Accepted, after the 2xx was sent. An ACK that arrived in between matched no transaction and was dropped, so the dialog was confirmed only by a later re-ACK, or ended at Timer L. respond() now adds the route before the send, and removes it again when the send fails or the respond() future is dropped. Fixes #192. --- .../tests/test_server_invite_ack.rs | 92 +++++++++++++++++++ src/transaction/transaction.rs | 58 ++++++++++-- 2 files changed, 141 insertions(+), 9 deletions(-) diff --git a/src/transaction/tests/test_server_invite_ack.rs b/src/transaction/tests/test_server_invite_ack.rs index 827982a4..e90b1f2d 100644 --- a/src/transaction/tests/test_server_invite_ack.rs +++ b/src/transaction/tests/test_server_invite_ack.rs @@ -255,3 +255,95 @@ async fn test_server_invite_ignores_ack_with_other_cseq_over_tcp() { token.cancel(); serve_handle.abort(); } + +/// The peer can ACK a final response as soon as it leaves the socket, +/// before `respond()` has moved the transaction on (issue #192). The send is +/// held in a full channel while the ACK arrives: the ACK must still reach +/// the transaction, for a 2xx (new branch, routed by dialog and CSeq) as for +/// a non-2xx (INVITE branch). A failed or cancelled send leaves no ACK +/// route behind. +#[tokio::test] +async fn test_server_invite_ack_during_final_response_send() { + use crate::transaction::key::{TransactionKey, TransactionRole}; + use crate::transaction::transaction::Transaction; + use crate::transport::{channel::ChannelConnection, SipAddr}; + use std::future::Future; + use std::task::{Context, Waker}; + + let endpoint = EndpointBuilder::new().build(); + let addr = SipAddr::from("127.0.0.1:5060".parse::().unwrap()); + for (cseq, status, outcome, after_ack) in [ + (3, StatusCode::OK, "send fails", TransactionState::Trying), + (5, StatusCode::OK, "cancelled", TransactionState::Trying), + (2, StatusCode::OK, "ACK", TransactionState::Terminated), + (4, StatusCode::BusyHere, "ACK", TransactionState::Confirmed), + ] { + let invite = make_reinvite("UDP", cseq); + let (_in_tx, in_rx) = tokio::sync::mpsc::unbounded_channel(); + let (out_tx, mut out_rx) = tokio::sync::mpsc::channel(1); + let conn: SipConnection = + ChannelConnection::create_connection_bounded(in_rx, out_tx.clone(), addr.clone(), None) + .await + .unwrap() + .into(); + let key = TransactionKey::from_request(&invite, TransactionRole::Server).unwrap(); + let mut tx = Transaction::new_server( + key, + invite.clone(), + endpoint.inner.clone(), + Some(conn.clone()), + ); + if outcome == "send fails" { + drop(out_rx); + assert!(tx.reply(status).await.is_err()); + assert_eq!(tx.state, after_ack); + assert_eq!(endpoint.inner.waiting_ack_cseq.len(), 0); + assert_eq!(endpoint.inner.waiting_ack.len(), 0); + continue; + } + let ack = if status == StatusCode::OK { + make_2xx_ack(&invite, cseq) + } else { + let mut ack = invite.clone(); + ack.method = Method::Ack; + for h in ack.headers.iter_mut() { + if let crate::sip::Header::CSeq(c) = h { + *c = CSeq::new(format!("{} ACK", cseq)); + } + } + ack + }; + // Fill the channel so the response send waits for room. + let filler = crate::transport::TransportEvent::Incoming( + invite.clone().into(), + conn.clone(), + addr.clone(), + ); + out_tx.try_send(filler).unwrap(); + let mut reply = Box::pin(tx.reply(status.clone())); + let mut cx = Context::from_waker(Waker::noop()); + assert!(reply.as_mut().poll(&mut cx).is_pending()); + if outcome == "cancelled" { + drop(reply); + assert_eq!(tx.state, after_ack); + assert_eq!(endpoint.inner.waiting_ack_cseq.len(), 0); + continue; + } + // The ACK arrives while the response is being sent. + endpoint + .inner + .on_received_message(ack.into(), conn.clone(), &addr) + .await + .unwrap(); + out_rx.recv().await.unwrap(); + reply.await.expect("reply"); + assert!(out_rx.recv().await.is_some(), "the response was sent"); + let msg = timeout(Duration::from_millis(200), tx.receive()) + .await + .unwrap_or_else(|_| panic!("{status}: the ACK must reach the transaction")) + .expect("transaction ended before the ACK"); + assert!(matches!(msg, SipMessage::Request(ref r) if r.method == Method::Ack)); + assert_eq!(tx.state, after_ack); + assert_eq!(endpoint.inner.waiting_ack_cseq.len(), 0); + } +} diff --git a/src/transaction/transaction.rs b/src/transaction/transaction.rs index 4c70e766..5cece64c 100644 --- a/src/transaction/transaction.rs +++ b/src/transaction/transaction.rs @@ -428,8 +428,10 @@ impl Transaction { /// # Cancel safety /// /// **NOT cancel-safe.** The mutation order is: store `last_response` - /// → `await` the transport send → transition the state machine. If the - /// future is dropped during the `await`, `last_response` will already + /// (and, for a server INVITE 2xx, add its ACK route) → `await` the + /// transport send → transition the state machine. If the + /// future is dropped during the `await`, the ACK route is removed + /// again but `last_response` will already /// be updated while the response may have been only partially /// transmitted (or not at all) and the state did not advance. If it /// is dropped between the send completing and the transition, the @@ -496,10 +498,35 @@ impl Transaction { SipMessage::Response(resp) => self.last_response.replace(resp), _ => None, }; + // The peer can ACK the 2xx before the transition below runs: add + // the route the ACK takes (new branch, matched by dialog and CSeq) + // before sending, so that ACK waits in this transaction's queue. + // The guard removes it again if the send fails or is cancelled. + let ack_route = self + .last_response + .as_ref() + .filter(|_| { + new_state == TransactionState::Accepted && self.state != TransactionState::Accepted + }) + .and_then(|resp| DialogId::try_from((resp, TransactionRole::Server)).ok()) + .and_then(|dialog_id| self.add_ack_cseq_route(&dialog_id)); + let mut guard = AckRouteGuard( + ack_route.map(|route| (self.endpoint_inner.clone(), route, self.key.clone())), + ); connection.send(response, self.destination.as_ref()).await?; + guard.0 = None; self.transition(new_state).map(|_| ()) } + fn add_ack_cseq_route(&self, dialog_id: &DialogId) -> Option<(DialogId, u32)> { + let seq = self.original.cseq_header().and_then(|c| c.seq()).ok()?; + let route = (dialog_id.clone(), seq); + self.endpoint_inner + .waiting_ack_cseq + .insert(route.clone(), self.key.clone()); + Some(route) + } + fn can_transition(&self, target: &TransactionState) -> Result<()> { match (&self.state, target) { (&TransactionState::Nothing, &TransactionState::Calling) @@ -1361,13 +1388,10 @@ impl Transaction { // one was answered, and must reach its own // transaction (RFC 3261 §13.2.2.4). A non-2xx ACK // matches by branch (§17.2.3) — unreachable here, - // Accepted only carries 2xx. - let seq = self.original.cseq_header().and_then(|c| c.seq()).ok(); - if let Some(seq) = seq { - self.endpoint_inner - .waiting_ack_cseq - .insert((dialog_id.clone(), seq), self.key.clone()); - } + // Accepted only carries 2xx. `respond` adds the + // CSeq route before sending the 2xx (an ACK can + // arrive before this transition). + self.add_ack_cseq_route(&dialog_id); self.endpoint_inner .waiting_ack .insert(dialog_id, self.key.clone()); @@ -1634,6 +1658,22 @@ impl Transaction { } } +/// A 2xx ACK route added by `respond` before the send; removed on drop +/// unless disarmed once the send completed. +struct AckRouteGuard(Option<(EndpointInnerRef, (DialogId, u32), TransactionKey)>); + +impl Drop for AckRouteGuard { + fn drop(&mut self) { + if let Some((endpoint_inner, route, key)) = self.0.take() { + endpoint_inner.waiting_ack_cseq.with_mut(|m| { + if m.get(&route) == Some(&key) { + m.remove(&route); + } + }); + } + } +} + impl Drop for Transaction { fn drop(&mut self) { self.cleanup();