From eb18d486ca44333045469b3a4b470bb9cb4f703f Mon Sep 17 00:00:00 2001 From: jinti Date: Tue, 6 Oct 2026 22:00:17 +0800 Subject: [PATCH] feat(transaction): RFC 6026 Accepted state + Timer L/M for INVITE 2xx MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Squashed adoption of #128 by @andrico21 (closes #127), rebased onto the current transaction layer and adapted to the post-#146/#149/#155 semantics ('option B' — keep the 0.5.x wire behavior inside the new state machine): - client INVITE 2xx routes to Accepted with Timer M = 64*T1; the convenience auto-ACK still fires (documented deviation) and re-ACKs every retransmitted/forked 2xx, each delivered to the TU (§7.2); auto_ack_2xx = false (proxy mode) still terminates at once (#146); upstream-Via forwarding transactions park in Accepted without ACK - server INVITE 2xx routes to Accepted with Timer L = 64*T1 and waiting_ack routing; Timer G (T1 doubling to T2, #149) is kept as a documented deviation because the dialog layer does not retransmit 2xx, and stops when the matching ACK arrives - the ACK CSeq guard (#155) applies in Accepted as in Completed/Confirmed - DialogLayer::do_invite / do_invite_async drain the Accepted client transaction until Timer M so retransmissions keep being re-ACKed and the transaction is detached (no leak) - TransactionState/TransactionTimer are now #[non_exhaustive] (breaking, as flagged in #128) Co-authored-by: andrico21 --- src/dialog/invitation.rs | 23 + src/lib.rs | 15 +- src/transaction/mod.rs | 85 +++- src/transaction/tests/test_auto_ack_2xx.rs | 37 +- src/transaction/tests/test_client.rs | 30 +- .../tests/test_server_invite_ack.rs | 20 +- .../tests/test_transaction_states.rs | 316 ++++++++++++- src/transaction/transaction.rs | 440 +++++++++++++++++- 8 files changed, 892 insertions(+), 74 deletions(-) diff --git a/src/dialog/invitation.rs b/src/dialog/invitation.rs index 1169e5bf..54db4efe 100644 --- a/src/dialog/invitation.rs +++ b/src/dialog/invitation.rs @@ -691,6 +691,19 @@ impl DialogLayer { self.inner .dialogs .insert(new_dialog_id.to_string(), Dialog::Invite(dialog.clone())); + + // RFC 6026 §7.2: the client transaction is in Accepted + // with Timer M (64*T1) armed. Keep receiving until it + // ends so retransmitted 2xx keep being re-ACKed and + // forked 2xx are observed; abandoning the transaction + // here would leave it (and its timers) in the + // endpoint's table and silently stop the re-ACKs. + if let Some(mut tx) = guard.invite_tx.take() { + crate::platform::spawn(async move { + while tx.receive().await.is_some() {} + debug!(id = %new_dialog_id, "accepted transaction drained (Timer M expired)"); + }); + } } _ => {} } @@ -756,6 +769,16 @@ impl DialogLayer { inner .dialogs .insert(new_id.to_string(), Dialog::Invite(dialog_clone.clone())); + + // RFC 6026 §7.2: keep draining the Accepted transaction + // until Timer M ends it (re-ACKs retransmitted 2xx, + // observes forked 2xx, and detaches it from the + // endpoint's table). See do_invite for the rationale. + let confirmed_id = new_id.clone(); + crate::platform::spawn(async move { + while tx.receive().await.is_some() {} + debug!(id = %confirmed_id, "accepted transaction drained (Timer M expired)"); + }); } } Err(e) => debug!(%id0, error = %e, "async invite failed"), diff --git a/src/lib.rs b/src/lib.rs index 19f3ecc1..f3d1c713 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -198,7 +198,20 @@ //! //! * **RFC 3261** - SIP: Session Initiation Protocol (core specification) //! * **RFC 3581** - Symmetric Response Routing (rport) -//! * **RFC 6026** - Correct Transaction Handling for 2xx Responses to INVITE +//! * **RFC 6026** - Correct Transaction Handling for 2xx Responses to INVITE. +//! Server `Accepted` state + Timer L (§7.1); client `Accepted` state + Timer M +//! (§7.2). Two documented deviations keep the rsipstack 0.5.x wire behavior +//! inside the new state machine. Client deviation: the transaction auto-ACKs +//! a 2xx (and re-ACKs every retransmitted/forked 2xx) while parked in +//! `Accepted`; strict §7.2 + RFC 3261 §17.1.1.3 place that responsibility on +//! the TU — set `EndpointOption::auto_ack_2xx = false` for strict TU-owned +//! ACKs (proxy mode: the 2xx then terminates the transaction at once). +//! Server deviation: the transaction keeps Timer G (T1, doubling to T2, +//! RFC 3261 §13.3.1.4) while in `Accepted` and stops it when the ACK +//! arrives, because the rsipstack dialog layer does not retransmit 2xx +//! itself. Dialog users must keep receiving the client transaction until +//! Timer M ends it (`DialogLayer::do_invite`/`do_invite_async` do this); +//! the absorption window delivers every retransmitted/forked 2xx to the TU. //! //! ## Performance //! diff --git a/src/transaction/mod.rs b/src/transaction/mod.rs index 7a89fc94..1982c2b1 100644 --- a/src/transaction/mod.rs +++ b/src/transaction/mod.rs @@ -43,7 +43,8 @@ pub type TransactionSender = UnboundedSender; /// /// `TransactionState` represents the various states a SIP transaction can be in /// during its lifecycle. These states implement the transaction state machines -/// defined in RFC 3261 for both client and server transactions. +/// defined in RFC 3261 for both client and server transactions, with the RFC 6026 +/// `Accepted` state for INVITE 2xx response handling. /// /// # States /// @@ -51,29 +52,39 @@ pub type TransactionSender = UnboundedSender; /// * `Calling` - Initial state for client transactions when request is sent or received /// * `Trying` - Request has been sent/received, waiting for response/processing /// * `Proceeding` - Provisional response received/sent (1xx except 100 Trying) -/// * `Completed` - Final response received/sent, waiting for ACK (INVITE) or cleanup -/// * `Confirmed` - ACK received/sent for INVITE transactions +/// * `Accepted` - INVITE transaction received/sent a 2xx final response (RFC 6026 §7.1/§7.2); +/// server transaction waits for ACKs and absorbs 2xx retransmissions from the TU until +/// Timer L fires; client transaction absorbs 2xx retransmissions from the server until +/// Timer M fires. +/// * `Completed` - Final non-2xx response received/sent (RFC 3261 §17), waiting for ACK (server INVITE) +/// or response retransmissions (client). For INVITE 2xx, see `Accepted` (RFC 6026 supersedes RFC 3261 §17.2.1 paragraph 4 / §17.1.1.2). +/// * `Confirmed` - ACK received/sent for INVITE non-2xx (3xx-6xx) transactions /// * `Terminated` - Transaction has completed and is being cleaned up /// /// # State Transitions /// -/// ## Client Non-INVITE Transaction +/// ## Client Non-INVITE Transaction (RFC 3261 §17.1.2) /// ```text /// Nothing → Calling → Trying → Proceeding → Completed → Terminated /// ``` /// -/// ## Client INVITE Transaction +/// ## Client INVITE Transaction (RFC 3261 §17.1.1 + RFC 6026 §7.2) /// ```text -/// Nothing → Calling → Trying → Proceeding → Completed → Terminated -/// ↓ -/// Confirmed → Terminated +/// Nothing → Calling → Trying → Proceeding ──2xx──→ Accepted ──Timer M──→ Terminated +/// │ +/// └──3xx-6xx──→ Completed → Confirmed → Terminated +/// ``` +/// +/// ## Server INVITE Transaction (RFC 3261 §17.2.1 + RFC 6026 §7.1) +/// ```text +/// Calling → Trying → Proceeding ──2xx──→ Accepted ──Timer L──→ Terminated +/// │ +/// └──3xx-6xx──→ Completed ──ACK──→ Confirmed → Terminated /// ``` /// -/// ## Server Transactions +/// ## Server Non-INVITE Transaction (RFC 3261 §17.2.2) /// ```text /// Calling → Trying → Proceeding → Completed → Terminated -/// ↓ -/// Confirmed → Terminated (INVITE only) /// ``` /// /// # Examples @@ -87,17 +98,28 @@ pub type TransactionSender = UnboundedSender; /// TransactionState::Calling => println!("Request sent"), /// TransactionState::Trying => println!("Request sent/received"), /// TransactionState::Proceeding => println!("Provisional response"), -/// TransactionState::Completed => println!("Final response"), +/// TransactionState::Accepted => println!("2xx accepted; waiting for Timer L/M"), +/// TransactionState::Completed => println!("Final non-2xx response"), /// TransactionState::Confirmed => println!("ACK received/sent"), /// TransactionState::Terminated => println!("Transaction complete"), +/// // `#[non_exhaustive]`: downstream code MUST keep a wildcard arm to +/// // remain forward-compatible with future state additions. +/// _ => println!("future RFC-extension state"), /// } /// ``` #[derive(Debug, Clone, PartialEq)] +#[non_exhaustive] pub enum TransactionState { Nothing, Calling, Trying, Proceeding, + /// RFC 6026 §7.1 (server) / §7.2 (client): an INVITE transaction has sent or received + /// a 2xx final response. The server absorbs 2xx retransmissions from the TU and waits + /// for ACK until Timer L (= 64*T1) fires; the client absorbs server-retransmitted 2xx + /// responses until Timer M (= 64*T1) fires. Replaces the RFC 3261 §17.1.1.2 / §17.2.1 + /// behaviour that incorrectly routed 2xx through the `Completed` state. + Accepted, Completed, Confirmed, Terminated, @@ -110,6 +132,7 @@ impl core::fmt::Display for TransactionState { TransactionState::Calling => write!(f, "Calling"), TransactionState::Trying => write!(f, "Trying"), TransactionState::Proceeding => write!(f, "Proceeding"), + TransactionState::Accepted => write!(f, "Accepted"), TransactionState::Completed => write!(f, "Completed"), TransactionState::Confirmed => write!(f, "Confirmed"), TransactionState::Terminated => write!(f, "Terminated"), @@ -186,21 +209,22 @@ impl core::fmt::Display for TransactionType { /// SIP Transaction Timers /// /// `TransactionTimer` represents the various timers used in SIP transactions -/// as defined in RFC 3261. These timers ensure reliable message delivery -/// and proper transaction cleanup. +/// as defined in RFC 3261 and RFC 6026. These timers ensure reliable message +/// delivery and proper transaction cleanup. /// /// # Timer Types /// /// * `TimerA` - Retransmission timer for client transactions (unreliable transport) /// * `TimerB` - Transaction timeout timer for client transactions +/// * `TimerC` - Proceeding timeout for client INVITE /// * `TimerD` - Wait timer for response retransmissions (client) -/// * `TimerE` - Retransmission timer for non-INVITE server transactions -/// * `TimerF` - Transaction timeout timer for non-INVITE server transactions -/// * `TimerK` - Wait timer for ACK (server INVITE transactions) -/// * `TimerG` - Retransmission timer for INVITE server transactions +/// * `TimerK` - Wait timer for ACK (server INVITE non-2xx) / cleanup (client non-INVITE) +/// * `TimerG` - Retransmission timer for INVITE server transactions (non-2xx only per RFC 6026 §7.1) +/// * `TimerL` - RFC 6026 §7.1 server INVITE Accepted-state timer (64*T1) +/// * `TimerM` - RFC 6026 §7.2 client INVITE Accepted-state timer (64*T1) /// * `TimerCleanup` - Internal cleanup timer for transaction removal /// -/// # Timer Values (RFC 3261) +/// # Timer Values (RFC 3261 + RFC 6026) /// /// * T1 = 500ms (RTT estimate) /// * T2 = 4s (maximum retransmit interval) @@ -212,8 +236,10 @@ impl core::fmt::Display for TransactionType { /// * Timer D: 32 seconds for unreliable, 0 for reliable transports /// * Timer E: starts at T1, doubles up to T2 /// * Timer F: 64*T1 (32 seconds) -/// * Timer G: starts at T1, doubles up to T2 +/// * Timer G: starts at T1, doubles up to T2 (non-2xx final response retransmits only) /// * Timer K: T4 for unreliable, 0 for reliable transports +/// * **Timer L: 64*T1 (32 seconds) — RFC 6026 §7.1** +/// * **Timer M: 64*T1 (32 seconds) — RFC 6026 §7.2** /// /// # Examples /// @@ -245,6 +271,9 @@ impl core::fmt::Display for TransactionType { /// TransactionTimer::TimerB(key) => { /// println!("Transaction {} timed out", key); /// }, +/// // `#[non_exhaustive]`: downstream code MUST keep a wildcard arm to +/// // remain forward-compatible with future timer additions (e.g. RFC +/// // extensions like Timer L / Timer M / future RFC variants). /// _ => {} /// } /// # Ok(()) @@ -258,6 +287,7 @@ impl core::fmt::Display for TransactionType { /// * Cancelled when leaving states or receiving responses /// * Fire events that drive state machine transitions /// * Handle retransmissions and timeouts +#[non_exhaustive] pub enum TransactionTimer { TimerA(TransactionKey, Duration), TimerB(TransactionKey), @@ -265,6 +295,17 @@ pub enum TransactionTimer { TimerD(TransactionKey), TimerK(TransactionKey), TimerG(TransactionKey, Duration), + /// RFC 6026 §7.1: server INVITE Accepted-state timer. Fires once at 64*T1 + /// after a 2xx final response is sent, then transitions the transaction to + /// Terminated. Absorbs ACKs for the 2xx and late 2xx retransmissions from + /// the TU. Replaces the RFC 3261 §17.2.1 Timer K (T4) usage for 2xx + /// responses, which was too short to handle proxy-chain ACK fan-in. + TimerL(TransactionKey), + /// RFC 6026 §7.2: client INVITE Accepted-state timer. Fires once at 64*T1 + /// after a 2xx final response is received, then transitions the + /// transaction to Terminated. Absorbs server-retransmitted 2xx responses + /// (the TU is responsible for the ACK). + TimerM(TransactionKey), TimerCleanup(TransactionKey), } @@ -277,6 +318,8 @@ impl TransactionTimer { TransactionTimer::TimerD(key) => key, TransactionTimer::TimerG(key, _) => key, TransactionTimer::TimerK(key) => key, + TransactionTimer::TimerL(key) => key, + TransactionTimer::TimerM(key) => key, TransactionTimer::TimerCleanup(key) => key, } } @@ -295,6 +338,8 @@ impl core::fmt::Display for TransactionTimer { write!(f, "TimerG: {} {}", key, duration.as_millis()) } TransactionTimer::TimerK(key) => write!(f, "TimerK: {}", key), + TransactionTimer::TimerL(key) => write!(f, "TimerL: {}", key), + TransactionTimer::TimerM(key) => write!(f, "TimerM: {}", key), TransactionTimer::TimerCleanup(key) => write!(f, "TimerCleanup: {}", key), } } diff --git a/src/transaction/tests/test_auto_ack_2xx.rs b/src/transaction/tests/test_auto_ack_2xx.rs index a5053756..550e5066 100644 --- a/src/transaction/tests/test_auto_ack_2xx.rs +++ b/src/transaction/tests/test_auto_ack_2xx.rs @@ -162,7 +162,9 @@ async fn test_default_auto_acks_2xx_once() -> Result<()> { let mut tx = Transaction::new_client(key, invite, inner.clone(), None); tx.send().await?; - // The 2xx reaches the TU exactly once ... + // The 2xx reaches the TU and the transaction enters Accepted with + // Timer M armed (RFC 6026 §7.2; the auto-ACK is the documented + // rsipstack 0.5.x deviation). let first = tokio::time::timeout(Duration::from_secs(2), tx.receive()) .await .expect("timeout waiting for the 200") @@ -173,18 +175,21 @@ async fn test_default_auto_acks_2xx_once() -> Result<()> { } other => panic!("expected the 200 OK, got {other}"), } - // ... a retransmitted 200 OK must not reach the TU again: after the - // auto-ACK the transaction terminates and the TU channel closes, so - // receive() returns None instead of ever repeating the 2xx. - let next = tokio::time::timeout(Duration::from_millis(1200), tx.receive()).await; - assert!( - matches!(next, Ok(None)), - "the TU channel must close after the auto-ACK (no duplicate 2xx), got {next:?}" - ); - // RFC 3261 §17.1.1.2: the client INVITE transaction terminates - // immediately after ACKing the 2xx; retransmissions are absorbed - // below the TU (finished_transactions replays the stored ACK). - assert_eq!(tx.state, TransactionState::Terminated); + assert_eq!(tx.state, TransactionState::Accepted); + + // A retransmitted 200 OK is delivered to the TU again (§7.2) and + // re-ACKed below the TU. + let second = tokio::time::timeout(Duration::from_millis(1200), tx.receive()) + .await + .expect("timeout waiting for the retransmitted 200") + .expect("transaction ended before the retransmitted 200"); + match second { + SipMessage::Response(ref resp) => { + assert_eq!(resp.status_code.code(), 200); + } + other => panic!("expected the retransmitted 200 OK, got {other}"), + } + assert_eq!(tx.state, TransactionState::Accepted); Ok::<_, crate::Error>(()) }; @@ -212,10 +217,10 @@ async fn test_default_auto_acks_2xx_once() -> Result<()> { "the ACK must carry the 2xx's To tag, got {ack}" ); - // Each retransmitted 200 OK is absorbed with a re-ACK (the endpoint - // replays the stored ACK), never forwarded to the TU. + // Each retransmitted 200 OK is re-ACKed (RFC 6026 §7.2 absorption with + // the documented auto-ACK deviation). let re_ack = next_wire_event(&mut peer, Duration::from_millis(1100)).await; - let re_ack = re_ack.expect("a retransmitted 200 OK must be re-ACKed (absorbed)"); + let re_ack = re_ack.expect("a retransmitted 200 OK must be re-ACKed"); assert!( re_ack.contains("CSeq: 1 ACK"), "the re-ACK must carry the INVITE's CSeq, got {re_ack}" diff --git a/src/transaction/tests/test_client.rs b/src/transaction/tests/test_client.rs index bcaa3faa..70b9cd52 100644 --- a/src/transaction/tests/test_client.rs +++ b/src/transaction/tests/test_client.rs @@ -408,6 +408,7 @@ async fn test_invite_2xx_upstream_via_delivery_and_ack() -> Result<()> { response.to_header_mut()?.mut_tag(crate::transaction::make_tag())?; response.headers.push(Contact::new("").into()); let suppress_ack = proxy && status == crate::sip::StatusCode::OK; + let is_2xx = status == crate::sip::StatusCode::OK; let mut deadline = None; for iteration in 0..3 { if iteration == 2 { @@ -415,31 +416,48 @@ async fn test_invite_2xx_upstream_via_delivery_and_ack() -> Result<()> { response.to_header_mut()?.mut_tag(crate::transaction::make_tag())?; } endpoint.inner.on_received_message(response.clone().into(), conn.clone(), &addr).await?; - if iteration == 0 || suppress_ack { + // RFC 6026 §7.2: an Accepted client INVITE delivers every 2xx + // (retransmissions and forks) to the TU, so the test must + // keep receiving them; proxy-mode (suppress_ack) transactions + // in Accepted do the same. + if iteration == 0 || suppress_ack || is_2xx { let delivered = tokio::time::timeout(Duration::from_secs(1), tx.receive()).await .expect("response must reach application").expect("response"); assert_eq!(delivered, SipMessage::Response(response.clone())); } if suppress_ack { - assert_eq!(tx.state, TransactionState::Completed); + // RFC 6026 §7.2 with the documented auto-ACK deviation: + // an upstream-Via 2xx parks the forwarding transaction in + // Accepted (Timer M) and the TU owns the ACK. + assert_eq!(tx.state, TransactionState::Accepted); assert!(tx.last_ack.is_none()); assert!(outgoing_rx.try_recv().is_err(), "proxy must not send ACK"); if iteration == 0 { - deadline = tx.timer_d; + deadline = tx.timer_m; assert!(deadline.is_some()); } - assert_eq!(tx.timer_d, deadline, "duplicates must not extend lifetime"); + assert_eq!(tx.timer_m, deadline, "duplicates must not extend lifetime"); + } else if is_2xx { + // UA single-Via 2xx: Accepted with the convenience + // auto-ACK; every retransmission/fork is re-ACKed. + assert_eq!(tx.state, TransactionState::Accepted); + let TransportEvent::Incoming(SipMessage::Request(ack), _, _) = + outgoing_rx.try_recv().expect("UA 2xx requires ACK") else { + panic!("expected ACK"); + }; + assert_eq!(ack.method, crate::sip::Method::Ack); } else { + // 3xx-6xx: Completed → ACK → Terminated (RFC 3261 §17.1.1.3). assert_eq!(tx.state, TransactionState::Terminated); let TransportEvent::Incoming(SipMessage::Request(ack), _, _) = - outgoing_rx.try_recv().expect("UA 2xx and proxy non-2xx require ACK") else { + outgoing_rx.try_recv().expect("proxy non-2xx requires ACK") else { panic!("expected ACK"); }; assert_eq!(ack.method, crate::sip::Method::Ack); } } if suppress_ack { - tx.tu_sender.send(TransactionEvent::Timer(TransactionTimer::TimerD(key.clone()))).unwrap(); + tx.tu_sender.send(TransactionEvent::Timer(TransactionTimer::TimerM(key.clone()))).unwrap(); assert!(tokio::time::timeout(Duration::from_secs(1), tx.receive()).await.unwrap().is_none()); assert_eq!(tx.state, TransactionState::Terminated); assert!(!endpoint.inner.transactions.contains_key(&key)); diff --git a/src/transaction/tests/test_server_invite_ack.rs b/src/transaction/tests/test_server_invite_ack.rs index 8a1ea20e..96752676 100644 --- a/src/transaction/tests/test_server_invite_ack.rs +++ b/src/transaction/tests/test_server_invite_ack.rs @@ -111,7 +111,7 @@ async fn test_server_invite_ignores_ack_with_other_cseq() { assert_eq!(tx.original.method, Method::Invite); tx.reply(StatusCode::OK).await.expect("reply 200"); - assert_eq!(tx.state, TransactionState::Completed); + assert_eq!(tx.state, TransactionState::Accepted); assert!(tx.timer_g.is_some(), "Timer G must run for 2xx over UDP"); let mut buf = vec![0u8; 4096]; @@ -136,7 +136,7 @@ async fn test_server_invite_ignores_ack_with_other_cseq() { ); assert_eq!( tx.state, - TransactionState::Completed, + TransactionState::Accepted, "an ACK with CSeq 1 must not confirm the CSeq 2 INVITE transaction" ); assert!(tx.timer_g.is_some(), "Timer G must keep running"); @@ -170,9 +170,12 @@ async fn test_server_invite_ignores_ack_with_other_cseq() { } other => panic!("expected ACK, got {}", other), } - assert_eq!(tx.state, TransactionState::Confirmed); + // RFC 6026 §7.1: the ACK does not end the Accepted state (Timer L + // does); retransmissions stop and the waiting_ack entry stays so any + // retransmitted ACK keeps being routed here until then. + assert_eq!(tx.state, TransactionState::Accepted); assert!(tx.timer_g.is_none(), "Timer G must stop once ACKed"); - assert_eq!(endpoint.inner.waiting_ack.len(), 0); + assert_eq!(endpoint.inner.waiting_ack.len(), 1); token.cancel(); serve_handle.abort(); @@ -225,7 +228,7 @@ async fn test_server_invite_ignores_ack_with_other_cseq_over_tcp() { .expect("timeout waiting for incoming transaction") .expect("no incoming transaction"); tx.reply(StatusCode::OK).await.expect("reply 200"); - assert_eq!(tx.state, TransactionState::Completed); + assert_eq!(tx.state, TransactionState::Accepted); // Delayed ACK of CSeq 1, then the ACK of CSeq 2, on the ordered stream: // the first ACK the transaction delivers must be the CSeq 2 one. @@ -242,8 +245,11 @@ async fn test_server_invite_ignores_ack_with_other_cseq_over_tcp() { } other => panic!("expected ACK, got {}", other), } - assert_eq!(tx.state, TransactionState::Confirmed); - assert_eq!(endpoint.inner.waiting_ack.len(), 0); + // RFC 6026 §7.1: the ACK does not end the Accepted state (Timer L + // does); the waiting_ack entry stays so retransmitted ACKs keep being + // routed here until then. + assert_eq!(tx.state, TransactionState::Accepted); + assert_eq!(endpoint.inner.waiting_ack.len(), 1); token.cancel(); serve_handle.abort(); diff --git a/src/transaction/tests/test_transaction_states.rs b/src/transaction/tests/test_transaction_states.rs index ebf5fe8d..8c3e95b2 100644 --- a/src/transaction/tests/test_transaction_states.rs +++ b/src/transaction/tests/test_transaction_states.rs @@ -8,8 +8,10 @@ use crate::sip::headers::*; use crate::transaction::{ key::{TransactionKey, TransactionRole}, transaction::Transaction, - TransactionState, TransactionType, + TransactionState, TransactionTimer, TransactionType, }; +use crate::transport::udp::UdpConnection; +use crate::transport::SipConnection; /// Test helper to create a mock request fn create_test_request(method: crate::sip::Method, branch: &str) -> crate::sip::Request { @@ -17,12 +19,12 @@ fn create_test_request(method: crate::sip::Method, branch: &str) -> crate::sip:: method, uri: crate::sip::Uri::try_from("sip:test.example.com:5060").unwrap(), headers: vec![ - Via::new(&format!( + Via::new(format!( "SIP/2.0/UDP test.example.com:5060;branch={}", branch )) .into(), - CSeq::new(&format!("1 {}", method)).into(), + CSeq::new(format!("1 {}", method)).into(), From::new("Alice ;tag=1928301774").into(), To::new("Bob ").into(), CallId::new("a84b4c76e66710@pc33.atlanta.com").into(), @@ -162,3 +164,311 @@ async fn test_transaction_types() -> crate::Result<()> { Ok(()) } + +// =========================================================================== +// RFC 6026 conformance tests +// +// Validate the Accepted state and Timer L / Timer M behaviour required by +// RFC 6026 §7.1 (server INVITE) + §7.2 (client INVITE). +// +// Test design notes: +// - Display-impl tests are pure unit tests — no async setup required. +// - State-machine tests use a mock UDP connection bound to 127.0.0.1:0 +// so respond() can issue a real UDP send into the void; UDP is +// connectionless so the send succeeds even with no peer listener, +// letting the state transition fire without a full peer orchestration. +// - Timer L / Timer M actually firing (transitioning Accepted → Terminated +// after 64*T1 ≈ 32s) is NOT exercised here because waiting 32s per test +// would balloon the suite runtime; instead these tests verify +// timer_l.is_some() / timer_m.is_some() to confirm the timers are armed. +// The transition logic itself is exercised by the unit-level state +// machine in transaction.rs on_timer dispatch. +// =========================================================================== + +/// Helper: mock UDP connection on localhost for tests that exercise respond(). +/// The connection sends UDP packets to a bogus loopback port — UDP is +/// connectionless so the send succeeds (packets are dropped at the OS). +async fn mock_udp_connection() -> crate::Result { + let conn = UdpConnection::create_connection("127.0.0.1:0".parse()?, None, None).await?; + Ok(conn.into()) +} + +#[test] +fn test_rfc6026_accepted_state_display() { + // RFC 6026 §7.1/§7.2 exposes the Accepted state via the public + // TransactionState Display impl. Lock the rendering so external + // observers (logs, metrics, debug dumps) see "Accepted". + assert_eq!( + format!("{}", TransactionState::Accepted), + "Accepted", + "TransactionState::Accepted must Display as `Accepted` (RFC 6026 §7.1/§7.2)", + ); +} + +#[test] +fn test_rfc6026_timer_l_display() { + // RFC 6026 §7.1 server INVITE Accepted-state timer. + let request = crate::sip::Request { + method: crate::sip::Method::Invite, + uri: crate::sip::Uri::try_from("sip:bob@example.com").unwrap(), + headers: vec![ + Via::new("SIP/2.0/UDP example.com:5060;branch=z9hG4bKtimerL").into(), + CSeq::new("1 INVITE").into(), + From::new("Alice ;tag=t1").into(), + CallId::new("call-l@example.com").into(), + ] + .into(), + version: crate::sip::Version::V2, + body: Default::default(), + }; + let key = TransactionKey::from_request(&request, TransactionRole::Server).unwrap(); + let timer = TransactionTimer::TimerL(key.clone()); + assert_eq!(format!("{}", timer), format!("TimerL: {}", key)); +} + +#[test] +fn test_rfc6026_timer_m_display() { + // RFC 6026 §7.2 client INVITE Accepted-state timer. + let request = crate::sip::Request { + method: crate::sip::Method::Invite, + uri: crate::sip::Uri::try_from("sip:bob@example.com").unwrap(), + headers: vec![ + Via::new("SIP/2.0/UDP example.com:5060;branch=z9hG4bKtimerM").into(), + CSeq::new("1 INVITE").into(), + From::new("Alice ;tag=t1").into(), + CallId::new("call-m@example.com").into(), + ] + .into(), + version: crate::sip::Version::V2, + body: Default::default(), + }; + let key = TransactionKey::from_request(&request, TransactionRole::Client).unwrap(); + let timer = TransactionTimer::TimerM(key.clone()); + assert_eq!(format!("{}", timer), format!("TimerM: {}", key)); +} + +#[tokio::test] +async fn test_rfc6026_server_invite_2xx_routes_to_accepted_with_timer_l() -> crate::Result<()> { + // RFC 6026 §7.1: server INVITE 2xx final response transitions + // Proceeding/Trying → Accepted (NOT Completed), and arms Timer L. + // Timer G IS armed as a documented deviation (see lib.rs): strict §7.1 + // leaves 2xx retransmission to the TU, but the rsipstack dialog layer + // does not retransmit, so the transaction keeps the RFC 3261 + // §13.3.1.4 wire behavior (2xx retransmitted until ACK or Timer L). + let endpoint = create_test_endpoint(Some("127.0.0.1:0")).await?; + let conn = mock_udp_connection().await?; + + let invite_req = create_test_request(crate::sip::Method::Invite, "z9hG4bK6026srv2xx"); + let key = TransactionKey::from_request(&invite_req, TransactionRole::Server)?; + + let mut tx = Transaction::new_server( + key.clone(), + invite_req.clone(), + endpoint.inner.clone(), + Some(conn), + ); + tx.destination = Some(crate::transport::SipAddr::from( + "127.0.0.1:1".parse::()?, + )); + + tx.reply(crate::sip::StatusCode::OK).await?; + + assert_eq!( + tx.state, + TransactionState::Accepted, + "server INVITE 2xx must route to Accepted per RFC 6026 §7.1, not Completed", + ); + assert!( + tx.timer_l.is_some(), + "Accepted state must arm Timer L (= 64*T1) per RFC 6026 §7.1", + ); + assert!( + tx.timer_g.is_some(), + "Accepted state arms Timer G (documented deviation): the dialog layer does not retransmit 2xx, so the transaction keeps the RFC 3261 §13.3.1.4 retransmission", + ); + Ok(()) +} + +#[tokio::test] +async fn test_rfc6026_server_invite_4xx_still_routes_to_completed() -> crate::Result<()> { + // RFC 3261 §17.2.1 regression: non-2xx final responses (3xx-6xx) must + // continue to route to Completed (not Accepted) so the existing Timer G + // retransmit + Timer K + Timer D semantics for ACK matching are preserved. + let endpoint = create_test_endpoint(Some("127.0.0.1:0")).await?; + let conn = mock_udp_connection().await?; + + let invite_req = create_test_request(crate::sip::Method::Invite, "z9hG4bK6026srv404"); + let key = TransactionKey::from_request(&invite_req, TransactionRole::Server)?; + + let mut tx = Transaction::new_server( + key.clone(), + invite_req.clone(), + endpoint.inner.clone(), + Some(conn), + ); + tx.destination = Some(crate::transport::SipAddr::from( + "127.0.0.1:1".parse::()?, + )); + + tx.reply(crate::sip::StatusCode::NotFound).await?; + + assert_eq!( + tx.state, + TransactionState::Completed, + "server INVITE 4xx must continue to route to Completed (RFC 3261 §17.2.1 retained)", + ); + Ok(()) +} + +#[tokio::test] +async fn test_rfc6026_server_invite_accepted_registers_waiting_ack() -> crate::Result<()> { + // RFC 6026 §7.1: when a server INVITE transaction enters Accepted, it + // MUST register itself in the dialog-layer ACK routing map so the + // forthcoming ACK (sent by the UAC outside this transaction per + // §17.1.1.3) lands at this transaction key. The waiting_ack map keyed + // by DialogId is rsipstack's canonical routing primitive. + use crate::dialog::DialogId; + + let endpoint = create_test_endpoint(Some("127.0.0.1:0")).await?; + let conn = mock_udp_connection().await?; + + let invite_req = create_test_request(crate::sip::Method::Invite, "z9hG4bK6026srvack"); + let key = TransactionKey::from_request(&invite_req, TransactionRole::Server)?; + + let mut tx = Transaction::new_server( + key.clone(), + invite_req.clone(), + endpoint.inner.clone(), + Some(conn), + ); + tx.destination = Some(crate::transport::SipAddr::from( + "127.0.0.1:1".parse::()?, + )); + + tx.reply(crate::sip::StatusCode::OK).await?; + + // The 200 OK has been stored as last_response; derive the dialog_id + // (server role) and check that the waiting_ack map points back to us. + let last_response = tx + .last_response + .as_ref() + .expect("respond() must store the 2xx in last_response"); + let dialog_id = DialogId::try_from((last_response, TransactionRole::Server))?; + let registered = endpoint.inner.waiting_ack.get(&dialog_id); + assert!( + registered.is_some(), + "Accepted-state entry must register the dialog in waiting_ack for ACK routing per RFC 6026 §7.1", + ); + assert_eq!( + registered.expect("registered"), + key, + "waiting_ack entry must point back to this transaction key", + ); + Ok(()) +} + +#[tokio::test] +async fn test_rfc6026_server_invite_3xx_still_arms_timer_g() -> crate::Result<()> { + // RFC 6026 §7.1 retains Timer G for non-2xx final responses + // (the §7.1 prohibition is "MUST NOT generate 2xx retransmissions", + // explicitly NOT a blanket ban). This regression test ensures the + // server-side 3xx-6xx Timer G retransmit path remains live. + let endpoint = create_test_endpoint(Some("127.0.0.1:0")).await?; + let conn = mock_udp_connection().await?; + + let invite_req = create_test_request(crate::sip::Method::Invite, "z9hG4bK6026srvtimerg"); + let key = TransactionKey::from_request(&invite_req, TransactionRole::Server)?; + + let mut tx = Transaction::new_server( + key.clone(), + invite_req.clone(), + endpoint.inner.clone(), + Some(conn), + ); + tx.destination = Some(crate::transport::SipAddr::from( + "127.0.0.1:1".parse::()?, + )); + + tx.reply(crate::sip::StatusCode::ServerInternalError) + .await?; + + assert_eq!(tx.state, TransactionState::Completed); + assert!( + tx.timer_g.is_some(), + "non-2xx final must continue to arm Timer G (RFC 3261 §17.2.1 retained; RFC 6026 §7.1 only prohibits 2xx retransmits)", + ); + Ok(()) +} + +#[tokio::test] +async fn test_rfc6026_server_invite_accepted_absorbs_duplicate_2xx() -> crate::Result<()> { + // RFC 6026 §7.1: 'absorb retransmissions of the INVITE after a 2xx + // response has been sent' implies the server transaction MUST tolerate + // a TU-retransmitted 2xx in the Accepted state — the response is + // re-sent through the transport but no state error is raised. The + // can_transition() (Accepted, Accepted) self-loop edge enables this. + let endpoint = create_test_endpoint(Some("127.0.0.1:0")).await?; + let conn = mock_udp_connection().await?; + + let invite_req = create_test_request(crate::sip::Method::Invite, "z9hG4bK6026srvdup"); + let key = TransactionKey::from_request(&invite_req, TransactionRole::Server)?; + + let mut tx = Transaction::new_server( + key.clone(), + invite_req.clone(), + endpoint.inner.clone(), + Some(conn), + ); + tx.destination = Some(crate::transport::SipAddr::from( + "127.0.0.1:1".parse::()?, + )); + + let response_1 = tx.reply(crate::sip::StatusCode::OK).await; + response_1?; + assert_eq!(tx.state, TransactionState::Accepted); + + // Second TU-driven 2xx (e.g. retransmit per §7.1 forwarding rule): + // must succeed without state-machine error; transaction stays in + // Accepted (the (Accepted, Accepted) edge in can_transition). + tx.reply(crate::sip::StatusCode::OK).await?; + assert_eq!( + tx.state, + TransactionState::Accepted, + "duplicate 2xx in Accepted must keep the transaction in Accepted, not raise a transition error", + ); + Ok(()) +} + +#[tokio::test] +async fn test_rfc6026_server_invite_accepted_rejects_3xx_transition() -> crate::Result<()> { + // can_transition matrix: once a server INVITE has entered Accepted (2xx + // already sent), a subsequent 3xx-6xx attempt must be rejected. The + // matrix has no (Accepted, Completed) edge so respond() should error. + let endpoint = create_test_endpoint(Some("127.0.0.1:0")).await?; + let conn = mock_udp_connection().await?; + + let invite_req = create_test_request(crate::sip::Method::Invite, "z9hG4bK6026srvreject"); + let key = TransactionKey::from_request(&invite_req, TransactionRole::Server)?; + + let mut tx = Transaction::new_server( + key.clone(), + invite_req.clone(), + endpoint.inner.clone(), + Some(conn), + ); + tx.destination = Some(crate::transport::SipAddr::from( + "127.0.0.1:1".parse::()?, + )); + + tx.reply(crate::sip::StatusCode::OK).await?; + assert_eq!(tx.state, TransactionState::Accepted); + + // Now try sending a 4xx — this is illegal per the can_transition matrix + // (Accepted has no edge to Completed). reply() should return an Err. + let err = tx.reply(crate::sip::StatusCode::BusyHere).await; + assert!( + err.is_err(), + "reply() must reject Accepted → Completed (no such edge in can_transition matrix)", + ); + Ok(()) +} diff --git a/src/transaction/transaction.rs b/src/transaction/transaction.rs index 2a1c29eb..5bf2d52e 100644 --- a/src/transaction/transaction.rs +++ b/src/transaction/transaction.rs @@ -180,7 +180,9 @@ pub struct Transaction { pub timer_c: Option, pub timer_d: Option, pub timer_k: Option, // server invite only - pub timer_g: Option, // server invite only + pub timer_g: Option, // server invite only (non-2xx final response retransmits per RFC 6026 §7.1) + pub timer_l: Option, // server invite only (Accepted-state 64*T1 per RFC 6026 §7.1) + pub timer_m: Option, // client invite only (Accepted-state 64*T1 per RFC 6026 §7.2) retransmission: bool, is_cleaned_up: bool, } @@ -220,6 +222,8 @@ impl Transaction { timer_k: None, timer_g: None, retransmission: true, + timer_l: None, + timer_m: None, tu_receiver, tu_sender, is_cleaned_up: false, @@ -406,7 +410,35 @@ impl Transaction { pub async fn reply(&mut self, status_code: StatusCode) -> Result<()> { self.reply_with(status_code, vec![], None).await } - // send server response + /// Send a server response. + /// + /// Validates the state transition, sends the response through the + /// transaction's connection, stores it as `last_response`, and + /// transitions the state machine. The response routing follows + /// RFC 3261 §17.2.1 + RFC 6026 §7.1 (server INVITE 2xx → Accepted; + /// server INVITE 3xx-6xx → Completed; server non-INVITE → Terminated). + /// + /// # Errors + /// Returns `Error::TransactionError` if: + /// - the transaction is a client transaction (`respond` is server-only), + /// - the requested transition is not allowed by `can_transition`, + /// - the connection is missing, + /// - the underlying transport send fails. + /// + /// # 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 + /// 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 + /// response is on the wire but the transaction remains in the previous + /// state and subsequent timer-driven retransmits may fire incorrectly. + /// Drive `respond` from an owning task; do not place it directly + /// inside `tokio::select!` / `tokio::time::timeout` unless this hazard + /// is acceptable. Prefer the `TransactionEvent::Respond` channel for + /// cancellation-aware integration. pub async fn respond(&mut self, response: Response) -> Result<()> { match self.transaction_type { TransactionType::ServerInvite | TransactionType::ServerNonInvite => {} @@ -423,6 +455,20 @@ impl Transaction { StatusCode::Trying => TransactionState::Trying, _ => TransactionState::Proceeding, }, + StatusCodeKind::Successful => match self.transaction_type { + // RFC 6026 §7.1: server INVITE 2xx routes to Accepted, NOT + // Completed. Pre-RFC-6026 behaviour incorrectly used the + // Completed state — that triggered Timer G 2xx retransmits + // (forbidden by §7.1) and used Timer K (T4 ≈ 5s) as the + // effective ACK window instead of Timer L (64*T1 ≈ 32s), + // causing spurious failures on proxy-chain ACK fan-in. + TransactionType::ServerInvite => TransactionState::Accepted, + // ServerNonInvite 2xx → Terminated (no ACK expected). + _ => TransactionState::Terminated, + }, + // Non-2xx final response (3xx, 4xx, 5xx, 6xx): RFC 3261 §17.2.1 + // semantics retained — server INVITE → Completed (waits for + // ACK + Timer K + Timer D); server non-INVITE → Terminated. _ => match self.transaction_type { TransactionType::ServerInvite => TransactionState::Completed, _ => TransactionState::Terminated, @@ -463,17 +509,22 @@ impl Transaction { | (&TransactionState::Calling, &TransactionState::Calling) | (&TransactionState::Calling, &TransactionState::Trying) | (&TransactionState::Calling, &TransactionState::Proceeding) + | (&TransactionState::Calling, &TransactionState::Accepted) // RFC 6026 §7.2 | (&TransactionState::Calling, &TransactionState::Completed) | (&TransactionState::Calling, &TransactionState::Terminated) | (&TransactionState::Trying, &TransactionState::Trying) // retransmission | (&TransactionState::Trying, &TransactionState::Proceeding) + | (&TransactionState::Trying, &TransactionState::Accepted) // RFC 6026 §7.1/§7.2 | (&TransactionState::Trying, &TransactionState::Completed) | (&TransactionState::Trying, &TransactionState::Confirmed) | (&TransactionState::Trying, &TransactionState::Terminated) | (&TransactionState::Proceeding, &TransactionState::Proceeding) + | (&TransactionState::Proceeding, &TransactionState::Accepted) // RFC 6026 §7.1/§7.2 | (&TransactionState::Proceeding, &TransactionState::Completed) | (&TransactionState::Proceeding, &TransactionState::Confirmed) | (&TransactionState::Proceeding, &TransactionState::Terminated) + | (&TransactionState::Accepted, &TransactionState::Accepted) // RFC 6026 §7.1: absorb 2xx retransmits + | (&TransactionState::Accepted, &TransactionState::Terminated) // RFC 6026 §7.1/§7.2: Timer L/M fires | (&TransactionState::Completed, &TransactionState::Confirmed) | (&TransactionState::Completed, &TransactionState::Terminated) | (&TransactionState::Confirmed, &TransactionState::Terminated) => Ok(()), @@ -517,6 +568,40 @@ impl Transaction { } } + /// Send an ACK for a final response on a client INVITE transaction. + /// + /// Constructs the ACK from `last_ack` (cached) or `last_response` + + /// the original INVITE, resolves the destination (Contact for 2xx + /// per RFC 3261 §13.2.2.4; Via/Record-Route for 3xx-6xx per §17.1.1.3), + /// and sends through either the supplied `connection` argument or, for + /// 2xx responses, a transport-layer connection looked up from the + /// resolved Contact. For connection-oriented transports the ACK reuses + /// the flow the response arrived on (RFC 7118 §5) instead of dialing + /// the Record-Route/Contact target. The ACK is recorded as `last_ack` + /// regardless of send outcome. + /// + /// State transitions per RFC 6026 §7.2 + RFC 3261 §17.1.1.3: + /// - Completed (3xx-6xx): transitions to Terminated. + /// - Accepted (2xx): no transition (Timer M drives Termination so + /// the §7.2 server-retransmitted-2xx absorption window is preserved). + /// + /// # Errors + /// Returns `Error::TransactionError` if: + /// - the transaction is not a client INVITE, + /// - the state is neither Completed nor Accepted, + /// - no `last_response` and no `last_ack` is available, + /// - destination resolution or transport send fails. + /// + /// # Cancel safety + /// + /// **NOT cancel-safe.** This method has multiple `.await` points + /// (locator lookup, transport lookup, transport send) interleaved + /// with `last_ack` mutation and state transition. If the future is + /// dropped between the ACK transmission and the state transition, + /// the ACK will have been sent but the transaction state will remain + /// in Completed/Accepted. The next inbound message may trigger + /// inappropriate retransmits. Do not place inside `tokio::select!` + /// unless this hazard is acceptable. pub async fn send_ack(&mut self, mut connection: Option) -> Result<()> { if self.transaction_type != TransactionType::ClientInvite { return Err(Error::TransactionError( @@ -526,7 +611,7 @@ impl Transaction { } match self.state { - TransactionState::Completed => {} // must be in completed state, to send ACK + TransactionState::Completed | TransactionState::Accepted => {} // RFC 3261 §17.1.1 (Completed, 3xx-6xx) or RFC 6026 §7.2 (Accepted, 2xx) _ => { return Err(Error::TransactionError( format!("invalid state for sending ACK {:?}", self.state), @@ -617,16 +702,55 @@ impl Transaction { } else { debug!(key = %self.key, "no connection for ack"); } - // client send ack and transition to Terminated - self.transition(TransactionState::Terminated).map(|_| ()) + // RFC 3261 §17.1.1.3 / RFC 6026 §7.2: ACK in Completed (3xx-6xx) + // immediately terminates the client transaction. ACK in Accepted + // (2xx) leaves the transaction in Accepted to absorb server- + // retransmitted 2xx duplicates per §7.2; Timer M drives the + // eventual transition to Terminated. Returning Ok(()) here lets + // the caller observe a successful ACK send without forcing a + // premature transition. + if self.state == TransactionState::Completed { + self.transition(TransactionState::Terminated).map(|_| ()) + } else { + debug_assert_eq!( + self.state, + TransactionState::Accepted, + "send_ack reached post-send dispatch in unexpected state; the entry guard restricts to Completed|Accepted", + ); + Ok(()) + } } - /// Receive the next message for the transaction. + /// Receive the next SIP message routed to this transaction. + /// + /// Loops on the transaction's TU receiver, dispatching incoming + /// requests/responses through `on_received_request` / + /// `on_received_response`, processing timer events via `on_timer`, + /// and forwarding `Respond` events through `respond`. Returns + /// `Some(msg)` when a request/response should be propagated to the + /// dialog/TU layer; returns `None` when the transaction is + /// terminated by a `Terminate` event or the underlying channel + /// closes. /// /// INVITE 2xx responses containing another Via hop are delivered without /// an automatic ACK, including retransmissions. Forwarding applications /// should retain the transaction and keep receiving until this returns None; /// the cleanup deadline is measured from the first such final response. + /// + /// # Cancel safety + /// + /// **NOT cancel-safe.** State mutations may occur around any inner + /// `.await` — for example, an incoming response is passed to + /// `on_received_response` which transitions state and stores + /// `last_response` BEFORE the message is returned to the caller. + /// If the future is dropped between those mutations and the + /// `return Some(msg)`, the TU will not observe the response while + /// the transaction has already advanced state. Drive + /// `Transaction::receive` from an owning task; do not place it + /// directly inside `tokio::select!` / `tokio::time::timeout` unless + /// dropped state is acceptable. If timeout is required, prefer a + /// dedicated cancellation channel that the transaction itself + /// observes via `TransactionEvent::Terminate`. pub async fn receive(&mut self) -> Option { while let Some(event) = self.tu_receiver.recv().await { match event { @@ -748,6 +872,47 @@ impl Transaction { self.respond(last_response.to_owned()).await.ok(); } } + TransactionState::Accepted => { + // RFC 6026 §7.1: server INVITE Accepted state. + // + // ACK received: pass to TU; remain in Accepted (transition to + // Terminated is driven by Timer L expiry, not by ACK arrival + // — §7.1 'Timer L reflects the amount of time the server + // transaction could receive 2xx responses for retransmission + // from the TU while it is waiting to receive an ACK'). + // + // INVITE retransmit: silently absorbed. Per §7.1 the server + // transaction MUST NOT retransmit the 2xx on its own (only + // the TU does, via respond()), so re-firing respond() here + // would re-issue the 2xx through the transport AND trigger + // a no-op Accepted-self-loop transition. Just drop. + // + // As in Completed/Confirmed, an ACK carries the CSeq number + // of the INVITE it acknowledges (RFC 3261 §13.2.2.4): a + // delayed ACK of an earlier re-INVITE routed here through + // `waiting_ack` must not be delivered to the TU of this + // transaction. + if req.method == Method::Ack { + let ack_seq = req.cseq_header().and_then(|c| c.seq()).ok(); + let invite_seq = self.original.cseq_header().and_then(|c| c.seq()).ok(); + if ack_seq != invite_seq { + debug!( + key = %self.key, + ?ack_seq, + ?invite_seq, + "ignoring ACK with a CSeq that does not match the INVITE" + ); + return None; + } + // The peer confirmed the dialog: stop the 2xx + // retransmissions (RFC 3261 §13.3.1.4) and remain in + // Accepted until Timer L ends the transaction. + if let Some(id) = self.timer_g.take() { + self.endpoint_inner.timers.cancel(id); + } + return Some(req.into()); + } + } TransactionState::Completed | TransactionState::Confirmed if req.method == Method::Ack => { @@ -790,10 +955,11 @@ impl Transaction { { // A response with an upstream Via belongs to a forwarding transaction. // Its TU must receive every 2xx, including retransmissions and forks, - // and the originating UA owns the ACK. Keep the first Timer D deadline. - if self.state != TransactionState::Completed { - self.can_transition(&TransactionState::Completed).ok()?; - self.transition(TransactionState::Completed).ok()?; + // and the originating UA owns the ACK. Keep the first Timer M deadline + // (RFC 6026 §7.2 Accepted window). + if self.state != TransactionState::Accepted { + self.can_transition(&TransactionState::Accepted).ok()?; + self.transition(TransactionState::Accepted).ok()?; } self.last_response.replace(resp.clone()); return Some(SipMessage::Response(resp)); @@ -806,26 +972,52 @@ impl Transaction { TransactionState::Proceeding } } - _ => { + StatusCodeKind::Successful => { if self.transaction_type == TransactionType::ClientInvite { - if resp.status_code.kind() == StatusCodeKind::Successful - && !self.endpoint_inner.option.auto_ack_2xx - { + if !self.endpoint_inner.option.auto_ack_2xx { // auto_ack_2xx = false (proxy mode): a 2xx terminates // the client INVITE transaction at once; the ACK is // the TU's job (RFC 3261 section 17.1.1.2). TransactionState::Terminated } else { - TransactionState::Completed + // RFC 6026 §7.2: client INVITE 2xx routes to Accepted + // with Timer M = 64*T1 as the server-retransmitted-2xx + // absorption window. The convenience auto-ACK still + // fires below (rsipstack 0.5.x UA behavior; the strict + // §7.2 TU-owned ACK can be had with auto_ack_2xx = + // false). + TransactionState::Accepted } } else { TransactionState::Terminated } } + _ => { + // 3xx-6xx final response: RFC 3261 §17.1.1 unchanged. + if self.transaction_type == TransactionType::ClientInvite { + TransactionState::Completed + } else { + TransactionState::Terminated + } + } }; self.can_transition(&new_state).ok()?; - if self.state == new_state { + + // RFC 6026 §7.2: every 2xx response received by a client INVITE in + // the Accepted state MUST be passed to the TU — both genuine + // server-retransmitted 2xx (which the TU/dialog re-acknowledges) + // and forked 2xx (which can share status code + body but differ + // in To-tag and so identify a different dialog). The pre-existing + // duplicate-suppression filter below is correct for non-INVITE + // transactions where the TU has no use for duplicates, but applying + // it to the Accepted self-loop would silently swallow forked dialogs + // and prevent the legacy auto-ACK from re-firing per retransmit. + let is_client_invite_2xx_in_accepted = self.transaction_type + == TransactionType::ClientInvite + && new_state == TransactionState::Accepted; + + if !is_client_invite_2xx_in_accepted && self.state == new_state { if let Some(last) = self.last_response.as_ref() { if last.status_code == resp.status_code && last.body == resp.body { // ignore duplicate response @@ -835,13 +1027,29 @@ impl Transaction { } self.last_response.replace(resp.clone()); - let is_completed_client_invite = self.transaction_type == TransactionType::ClientInvite - && new_state == TransactionState::Completed; + // Auto-ACK gate. Pre-RFC-6026 this fired only for new_state == + // Completed (which captured both 2xx and 3xx-6xx in the old routing). + // Post-RFC-6026 split: 2xx → Accepted, 3xx-6xx → Completed. Both + // states still benefit from the convenience auto-ACK; the transaction + // layer constructs and sends the ACK via send_ack below. send_ack + // transitions Completed → Terminated (RFC 3261 §17.1.1) but leaves + // Accepted alone so Timer M can fire and the §7.2 2xx-retransmit + // absorption window is preserved. + let auto_ack_client_invite = self.transaction_type == TransactionType::ClientInvite + && (new_state == TransactionState::Completed + || new_state == TransactionState::Accepted); self.transition(new_state).ok(); - if is_completed_client_invite { - self.send_ack(connection).await.ok(); + if auto_ack_client_invite { + if let Err(e) = self.send_ack(connection).await { + warn!( + key = %self.key, + state = %self.state, + error = %e, + "auto-ACK for client INVITE final response failed; downstream TU may need to handle ACK explicitly", + ); + } } Some(SipMessage::Response(resp)) @@ -928,7 +1136,20 @@ impl Transaction { } TransactionState::Completed => { if let TransactionTimer::TimerG(key, duration) = timer { - // resend the response + // RFC 6026 §7.1 defensive guard: the server transaction + // MUST NOT retransmit 2xx responses on its own. Per the + // RFC 6026 routing in respond() + on_received_response(), + // 2xx finals route to the Accepted state, not Completed — + // so `last_response` here should always be non-2xx. This + // guard catches any legacy / out-of-band code path that + // might land a 2xx in Completed; suppress the retransmit + // and let Timer D / Timer K handle Termination. + if let Some(last_response) = &self.last_response { + if last_response.status_code.kind() == StatusCodeKind::Successful { + return Ok(()); + } + } + // resend the response (non-2xx final — RFC 3261 §17.2.1) if let Some(last_response) = &self.last_response { if let Some(connection) = &self.connection { let last_response = if let Some(ref inspector) = @@ -959,6 +1180,79 @@ impl Transaction { self.transition(TransactionState::Terminated)?; } } + TransactionState::Accepted => { + // RFC 6026 §7.1 (server INVITE Timer L) / §7.2 (client INVITE + // Timer M): on expiry the transaction transitions to + // Terminated. Timer L is server-only per §7.1; Timer M is + // client-only per §7.2 — mismatched pairings indicate a + // programming bug and are logged for visibility. + // + // Timer G (server): retransmit the 2xx with the interval + // doubling up to T2 (RFC 3261 §13.3.1.4) — see the + // documented Timer G deviation on the Accepted entry. + // Stray Timer A/B/C/D/K/Cleanup firings are race remnants + // from prior states (the Accepted-state entry handler + // cancels them, but fire-in-flight races are possible); + // listed explicitly so future timer additions force + // compile-time review here. + match (&self.transaction_type, &timer) { + (TransactionType::ServerInvite, TransactionTimer::TimerL(_)) + | (TransactionType::ClientInvite, TransactionTimer::TimerM(_)) => { + self.transition(TransactionState::Terminated)?; + } + (TransactionType::ServerInvite, TransactionTimer::TimerG(key, duration)) => { + if let Some(last_response) = &self.last_response { + if let Some(connection) = &self.connection { + let last_response = if let Some(ref inspector) = + self.endpoint_inner.message_inspector + { + inspector.before_send( + last_response.to_owned().into(), + self.destination.as_ref(), + ) + } else { + last_response.to_owned().into() + }; + connection + .send(last_response, self.destination.as_ref()) + .await?; + } + } + // restart Timer G, doubling up to T2 (RFC 3261 §13.3.1.4) + let duration = (*duration * 2).min(self.endpoint_inner.option.t2); + let timer_g = self + .endpoint_inner + .timers + .timeout(duration, TransactionTimer::TimerG(key.clone(), duration)); + self.timer_g.replace(timer_g); + } + ( + TransactionType::ClientNonInvite + | TransactionType::ServerNonInvite + | TransactionType::ClientInvite, + TransactionTimer::TimerG(_, _), + ) => { + warn!( + key = %self.key, + tx_type = %self.transaction_type, + "Timer G fired outside the server Accepted state; ignoring", + ); + } + (_, TransactionTimer::TimerL(_) | TransactionTimer::TimerM(_)) => { + warn!( + key = %self.key, + tx_type = %self.transaction_type, + "RFC 6026 Accepted-state timer fired with mismatched transaction type", + ); + } + (_, TransactionTimer::TimerA(_, _)) + | (_, TransactionTimer::TimerB(_)) + | (_, TransactionTimer::TimerC(_)) + | (_, TransactionTimer::TimerD(_)) + | (_, TransactionTimer::TimerK(_)) + | (_, TransactionTimer::TimerCleanup(_)) => {} + } + } TransactionState::Confirmed => { if let TransactionTimer::TimerK(_) = timer { self.transition(TransactionState::Terminated)?; @@ -1030,6 +1324,104 @@ impl Transaction { } } } + TransactionState::Accepted => { + self.timer_a + .take() + .map(|id| self.endpoint_inner.timers.cancel(id)); + self.timer_b + .take() + .map(|id| self.endpoint_inner.timers.cancel(id)); + self.timer_c + .take() + .map(|id| self.endpoint_inner.timers.cancel(id)); + + match self.transaction_type { + TransactionType::ServerInvite => { + // RFC 6026 §7.1: server INVITE 2xx-Accepted entry. + // + // Start Timer L (64*T1). On expiry the transaction + // transitions to Terminated (handled in on_timer for + // the Accepted state). + // + // Register the dialog in `waiting_ack` so the dialog + // layer can route the ACK for this 2xx back to this + // transaction key. + // + // Timer G deviation (documented in lib.rs): strict + // §7.1 leaves 2xx retransmission to the TU; the + // rsipstack dialog layer does not retransmit, so the + // transaction keeps Timer G (T1, doubling to T2) on + // every transport — the wire behavior RFC 3261 + // §13.3.1.4 requires — until the ACK or Timer L. + let timer_l = self.endpoint_inner.timers.timeout( + self.endpoint_inner.option.t1x64, + TransactionTimer::TimerL(self.key.clone()), + ); + self.timer_l.replace(timer_l); + + let timer_g = self.endpoint_inner.timers.timeout( + self.endpoint_inner.option.t1, + TransactionTimer::TimerG( + self.key.clone(), + self.endpoint_inner.option.t1, + ), + ); + self.timer_g.replace(timer_g); + + if let Some(ref resp) = self.last_response { + let dialog_id = DialogId::try_from((resp, TransactionRole::Server))?; + self.endpoint_inner + .waiting_ack + .insert(dialog_id, self.key.clone()); + } + debug!( + key = %self.key, + "entered Accepted state (server); Timer L and Timer G armed, waiting for ACK" + ); + } + TransactionType::ClientInvite => { + // RFC 6026 §7.2: client INVITE 2xx-Accepted entry. + // + // Start Timer M (64*T1). On expiry the transaction + // transitions to Terminated. While in Accepted, the + // client absorbs server-retransmitted 2xx responses + // (these are forwarded to the TU as duplicates and + // ignored by the transaction state machine itself — + // see Accepted-self-loop edge in can_transition). + // + // Strict RFC 3261 §17.1.1.3 + RFC 6026 §7.2 make + // the ACK for 2xx the TU's responsibility. For + // rsipstack 0.5.x backward compatibility, the + // auto_ack_client_invite gate at the tail of + // on_received_response still emits that ACK after + // entering Accepted; see lib.rs Standards Compliance + // for the disclaimer. 3xx-6xx ACKs continue to be + // sent by the transaction layer through the + // Completed → Terminated path. + let timer_m = self.endpoint_inner.timers.timeout( + self.endpoint_inner.option.t1x64, + TransactionTimer::TimerM(self.key.clone()), + ); + self.timer_m.replace(timer_m); + debug!( + key = %self.key, + "entered Accepted state (client); Timer M armed, awaiting expiry or 2xx retransmits" + ); + } + _ => { + // Non-INVITE transactions never reach Accepted per + // RFC 6026 §4. can_transition() should have already + // rejected this; treat as a programming error. + return Err(Error::TransactionError( + format!( + "Accepted state is INVITE-only (transaction type was {})", + self.transaction_type + ), + self.key.clone(), + )); + } + } + } TransactionState::Completed => { self.timer_a.take().map(|id| -> Option { self.endpoint_inner.timers.cancel(id) @@ -1136,6 +1528,12 @@ impl Transaction { self.timer_g .take() .map(|id| -> Option { self.endpoint_inner.timers.cancel(id) }); + self.timer_l + .take() + .map(|id| -> Option { self.endpoint_inner.timers.cancel(id) }); + self.timer_m + .take() + .map(|id| -> Option { self.endpoint_inner.timers.cancel(id) }); } pub fn role(&self) -> TransactionRole {