sipx_call/
extension.rs

1//! Application-owned requests inside an established dialog.
2//!
3//! The state-machine-owned methods stay in [`Call::handle`](crate::Call::handle). This module
4//! owns only the bounded request snapshot and the exactly-once response capability described by
5//! `docs/specs/dialog-extensions.md`.
6
7use std::future::Future;
8use std::sync::Arc;
9use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
10use std::time::Duration;
11
12use bytes::Bytes;
13use sipx_sip::build::ResponseBuilder;
14use sipx_sip::transaction::TransactionKey;
15use sipx_sip::{Header, HeaderName, Headers, Method, Request, StatusCode};
16use sipx_transport::Handle;
17use tokio::sync::Notify;
18
19use crate::error::{Error, Result};
20
21/// Largest body the application-owned dialog path will retain.
22pub const MAX_APPLICATION_BODY: usize = 64 * 1024;
23
24/// How long an application may retain an unanswered server transaction.
25const RESPONSE_DEADLINE: Duration = Duration::from_secs(32);
26
27/// One application-owned request received inside a live dialog.
28///
29/// Clones share one exactly-once response capability. Dropping the final unanswered clone produces
30/// a bounded `500` refusal.
31#[derive(Debug, Clone)]
32pub struct ApplicationRequest {
33    method: Method,
34    headers: Headers,
35    body: Bytes,
36    response: ResponseCapability,
37}
38
39impl ApplicationRequest {
40    pub(crate) fn new(endpoint: Handle, key: TransactionKey, request: &Request) -> Result<Self> {
41        let response = ResponseCapability::new(endpoint, key, request.clone())?;
42        Ok(Self {
43            method: request.method.clone(),
44            headers: request.headers.clone(),
45            body: Bytes::copy_from_slice(request.body()),
46            response,
47        })
48    }
49
50    /// The admitted request method.
51    #[must_use]
52    pub fn method(&self) -> &Method {
53        &self.method
54    }
55
56    /// The complete parser-validated header collection, in wire order.
57    #[must_use]
58    pub fn headers(&self) -> &Headers {
59        &self.headers
60    }
61
62    /// The bounded request body.
63    #[must_use]
64    pub fn body(&self) -> &[u8] {
65        &self.body
66    }
67
68    /// Send the one final response owned by this request.
69    ///
70    /// Dialog and framing headers are copied or generated by the stack and cannot be supplied in
71    /// `headers`. A body requires an application-supplied `Content-Type`.
72    pub async fn respond(
73        self,
74        status: StatusCode,
75        reason: impl Into<Bytes>,
76        headers: &[Header],
77        body: Bytes,
78    ) -> Result<()> {
79        validate_response(status, headers, &body)?;
80        self.response
81            .respond(status, reason.into(), headers, body)
82            .await
83    }
84}
85
86#[derive(Debug)]
87struct ResponseState {
88    claimed: AtomicBool,
89    owners: AtomicUsize,
90    completed: Notify,
91    endpoint: Handle,
92    key: TransactionKey,
93    request: Request,
94    runtime: tokio::runtime::Handle,
95}
96
97impl ResponseState {
98    fn claim(&self) -> bool {
99        self.claimed
100            .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
101            .is_ok()
102    }
103}
104
105#[derive(Debug)]
106struct ResponseCapability {
107    state: Arc<ResponseState>,
108}
109
110impl ResponseCapability {
111    fn new(endpoint: Handle, key: TransactionKey, request: Request) -> Result<Self> {
112        let runtime = tokio::runtime::Handle::try_current()
113            .map_err(|_| Error::ApplicationRuntimeUnavailable)?;
114        let state = Arc::new(ResponseState {
115            claimed: AtomicBool::new(false),
116            owners: AtomicUsize::new(1),
117            completed: Notify::new(),
118            endpoint,
119            key,
120            request,
121            runtime: runtime.clone(),
122        });
123        let deadline = Arc::clone(&state);
124        runtime.spawn(async move {
125            // A bound on failure: an abandoned application must not retain a server transaction.
126            if tokio::time::timeout(RESPONSE_DEADLINE, deadline.completed.notified())
127                .await
128                .is_err()
129                && deadline.claim()
130            {
131                send_fallback(&deadline, 504, "Server Time-out").await;
132            }
133        });
134        Ok(Self { state })
135    }
136
137    async fn respond(
138        &self,
139        status: StatusCode,
140        reason: Bytes,
141        headers: &[Header],
142        body: Bytes,
143    ) -> Result<()> {
144        let mut builder = ResponseBuilder::to_request(&self.state.request, status, reason)?;
145        for header in headers {
146            builder = builder.header(
147                header.name().clone(),
148                Bytes::copy_from_slice(header.raw_value()),
149            )?;
150        }
151        let response = builder.body(body).build();
152        if !self.state.claim() {
153            return Err(Error::ApplicationResponseAlreadySent);
154        }
155        // The final response is committed before I/O. A stream write may fail after sending a
156        // prefix—or the complete response—so reopening the capability on transport error would
157        // permit a contradictory second final response on the same server transaction. The
158        // transport error is returned, but exactly-once ownership remains spent.
159        self.state.completed.notify_one();
160        let endpoint = self.state.endpoint.clone();
161        let key = self.state.key.clone();
162        await_committed_send(self.state.runtime.clone(), async move {
163            endpoint.respond(&key, response).await
164        })
165        .await
166    }
167}
168
169async fn await_committed_send<F>(runtime: tokio::runtime::Handle, send: F) -> Result<()>
170where
171    F: Future<Output = sipx_transport::Result<()>> + Send + 'static,
172{
173    // There is no await between claiming the capability and spawning this task. The runtime was
174    // captured when the request became an event, so an application may poll `respond` from a thread
175    // without an entered Tokio context. Once spawned, the transport operation owns the selected
176    // response and survives cancellation of the application future that only waits for its result.
177    runtime
178        .spawn(send)
179        .await
180        .map_err(|_| sipx_transport::Error::EndpointClosed)??;
181    Ok(())
182}
183
184// `fetch_update` is the name available at the workspace MSRV; current nightly calls it deprecated
185// before the replacement is available on that supported toolchain.
186#[allow(deprecated)]
187impl Clone for ResponseCapability {
188    fn clone(&self) -> Self {
189        // Saturation keeps cloning panic-free. Reaching usize::MAX live application handles is
190        // outside any addressable process, but the counter remains well-defined even there.
191        // discard: saturation is the defined clone result; no fallible state transition is lost.
192        let _ = self
193            .state
194            .owners
195            .fetch_update(Ordering::AcqRel, Ordering::Acquire, |owners| {
196                Some(owners.saturating_add(1))
197            });
198        Self {
199            state: Arc::clone(&self.state),
200        }
201    }
202}
203
204// Keep the same MSRV/nightly compatibility bridge as the clone-side owner update above.
205#[allow(deprecated)]
206impl Drop for ResponseCapability {
207    fn drop(&mut self) {
208        let was_last = self
209            .state
210            .owners
211            .fetch_update(Ordering::AcqRel, Ordering::Acquire, |owners| {
212                Some(owners.saturating_sub(1))
213            })
214            .is_ok_and(|owners| owners == 1);
215        if !was_last {
216            return;
217        }
218        if !self.state.claim() {
219            return;
220        }
221        self.state.completed.notify_one();
222        let state = Arc::clone(&self.state);
223        self.state.runtime.spawn(async move {
224            send_fallback(&state, 500, "Server Internal Error").await;
225        });
226    }
227}
228
229async fn send_fallback(state: &ResponseState, status: u16, reason: &'static str) {
230    let Some(status) = StatusCode::new(status) else {
231        return;
232    };
233    let Ok(builder) = ResponseBuilder::to_request(&state.request, status, reason) else {
234        return;
235    };
236    // discard: the capability is already being abandoned; the transport records an unsent
237    // response, and there is no remaining owner to receive this error.
238    let _ = state.endpoint.respond(&state.key, builder.build()).await;
239}
240
241pub(crate) fn application_owned(method: &Method, admitted: &[Bytes]) -> bool {
242    match method {
243        Method::Info | Method::Message => true,
244        Method::Other(token) => {
245            matches!(Method::parse(token), Method::Other(_))
246                && admitted.iter().any(|known| known == token)
247        }
248        _ => false,
249    }
250}
251
252pub(crate) fn validate_method_for_admission(method: &Method) -> Result<Bytes> {
253    match method {
254        Method::Other(token) => match Method::parse(token) {
255            Method::Other(_) => Ok(token.clone()),
256            known => Err(Error::StackOwnedDialogMethod(known)),
257        },
258        _ => Err(Error::StackOwnedDialogMethod(method.clone())),
259    }
260}
261
262pub(crate) fn validate_request_parts(headers: &[Header], body: &[u8]) -> Result<()> {
263    if body.len() > MAX_APPLICATION_BODY {
264        return Err(Error::ApplicationBodyTooLarge {
265            actual: body.len(),
266            limit: MAX_APPLICATION_BODY,
267        });
268    }
269    for header in headers {
270        if protected_header(header.name()) {
271            return Err(Error::ProtectedApplicationHeader(header.name().clone()));
272        }
273    }
274    if !body.is_empty()
275        && !headers
276            .iter()
277            .any(|header| header.name() == &HeaderName::ContentType)
278    {
279        return Err(Error::ApplicationContentTypeRequired);
280    }
281    Ok(())
282}
283
284fn validate_response(status: StatusCode, headers: &[Header], body: &[u8]) -> Result<()> {
285    if !status.is_final() {
286        return Err(Error::ApplicationFinalResponseRequired(status.code()));
287    }
288    validate_request_parts(headers, body)
289}
290
291fn protected_header(name: &HeaderName) -> bool {
292    matches!(
293        name,
294        HeaderName::Via
295            | HeaderName::Route
296            | HeaderName::RecordRoute
297            | HeaderName::From
298            | HeaderName::To
299            | HeaderName::CallId
300            | HeaderName::CSeq
301            | HeaderName::MaxForwards
302            | HeaderName::ContentLength
303            | HeaderName::Authorization
304            | HeaderName::ProxyAuthorization
305    )
306}
307
308#[cfg(test)]
309#[allow(
310    clippy::unwrap_used,
311    clippy::expect_used,
312    clippy::panic,
313    clippy::indexing_slicing
314)]
315mod tests {
316    use std::sync::{Arc, mpsc};
317    use std::task::{Context, Poll, Waker};
318
319    use tokio::sync::{Notify, oneshot};
320
321    use super::*;
322
323    #[test]
324    fn only_info_message_and_admitted_private_tokens_are_application_owned() {
325        let private = Bytes::from_static(b"PRIVATE");
326        let admitted = [private.clone()];
327        assert!(application_owned(&Method::Info, &admitted));
328        assert!(application_owned(&Method::Message, &admitted));
329        assert!(application_owned(&Method::Other(private), &admitted));
330        assert!(!application_owned(
331            &Method::Other(Bytes::from_static(b"private")),
332            &admitted
333        ));
334        for method in [
335            Method::Options,
336            Method::Bye,
337            Method::Invite,
338            Method::Update,
339            Method::Refer,
340            Method::Notify,
341        ] {
342            assert!(!application_owned(&method, &admitted), "{method}");
343        }
344    }
345
346    #[test]
347    fn canonical_known_tokens_cannot_be_admitted_as_other_methods() {
348        for (token, known) in [
349            (Bytes::from_static(b"INVITE"), Method::Invite),
350            (Bytes::from_static(b"ACK"), Method::Ack),
351            (Bytes::from_static(b"BYE"), Method::Bye),
352            (Bytes::from_static(b"CANCEL"), Method::Cancel),
353            (Bytes::from_static(b"REGISTER"), Method::Register),
354            (Bytes::from_static(b"OPTIONS"), Method::Options),
355            (Bytes::from_static(b"INFO"), Method::Info),
356            (Bytes::from_static(b"PRACK"), Method::Prack),
357            (Bytes::from_static(b"UPDATE"), Method::Update),
358            (Bytes::from_static(b"SUBSCRIBE"), Method::Subscribe),
359            (Bytes::from_static(b"NOTIFY"), Method::Notify),
360            (Bytes::from_static(b"REFER"), Method::Refer),
361            (Bytes::from_static(b"MESSAGE"), Method::Message),
362            (Bytes::from_static(b"PUBLISH"), Method::Publish),
363        ] {
364            let alias = Method::Other(token.clone());
365            assert!(matches!(
366                validate_method_for_admission(&alias),
367                Err(Error::StackOwnedDialogMethod(method)) if method == known
368            ));
369            assert!(!application_owned(&alias, &[token]));
370        }
371    }
372
373    #[tokio::test]
374    async fn a_committed_send_survives_cancellation_of_its_waiter() {
375        let started = Arc::new(Notify::new());
376        let release = Arc::new(Notify::new());
377        let (completed, completion) = oneshot::channel();
378        let send_started = Arc::clone(&started);
379        let send_release = Arc::clone(&release);
380
381        let waiter = tokio::spawn(await_committed_send(
382            tokio::runtime::Handle::current(),
383            async move {
384                send_started.notify_one();
385                send_release.notified().await;
386                let _ = completed.send(());
387                Ok(())
388            },
389        ));
390        started.notified().await;
391        waiter.abort();
392        let cancelled = waiter
393            .await
394            .expect_err("the application waiter was cancelled");
395        assert!(cancelled.is_cancelled());
396
397        release.notify_one();
398        tokio::time::timeout(Duration::from_secs(1), completion)
399            .await
400            .expect("a bound on failure waiting for the owned send")
401            .expect("the owned send completes after waiter cancellation");
402    }
403
404    #[test]
405    fn a_committed_send_uses_its_captured_runtime_outside_runtime_context() {
406        let runtime = tokio::runtime::Builder::new_multi_thread()
407            .worker_threads(1)
408            .enable_all()
409            .build()
410            .expect("runtime builds");
411        let handle = runtime.handle().clone();
412        assert!(
413            tokio::runtime::Handle::try_current().is_err(),
414            "the regression requires no entered Tokio runtime"
415        );
416        let (completed, completion) = mpsc::channel();
417        let mut send = Box::pin(await_committed_send(handle, async move {
418            completed.send(()).expect("observer remains live");
419            Ok(())
420        }));
421
422        let waker = Waker::noop();
423        let mut context = Context::from_waker(waker);
424        let first = send.as_mut().poll(&mut context);
425        completion
426            .recv_timeout(Duration::from_secs(1))
427            .expect("a bound on failure waiting for the captured runtime");
428        match first {
429            Poll::Ready(result) => result.expect("send succeeds"),
430            Poll::Pending => runtime
431                .block_on(send.as_mut())
432                .expect("join completes on the captured runtime"),
433        }
434    }
435
436    #[test]
437    fn protected_headers_are_rejected_before_a_request_can_be_sent() {
438        let header = Header::build(HeaderName::CSeq, "99 MESSAGE").expect("valid header syntax");
439        assert!(matches!(
440            validate_request_parts(&[header], &[]),
441            Err(Error::ProtectedApplicationHeader(HeaderName::CSeq))
442        ));
443    }
444
445    #[test]
446    fn body_limit_and_content_type_are_checked_without_partial_send() {
447        assert!(matches!(
448            validate_request_parts(&[], b"hello"),
449            Err(Error::ApplicationContentTypeRequired)
450        ));
451        let content_type =
452            Header::build(HeaderName::ContentType, "text/plain").expect("valid header syntax");
453        let oversized = vec![0; MAX_APPLICATION_BODY + 1];
454        assert!(matches!(
455            validate_request_parts(&[content_type], &oversized),
456            Err(Error::ApplicationBodyTooLarge { .. })
457        ));
458    }
459}
sipx_call/extension.rs

sipx_call/
extension.rs
