sipx_call/
notifier.rs

1//! The socket driver for the RFC 6665 notifier state machine.
2//!
3//! [`sipx_ua::subscribe::Subscriptions`] remains the only protocol store. This module adds the
4//! dialog target, package document and owned expiry task that only a live endpoint can supply.
5//! The decision table and lifetime rules are in
6//! `docs/specs/event-notifier.md`.
7
8use std::collections::HashMap;
9use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
10use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
11use std::time::Duration;
12
13use bytes::Bytes;
14use sipx_sip::build::{RequestBuilder, ResponseBuilder};
15use sipx_sip::event::{Packages, Reason, Subscription};
16use sipx_sip::headers::{CSeq, Contact, Expires, From as FromHeader, To};
17use sipx_sip::{HeaderName, Method, Request, StatusCode};
18use sipx_transport::{Handle, Incoming, Target};
19use sipx_ua::packages::{DIALOG_INFO_TYPE, DialogWatch, REGINFO_TYPE, RegistrationWatch};
20use sipx_ua::presence::{PIDF_TYPE, Pidf};
21use sipx_ua::subscribe::{Answer, Id, Subscriptions};
22use tokio::sync::watch;
23use tokio::task::JoinHandle;
24use tokio::time::Instant;
25
26use crate::call::{add_routes, contact_for, in_dialog_target, token};
27use crate::dialog::{Dialog, to_tag};
28use crate::dispatch::with_to_tag;
29
30const RETRY_AFTER: &[u8] = b"5";
31const NOTIFY_RESPONSE_BOUND: Duration = Duration::from_secs(2);
32
33/// Runtime measurements for an endpoint event notifier.
34#[derive(Debug, Clone, Copy, PartialEq, Eq)]
35pub struct NotifierCounts {
36    /// Expiry/notification tasks currently alive.
37    pub active_tasks: usize,
38    /// Tasks started since this notifier was constructed.
39    pub started_tasks: u64,
40    /// Tasks which have exited, including cancellation on dispatcher drop.
41    pub finished_tasks: u64,
42    /// New subscriptions refused at the configured capacity.
43    pub shed: u64,
44}
45
46#[derive(Debug, Default)]
47struct Counters {
48    active: AtomicUsize,
49    started: AtomicU64,
50    finished: AtomicU64,
51    shed: AtomicU64,
52}
53
54/// Read-only application handle for the notifier attached to a dispatcher.
55///
56/// The returned store is the exact allocation the socket driver mutates. Exposing it makes state
57/// observable for policy, diagnostics and deterministic tests without asking the application to
58/// route SUBSCRIBE itself.
59#[derive(Debug, Clone)]
60pub struct NotifierHandle {
61    store: Arc<Mutex<Subscriptions>>,
62    counts: Arc<Counters>,
63}
64
65impl NotifierHandle {
66    /// The one subscription store used by both the library and socket paths.
67    #[must_use]
68    pub fn subscriptions(&self) -> Arc<Mutex<Subscriptions>> {
69        Arc::clone(&self.store)
70    }
71
72    /// A point-in-time runtime measurement.
73    #[must_use]
74    pub fn counts(&self) -> NotifierCounts {
75        NotifierCounts {
76            active_tasks: self.counts.active.load(Ordering::Relaxed),
77            started_tasks: self.counts.started.load(Ordering::Relaxed),
78            finished_tasks: self.counts.finished.load(Ordering::Relaxed),
79            shed: self.counts.shed.load(Ordering::Relaxed),
80        }
81    }
82}
83
84/// A bounded RFC 6665 notifier ready to attach to one [`crate::Dispatcher`].
85#[derive(Debug)]
86pub struct Notifier {
87    endpoint: Option<Handle>,
88    store: Arc<Mutex<Subscriptions>>,
89    counts: Arc<Counters>,
90    tasks: HashMap<Id, Running>,
91    origin: Instant,
92}
93
94impl Notifier {
95    /// Serve the three built-in event packages, granting at most `policy_maximum` and holding no
96    /// more than `capacity` concurrent subscriptions.
97    #[must_use]
98    pub fn new(policy_maximum: Duration, capacity: usize) -> Self {
99        let packages = Packages::new()
100            .with(DialogWatch::package())
101            .with(RegistrationWatch::package())
102            .with("presence");
103        Self {
104            endpoint: None,
105            store: Arc::new(Mutex::new(
106                Subscriptions::new(packages, policy_maximum).with_capacity(capacity),
107            )),
108            counts: Arc::new(Counters::default()),
109            tasks: HashMap::new(),
110            origin: Instant::now(),
111        }
112    }
113
114    /// A cloneable observation handle. It does not own runtime tasks.
115    #[must_use]
116    pub fn handle(&self) -> NotifierHandle {
117        NotifierHandle {
118            store: Arc::clone(&self.store),
119            counts: Arc::clone(&self.counts),
120        }
121    }
122
123    pub(crate) fn attach(&mut self, endpoint: Handle) {
124        self.endpoint = Some(endpoint);
125    }
126
127    /// Consume one SUBSCRIBE. The dispatcher has already selected the method.
128    pub(crate) async fn receive(&mut self, incoming: &Incoming) {
129        self.tasks.retain(|_, task| !task.join.is_finished());
130        let Some(endpoint) = self.endpoint.clone() else {
131            return;
132        };
133        if !valid_subscribe_headers(&incoming.request) {
134            answer(&endpoint, incoming, 400, "Bad Request", None, None, None).await;
135            return;
136        }
137        let Some(id) = Id::from_request(&incoming.request) else {
138            answer(&endpoint, incoming, 400, "Bad Request", None, None, None).await;
139            return;
140        };
141
142        if !PackageState::supports(&id.event) {
143            let allow = lock(&self.store).packages().allow_events();
144            answer(
145                &endpoint,
146                incoming,
147                489,
148                "Bad Event",
149                Some((HeaderName::AllowEvents, Bytes::from(allow))),
150                None,
151                None,
152            )
153            .await;
154            return;
155        }
156
157        let is_initial = to_tag(&incoming.request.headers).is_none();
158        if is_initial && self.tasks.contains_key(&id) {
159            answer(
160                &endpoint,
161                incoming,
162                481,
163                "Call/Transaction Does Not Exist",
164                None,
165                None,
166                None,
167            )
168            .await;
169            return;
170        }
171
172        if let Some(tag) = to_tag(&incoming.request.headers) {
173            let known = self
174                .tasks
175                .get(&id)
176                .is_some_and(|task| dialog_tag_matches(&task.local_tag, &tag));
177            if !known {
178                answer(
179                    &endpoint,
180                    incoming,
181                    481,
182                    "Call/Transaction Does Not Exist",
183                    None,
184                    None,
185                    None,
186                )
187                .await;
188                return;
189            }
190        }
191
192        // A terminating task still owns one package document and one scheduler slot. Do not let a
193        // rapid unsubscribe/re-subscribe cycle exceed the configured peer-driven task bound while
194        // that final NOTIFY is leaving.
195        let at_runtime_capacity = {
196            let store = lock(&self.store);
197            is_initial
198                && !self.tasks.contains_key(&id)
199                && store.packages().serves(&id.event)
200                && self.tasks.len() >= store.capacity()
201        };
202        if at_runtime_capacity {
203            self.counts.shed.fetch_add(1, Ordering::Relaxed);
204            answer(
205                &endpoint,
206                incoming,
207                503,
208                "Service Unavailable",
209                Some((HeaderName::RetryAfter, Bytes::from_static(RETRY_AFTER))),
210                None,
211                None,
212            )
213            .await;
214            return;
215        }
216
217        let now = self.origin.elapsed().as_secs();
218        let outcome = lock(&self.store).on_subscribe(&incoming.request, now);
219        self.apply(endpoint, incoming, outcome).await;
220    }
221
222    async fn apply(&mut self, endpoint: Handle, incoming: &Incoming, outcome: Answer) {
223        match outcome {
224            Answer::Malformed => {
225                answer(&endpoint, incoming, 400, "Bad Request", None, None, None).await;
226            }
227            Answer::OutOfOrder { .. } => {
228                answer(
229                    &endpoint,
230                    incoming,
231                    500,
232                    "Server Internal Error",
233                    None,
234                    None,
235                    None,
236                )
237                .await;
238            }
239            Answer::Unserved { status } => {
240                let allow = lock(&self.store).packages().allow_events();
241                answer(
242                    &endpoint,
243                    incoming,
244                    status,
245                    "Bad Event",
246                    Some((HeaderName::AllowEvents, Bytes::from(allow))),
247                    None,
248                    None,
249                )
250                .await;
251            }
252            Answer::AtCapacity => {
253                self.counts.shed.fetch_add(1, Ordering::Relaxed);
254                answer(
255                    &endpoint,
256                    incoming,
257                    503,
258                    "Service Unavailable",
259                    Some((HeaderName::RetryAfter, Bytes::from_static(RETRY_AFTER))),
260                    None,
261                    None,
262                )
263                .await;
264            }
265            Answer::Established { id, expires } => {
266                self.establish(&endpoint, incoming, id, expires).await;
267            }
268            Answer::Refreshed { id, expires } => {
269                self.refresh(&endpoint, incoming, &id, expires).await;
270            }
271            Answer::Unsubscribed { id } => {
272                self.unsubscribe(&endpoint, incoming, &id).await;
273            }
274        }
275    }
276
277    async fn refresh(&self, endpoint: &Handle, incoming: &Incoming, id: &Id, expires: Duration) {
278        let Some(running) = self.tasks.get(id) else {
279            answer(
280                endpoint,
281                incoming,
282                481,
283                "Call/Transaction Does Not Exist",
284                None,
285                None,
286                None,
287            )
288            .await;
289            return;
290        };
291        let allow = lock(&self.store).packages().allow_events();
292        answer(
293            endpoint,
294            incoming,
295            200,
296            "OK",
297            Some((HeaderName::AllowEvents, Bytes::from(allow))),
298            Some(&running.local_tag),
299            Some(expires),
300        )
301        .await;
302        // discard: a vanished task has already released this subscription; the 200 is final.
303        let _ = running.command.send(Command::Refresh(expires));
304    }
305
306    async fn unsubscribe(&self, endpoint: &Handle, incoming: &Incoming, id: &Id) {
307        let Some(running) = self.tasks.get(id) else {
308            answer(
309                endpoint,
310                incoming,
311                481,
312                "Call/Transaction Does Not Exist",
313                None,
314                None,
315                None,
316            )
317            .await;
318            return;
319        };
320        let allow = lock(&self.store).packages().allow_events();
321        answer(
322            endpoint,
323            incoming,
324            200,
325            "OK",
326            Some((HeaderName::AllowEvents, Bytes::from(allow))),
327            Some(&running.local_tag),
328            Some(Duration::ZERO),
329        )
330        .await;
331        // discard: a vanished task has already released this subscription; the 200 is final.
332        let _ = running
333            .command
334            .send(Command::Terminate(Reason::Deactivated));
335    }
336
337    async fn establish(
338        &mut self,
339        endpoint: &Handle,
340        incoming: &Incoming,
341        id: Id,
342        expires: Duration,
343    ) {
344        let local_tag = token();
345        let Some(dialog) = Dialog::from_request(&incoming.request, &local_tag) else {
346            lock(&self.store).terminate(&id, Reason::Rejected);
347            lock(&self.store).sweep();
348            answer(endpoint, incoming, 400, "Bad Request", None, None, None).await;
349            return;
350        };
351        let Some(package) = PackageState::for_request(&incoming.request, &id.event) else {
352            lock(&self.store).terminate(&id, Reason::Rejected);
353            lock(&self.store).sweep();
354            answer(endpoint, incoming, 400, "Bad Request", None, None, None).await;
355            return;
356        };
357
358        let allow = lock(&self.store).packages().allow_events();
359        answer(
360            endpoint,
361            incoming,
362            200,
363            "OK",
364            Some((HeaderName::AllowEvents, Bytes::from(allow))),
365            Some(&local_tag),
366            Some(expires),
367        )
368        .await;
369
370        let (command, receiver) = watch::channel(Command::Refresh(expires));
371        let task = Lifecycle {
372            endpoint: endpoint.clone(),
373            target: in_dialog_target(&dialog, Target::new(incoming.source, incoming.transport)),
374            dialog,
375            package,
376            id: id.clone(),
377            store: Arc::clone(&self.store),
378            receiver,
379        };
380        self.counts.active.fetch_add(1, Ordering::Relaxed);
381        self.counts.started.fetch_add(1, Ordering::Relaxed);
382        let guard = TaskGuard(Arc::clone(&self.counts));
383        let join = tokio::spawn(task.run(expires, guard));
384        self.tasks.insert(
385            id,
386            Running {
387                local_tag,
388                command,
389                join,
390            },
391        );
392    }
393
394    /// Terminate every service dialog and join its notification task.
395    pub(crate) async fn shutdown(&mut self) {
396        let running: Vec<_> = self.tasks.drain().map(|(_, running)| running).collect();
397        for task in &running {
398            // discard: an already-finished task has already released its subscription.
399            let _ = task.command.send(Command::Terminate(Reason::Deactivated));
400        }
401        for task in running {
402            if let Err(error) = task.join.await {
403                tracing::warn!(%error, "notifier task did not join cleanly during shutdown");
404            }
405        }
406    }
407}
408
409impl Drop for Notifier {
410    fn drop(&mut self) {
411        for task in self.tasks.values() {
412            task.join.abort();
413        }
414    }
415}
416
417#[derive(Debug)]
418struct Running {
419    local_tag: String,
420    command: watch::Sender<Command>,
421    join: JoinHandle<()>,
422}
423
424#[derive(Debug, Clone, Copy)]
425enum Command {
426    Refresh(Duration),
427    Terminate(Reason),
428}
429
430#[derive(Debug)]
431struct Lifecycle {
432    endpoint: Handle,
433    target: Target,
434    dialog: Dialog,
435    package: PackageState,
436    id: Id,
437    store: Arc<Mutex<Subscriptions>>,
438    receiver: watch::Receiver<Command>,
439}
440
441impl Lifecycle {
442    async fn run(mut self, expires: Duration, _guard: TaskGuard) {
443        let mut deadline = Instant::now() + expires;
444        let active = Subscription::active(expires);
445        self.send_notify(&active).await;
446
447        loop {
448            tokio::select! {
449                // This timer defines protocol expiry; it is not a happens-before substitute.
450                () = tokio::time::sleep_until(deadline) => {
451                    let terminated = lock(&self.store)
452                        .terminate(&self.id, Reason::Timeout)
453                        .unwrap_or_else(|| Subscription::terminated(Reason::Timeout));
454                    self.send_notify(&terminated).await;
455                    lock(&self.store).sweep();
456                    return;
457                }
458                changed = self.receiver.changed() => {
459                    if changed.is_err() {
460                        return;
461                    }
462                    let command = *self.receiver.borrow_and_update();
463                    match command {
464                        Command::Refresh(duration) => deadline = Instant::now() + duration,
465                        Command::Terminate(reason) => {
466                            let terminated = Subscription::terminated(reason);
467                            self.send_notify(&terminated).await;
468                            lock(&self.store).sweep();
469                            return;
470                        }
471                    }
472                }
473            }
474        }
475    }
476
477    async fn send_notify(&mut self, state: &Subscription) {
478        let (local, remote) = self.dialog.local_and_remote();
479        let (uri, routes) = self.dialog.request_target();
480        let cseq = self.dialog.next_cseq();
481        let (content_type, body) = self.package.document();
482        let built = RequestBuilder::new(Method::Notify, uri)
483            .header(HeaderName::To, Bytes::from(remote))
484            .and_then(|builder| builder.header(HeaderName::From, Bytes::from(local)))
485            .and_then(|builder| {
486                builder.header(
487                    HeaderName::CallId,
488                    Bytes::from(self.dialog.id.call_id.clone()),
489                )
490            })
491            .and_then(|builder| builder.cseq(cseq, &Method::Notify))
492            .and_then(|builder| {
493                builder.header(
494                    HeaderName::Contact,
495                    Bytes::from(contact_for(&self.endpoint, self.target.transport)),
496                )
497            })
498            .and_then(|builder| {
499                builder.header(HeaderName::Event, Bytes::from(self.id.event.clone()))
500            })
501            .and_then(|builder| {
502                builder.header(HeaderName::SubscriptionState, Bytes::from(state.to_value()))
503            })
504            .and_then(|builder| {
505                builder.header(HeaderName::ContentType, Bytes::from_static(content_type))
506            })
507            .and_then(|builder| add_routes(builder.max_forwards(70).body(body), &routes));
508        let request = match built {
509            Ok(builder) => builder.build(),
510            Err(error) => {
511                tracing::warn!(%error, "could not build subscription NOTIFY");
512                return;
513            }
514        };
515        match self.endpoint.send(request, self.target.clone()).await {
516            Ok(mut responses) => {
517                // This duration bounds a failed NOTIFY transaction; subscription state does not
518                // depend on whether the peer supplies the final response.
519                // discard: the bounded response is deliberately observational, never authoritative.
520                let _ =
521                    tokio::time::timeout(NOTIFY_RESPONSE_BOUND, responses.final_response()).await;
522            }
523            Err(error) => {
524                tracing::warn!(%error, "could not send subscription NOTIFY");
525            }
526        }
527    }
528}
529
530struct TaskGuard(Arc<Counters>);
531
532impl Drop for TaskGuard {
533    fn drop(&mut self) {
534        self.0.active.fetch_sub(1, Ordering::Relaxed);
535        self.0.finished.fetch_add(1, Ordering::Relaxed);
536    }
537}
538
539#[derive(Debug)]
540enum PackageState {
541    Dialog(DialogWatch),
542    Registration(RegistrationWatch),
543    Presence(Pidf),
544}
545
546impl PackageState {
547    fn supports(event: &str) -> bool {
548        matches!(
549            event.split(';').next().map(str::trim),
550            Some(package)
551                if package.eq_ignore_ascii_case("dialog")
552                    || package.eq_ignore_ascii_case("reg")
553                    || package.eq_ignore_ascii_case("presence")
554        )
555    }
556
557    fn for_request(request: &Request, event: &str) -> Option<Self> {
558        let entity = String::from_utf8_lossy(&request.uri.to_bytes()).into_owned();
559        match event.split(';').next()?.trim() {
560            package if package.eq_ignore_ascii_case("dialog") => {
561                Some(Self::Dialog(DialogWatch::new(entity)))
562            }
563            package if package.eq_ignore_ascii_case("reg") => {
564                Some(Self::Registration(RegistrationWatch::new(entity)))
565            }
566            package if package.eq_ignore_ascii_case("presence") => {
567                Some(Self::Presence(Pidf::new(entity)))
568            }
569            _ => None,
570        }
571    }
572
573    fn document(&mut self) -> (&'static [u8], Bytes) {
574        match self {
575            Self::Dialog(watch) => (
576                DIALOG_INFO_TYPE.as_bytes(),
577                Bytes::from(watch.document(&[])),
578            ),
579            Self::Registration(watch) => {
580                (REGINFO_TYPE.as_bytes(), Bytes::from(watch.document(&[])))
581            }
582            Self::Presence(pidf) => (PIDF_TYPE.as_bytes(), Bytes::from(pidf.to_xml())),
583        }
584    }
585}
586
587fn lock(store: &Arc<Mutex<Subscriptions>>) -> MutexGuard<'_, Subscriptions> {
588    store.lock().unwrap_or_else(PoisonError::into_inner)
589}
590
591fn valid_subscribe_headers(request: &Request) -> bool {
592    if request.method != Method::Subscribe
593        || request.headers.count(&HeaderName::CallId) != 1
594        || request.headers.count(&HeaderName::From) != 1
595        || request.headers.count(&HeaderName::To) != 1
596        || request.headers.count(&HeaderName::Event) != 1
597        || request.headers.count(&HeaderName::Contact) != 1
598        || request.headers.count(&HeaderName::CSeq) != 1
599        || request.headers.count(&HeaderName::Expires) > 1
600    {
601        return false;
602    }
603
604    let contacts: Vec<_> = request.headers.typed_all::<Contact>().collect();
605    matches!(contacts.as_slice(), [Ok(_)])
606        && matches!(
607            request.headers.typed::<CSeq>(),
608            Some(Ok(CSeq {
609                method: Method::Subscribe,
610                ..
611            }))
612        )
613        && !matches!(request.headers.typed::<Expires>(), Some(Err(_)))
614        && matches!(request.headers.typed::<FromHeader>(), Some(Ok(_)))
615        && matches!(request.headers.typed::<To>(), Some(Ok(_)))
616        && Id::from_request(request).is_some()
617}
618
619fn dialog_tag_matches(recorded: &str, received: &[u8]) -> bool {
620    recorded.as_bytes() == received
621}
622
623async fn answer(
624    endpoint: &Handle,
625    incoming: &Incoming,
626    status: u16,
627    reason: &'static str,
628    extra: Option<(HeaderName, Bytes)>,
629    tag: Option<&str>,
630    expires: Option<Duration>,
631) {
632    let Some(status) = StatusCode::new(status) else {
633        return;
634    };
635    let built = ResponseBuilder::to_request(&incoming.request, status, reason)
636        .and_then(|builder| {
637            if status.is_success() {
638                builder.header(
639                    HeaderName::Contact,
640                    Bytes::from(contact_for(endpoint, incoming.transport)),
641                )
642            } else {
643                Ok(builder)
644            }
645        })
646        .and_then(|builder| match extra {
647            Some((name, value)) => builder.header(name, value),
648            None => Ok(builder),
649        })
650        .and_then(|builder| match expires {
651            Some(value) => builder.header(
652                HeaderName::Expires,
653                Bytes::from(value.as_secs().to_string()),
654            ),
655            None => Ok(builder),
656        })
657        .and_then(|builder| with_to_tag(builder, &incoming.request, tag));
658    let response = match built {
659        Ok(builder) => builder.build(),
660        Err(error) => {
661            tracing::warn!(%error, "could not build SUBSCRIBE response");
662            return;
663        }
664    };
665    if let Err(error) = endpoint.respond(&incoming.key, response).await {
666        tracing::warn!(%error, "could not send SUBSCRIBE response");
667    }
668}
669
670#[cfg(test)]
671mod tests {
672    use super::dialog_tag_matches;
673
674    #[test]
675    fn opaque_dialog_tags_are_case_sensitive() {
676        assert!(dialog_tag_matches("LocalTag", b"LocalTag"));
677        assert!(!dialog_tag_matches("LocalTag", b"localtag"));
678    }
679}
sipx_call/notifier.rs

sipx_call/
notifier.rs
