diff --git a/src/transaction/tests/mod.rs b/src/transaction/tests/mod.rs index 54e977d0..f215e487 100644 --- a/src/transaction/tests/mod.rs +++ b/src/transaction/tests/mod.rs @@ -11,6 +11,7 @@ mod test_endpoint; mod test_provisional_responses; mod test_server; mod test_server_invite_drop; +mod test_stream_reconnect; mod test_transaction_states; pub(super) async fn create_test_endpoint(addr: Option<&str>) -> Result { diff --git a/src/transaction/tests/test_stream_reconnect.rs b/src/transaction/tests/test_stream_reconnect.rs new file mode 100644 index 00000000..a291433d --- /dev/null +++ b/src/transaction/tests/test_stream_reconnect.rs @@ -0,0 +1,271 @@ +//! A stream (TCP) connection that the peer closed must not be reused for +//! later requests: the next request has to open a fresh connection instead +//! of being written to the dead one and waiting out Timer B/F. +use crate::sip::{headers::*, Method, SipMessage, StatusCode}; +use crate::transaction::endpoint::EndpointOption; +use crate::transaction::key::{TransactionKey, TransactionRole}; +use crate::transaction::transaction::Transaction; +use crate::transaction::EndpointBuilder; +use crate::transport::tcp::TcpConnection; +use crate::transport::{udp::UdpConnection, SipAddr, SipConnection, TransportLayer}; +use crate::Result; +use std::convert::TryFrom; +use std::time::Duration; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::{TcpListener, TcpStream}; +use tokio_util::sync::CancellationToken; + +/// Read one SIP request (no body) from a raw stream. +async fn read_request(stream: &mut TcpStream) -> crate::sip::Request { + let mut buf = Vec::new(); + let mut chunk = [0u8; 2048]; + while !buf.windows(4).any(|w| w == b"\r\n\r\n") { + let n = stream.read(&mut chunk).await.expect("read request"); + assert!(n > 0, "peer closed before a full request"); + buf.extend_from_slice(&chunk[..n]); + } + let raw = String::from_utf8_lossy(&buf).to_string(); + match SipMessage::try_from(raw.as_str()).expect("parse request") { + SipMessage::Request(req) => req, + _ => panic!("expected a request"), + } +} + +/// Answer `req` with a 200 OK on a raw stream. +async fn answer(stream: &mut TcpStream, req: &crate::sip::Request) { + let mut resp = String::from("SIP/2.0 200 OK\r\n"); + for header in req.headers.iter() { + match header { + crate::sip::Header::Via(_) + | crate::sip::Header::From(_) + | crate::sip::Header::To(_) + | crate::sip::Header::CallId(_) + | crate::sip::Header::CSeq(_) => { + resp.push_str(&header.to_string()); + resp.push_str("\r\n"); + } + _ => {} + } + } + resp.push_str("Content-Length: 0\r\n\r\n"); + stream + .write_all(resp.as_bytes()) + .await + .expect("write response"); +} + +async fn answer_one(stream: &mut TcpStream) { + let req = read_request(stream).await; + answer(stream, &req).await; +} + +fn make_options(peer: std::net::SocketAddr, cseq: u32) -> Result { + make_request(Method::Options, peer, cseq) +} + +fn make_request( + method: Method, + peer: std::net::SocketAddr, + cseq: u32, +) -> Result { + let uri = crate::sip::Uri::try_from(format!("sip:{};transport=tcp", peer).as_str())?; + Ok(crate::sip::Request { + method, + uri, + headers: vec![ + Via::new(format!( + "SIP/2.0/TCP 127.0.0.1:5060;branch=z9hG4bK-reconnect-{cseq}" + )) + .into(), + CSeq::new(format!("{cseq} {method}")).into(), + From::new(";tag=reconnect").into(), + To::new("").into(), + CallId::new("stream-reconnect@127.0.0.1").into(), + MaxForwards::new("70").into(), + ] + .into(), + version: crate::sip::Version::V2, + body: vec![], + }) +} + +async fn final_status(tx: &mut Transaction) -> Option { + while let Some(msg) = tx.receive().await { + if let SipMessage::Response(resp) = msg { + if resp.status_code.kind() != crate::sip::StatusCodeKind::Provisional { + return Some(resp.status_code); + } + } + } + None +} + +async fn build_endpoint(t1: Duration) -> Result { + let token = CancellationToken::new(); + let tl = TransportLayer::new(token.child_token()); + let udp = UdpConnection::create_connection("127.0.0.1:0".parse()?, None, None).await?; + tl.add_transport(udp.into()); + let option = EndpointOption { + t1, + t1x64: Duration::from_secs(8), + ..Default::default() + }; + Ok(EndpointBuilder::new() + .with_user_agent("rsipstack-test") + .with_transport_layer(tl) + .with_option(option) + .build()) +} + +/// The peer idle-closes the connection the first request opened. A later +/// request to the same peer must be sent over a new connection. +#[tokio::test] +async fn test_request_after_peer_closed_stream_uses_new_connection() -> Result<()> { + // T1 is long so a Timer A redial cannot hide a stale cached connection: + // the request has to go out on a new connection straight away. + let endpoint = build_endpoint(Duration::from_secs(3)).await?; + let listener = TcpListener::bind("127.0.0.1:0").await?; + let peer = listener.local_addr()?; + + let peer_task = tokio::spawn(async move { + let (mut first, _) = listener.accept().await.expect("accept first"); + answer_one(&mut first).await; + drop(first); // the peer closes the flow after the first transaction + let (mut second, _) = listener.accept().await.expect("accept second"); + answer_one(&mut second).await; + tokio::time::sleep(Duration::from_secs(10)).await; + }); + + let inner = endpoint.inner.clone(); + let client = async move { + let req = make_options(peer, 1)?; + let key = TransactionKey::from_request(&req, TransactionRole::Client)?; + let mut tx = Transaction::new_client(key, req, inner.clone(), None); + tx.send().await?; + assert_eq!(final_status(&mut tx).await, Some(StatusCode::OK)); + let first_connection = tx.connection.clone().expect("first connection"); + drop(tx); + + // The transport layer sees the peer's FIN and marks the flow + // terminated, so flow affinity (RFC 5626) no longer picks it. + let token = first_connection.cancel_token().expect("stream token"); + tokio::time::timeout(Duration::from_secs(1), token.cancelled()) + .await + .expect("a flow the peer closed must be marked terminated"); + + let req = make_options(peer, 2)?; + let key = TransactionKey::from_request(&req, TransactionRole::Client)?; + let mut tx = Transaction::new_client(key, req, inner.clone(), None); + tx.send().await?; + let status = tokio::time::timeout(Duration::from_millis(1500), final_status(&mut tx)) + .await + .unwrap_or(None); + Ok::<_, crate::Error>(status) + }; + + let status = tokio::select! { + r = client => r?, + _ = endpoint.serve() => panic!("endpoint stopped"), + }; + peer_task.abort(); + assert_eq!( + status, + Some(StatusCode::OK), + "the request after the peer closed the TCP connection must be sent on a new \ + connection and answered, not written to the closed one" + ); + Ok(()) +} + +/// A client transaction bound to a stream connection whose write fails must +/// report the failure to the TU at once (a local 503, RFC 3261 §17.1.4 / +/// §8.1.3.1) instead of waiting for Timer B/F: nothing retransmits a request +/// on a reliable transport. +#[tokio::test] +async fn test_send_failure_on_stream_is_reported_at_once() -> Result<()> { + for method in [Method::Options, Method::Invite] { + let endpoint = build_endpoint(Duration::from_millis(500)).await?; + let listener = TcpListener::bind("127.0.0.1:0").await?; + let peer = listener.local_addr()?; + let peer_task = tokio::spawn(async move { + let (_first, _) = listener.accept().await.expect("accept"); + tokio::time::sleep(Duration::from_secs(10)).await; + }); + + let target = SipAddr::try_from(&crate::sip::Uri::try_from( + format!("sip:{};transport=tcp", peer).as_str(), + )?)?; + let dead_token = CancellationToken::new(); + let dead = + SipConnection::Tcp(TcpConnection::connect(&target, Some(dead_token.clone())).await?); + // A flow whose local side is already shut down: every write fails. + dead.close().await.ok(); + + let inner = endpoint.inner.clone(); + let request = make_request(method, peer, 1)?; + let client = async move { + let key = TransactionKey::from_request(&request, TransactionRole::Client)?; + let mut tx = Transaction::new_client(key, request, inner.clone(), Some(dead)); + tx.send().await?; + let status = tokio::time::timeout(Duration::from_secs(2), final_status(&mut tx)) + .await + .unwrap_or(None); + Ok::<_, crate::Error>(status) + }; + + let status = tokio::select! { + r = client => r?, + _ = endpoint.serve() => panic!("endpoint stopped"), + }; + peer_task.abort(); + assert_eq!( + status, + Some(StatusCode::ServiceUnavailable), + "a failed {method} write on a stream connection must be reported to the TU, \ + not left to time out" + ); + assert!( + dead_token.is_cancelled(), + "the connection whose {method} write failed must be retired" + ); + } + Ok(()) +} + +/// Retiring a dead connection must not drop a newer connection cached for the +/// same peer. +#[tokio::test] +async fn test_retire_keeps_newer_connection_to_same_peer() -> Result<()> { + let endpoint = build_endpoint(Duration::from_millis(500)).await?; + let listener = TcpListener::bind("127.0.0.1:0").await?; + let peer = listener.local_addr()?; + let peer_task = tokio::spawn(async move { + let mut held = Vec::new(); + while let Ok((stream, _)) = listener.accept().await { + held.push(stream); + } + }); + let target = SipAddr::try_from(&crate::sip::Uri::try_from( + format!("sip:{};transport=tcp", peer).as_str(), + )?)?; + let old = SipConnection::Tcp(TcpConnection::connect(&target, None).await?); + let newer = SipConnection::Tcp(TcpConnection::connect(&target, None).await?); + let transport_layer = &endpoint.inner.transport_layer; + transport_layer.add_connection(newer.clone()); + + transport_layer.retire_connection(&old); + let (cached, _) = transport_layer.lookup(&target, None).await?; + assert!( + cached.is_same_stream(&newer), + "retiring an old connection must keep the newer one" + ); + + transport_layer.retire_connection(&newer); + let (fresh, _) = transport_layer.lookup(&target, None).await?; + assert!( + !fresh.is_same_stream(&newer), + "a retired connection must not be returned by lookup" + ); + peer_task.abort(); + Ok(()) +} diff --git a/src/transaction/transaction.rs b/src/transaction/transaction.rs index 99f0638a..e90b352a 100644 --- a/src/transaction/transaction.rs +++ b/src/transaction/transaction.rs @@ -318,9 +318,11 @@ impl Transaction { // Try to send if we have a connection; log errors instead of returning // so the transaction always enters the state machine (Calling) and // timers (Timer A / Timer B) handle retries and timeouts. + let mut stream_failed = false; if let Some(connection) = self.connection.as_ref() { if let Err(e) = connection.send(message, self.destination.as_ref()).await { warn!(key = %self.key, error = %e, "send failed"); + stream_failed = connection.is_stream(); } } else { debug!(key = %self.key, "no connection, will retry on timer"); @@ -329,7 +331,29 @@ impl Transaction { // Always transition to Calling — the transaction enters the state machine // even when transport is unavailable. Timer A will retry the send (and // redo the transport lookup if needed), and Timer B handles timeout. - self.transition(TransactionState::Calling).map(|_| ()) + self.transition(TransactionState::Calling)?; + if stream_failed { + self.on_stream_send_failure()?; + } + Ok(()) + } + + /// A write on a stream connection (TCP/TLS/WS) failed. Nothing will ever + /// retransmit the request on it (no Timer A on reliable transports), so + /// instead of leaving the TU to wait for Timer B/F: retire the dead + /// connection, so later requests to the peer do not pick it again, and + /// report the failure to the TU as a local 503 (RFC 3261 §17.1.4, + /// §8.1.3.1). + fn on_stream_send_failure(&mut self) -> Result<()> { + if let Some(connection) = self.connection.as_ref() { + self.endpoint_inner + .transport_layer + .retire_connection(connection); + } + let response = + self.endpoint_inner + .make_response(&self.original, StatusCode::ServiceUnavailable, None); + self.inform_tu_response(response) } /// Resolve the target URI for sending a request. @@ -843,6 +867,9 @@ impl Transaction { .await { warn!(key = %self.key, error = %e, "timer A resend failed"); + if connection.is_stream() { + return self.on_stream_send_failure(); + } } } else { debug!(key = %self.key, "timer A: no connection yet"); diff --git a/src/transport/connection.rs b/src/transport/connection.rs index 3151e41b..c46a998b 100644 --- a/src/transport/connection.rs +++ b/src/transport/connection.rs @@ -199,6 +199,37 @@ impl SipConnection { !matches!(self, SipConnection::Udp(_)) } + /// Whether this is a connection-oriented stream (TCP, TLS or WebSocket). + pub(crate) fn is_stream(&self) -> bool { + match self { + #[cfg(feature = "platform-tokio")] + SipConnection::Tcp(_) => true, + #[cfg(all(feature = "platform-tokio", feature = "rustls"))] + SipConnection::Tls(_) => true, + #[cfg(all(feature = "platform-tokio", feature = "websocket"))] + SipConnection::WebSocket(_) => true, + _ => false, + } + } + + /// Whether `self` and `other` are handles to the same stream (TCP, TLS or + /// WebSocket) connection. Always false for other connection types. + pub(crate) fn is_same_stream(&self, other: &SipConnection) -> bool { + match (self, other) { + #[cfg(feature = "platform-tokio")] + (SipConnection::Tcp(a), SipConnection::Tcp(b)) => { + std::sync::Arc::ptr_eq(&a.inner, &b.inner) + } + #[cfg(all(feature = "platform-tokio", feature = "rustls"))] + (SipConnection::Tls(a), SipConnection::Tls(b)) => a.ptr_eq(b), + #[cfg(all(feature = "platform-tokio", feature = "websocket"))] + (SipConnection::WebSocket(a), SipConnection::WebSocket(b)) => { + std::sync::Arc::ptr_eq(&a.inner, &b.inner) + } + _ => false, + } + } + pub fn cancel_token(&self) -> Option { match self { SipConnection::Channel(transport) => transport.cancel_token(), diff --git a/src/transport/tls.rs b/src/transport/tls.rs index f5079413..084b33ac 100644 --- a/src/transport/tls.rs +++ b/src/transport/tls.rs @@ -464,6 +464,14 @@ enum TlsConnectionInner { } impl TlsConnection { + pub(crate) fn ptr_eq(&self, other: &TlsConnection) -> bool { + match (&self.inner, &other.inner) { + (TlsConnectionInner::Client(a), TlsConnectionInner::Client(b)) => Arc::ptr_eq(a, b), + (TlsConnectionInner::Server(a), TlsConnectionInner::Server(b)) => Arc::ptr_eq(a, b), + _ => false, + } + } + // Connect to a remote TLS server pub async fn connect( remote_addr: &SipAddr, diff --git a/src/transport/transport_layer.rs b/src/transport/transport_layer.rs index ba7ab337..af4a7d50 100644 --- a/src/transport/transport_layer.rs +++ b/src/transport/transport_layer.rs @@ -204,6 +204,14 @@ impl TransportLayer { self.inner.del_connection(addr) } + /// Retire a stream connection that can no longer carry messages: the next + /// lookup for its peer opens a new one (a newer connection cached for the + /// same peer is left alone), and its cancel token is cancelled so flow + /// affinity stops choosing it. + pub(crate) fn retire_connection(&self, connection: &SipConnection) { + retire_connection(&self.inner.connections, connection) + } + pub async fn lookup( &self, target: &SipAddr, @@ -471,6 +479,7 @@ impl TransportLayerInner { .map(|addr| addr.to_string()) .unwrap_or_else(|| "-".to_string()); info!(addr=%transport.get_addr(), remote=%remote_addr, "serve_connection: starting serve_loop"); + let connections = self.connections.clone(); crate::platform::spawn(async move { match sender_clone.send(TransportEvent::New(transport.clone())) { Ok(()) => { @@ -495,11 +504,35 @@ impl TransportLayerInner { } } info!(addr=%transport.get_addr(), remote=%remote_addr, "transport serve_loop exited"); + // The connection is gone: lookup must not hand it out again. + retire_connection(&connections, &transport); transport.close().await.ok(); sender_clone.send(TransportEvent::Closed(transport)).ok(); }); } } +fn retire_connection(connections: &RwMap, connection: &SipConnection) { + if !connection.is_stream() { + return; + } + // The token is the connection's own: cancelling it marks the flow as + // terminated for flow affinity (see `resolve_affinity_connection`). + if let Some(token) = connection.cancel_token() { + token.cancel(); + } + if let Some(remote_addr) = connection.get_remote_addr() { + let removed = connections.with_mut(|cached| { + let same = cached + .get(remote_addr) + .is_some_and(|c| c.is_same_stream(connection)); + same && cached.remove(remote_addr).is_some() + }); + if removed { + debug!(%remote_addr, "retire connection"); + } + } +} + impl Drop for TransportLayer { fn drop(&mut self) { self.inner.cancel_token.cancel();