liminal_server/server/connection/supervisor/
context_watch.rs

1use super::{ConnectionServices, LimitsConfig, ServerError};
2
3/// ACCESS-003 R1: the context watch drains its own subscription the way the
4/// connection pump does; each reading carries its message id, the stable
5/// event id Lys charges it under.
6impl crate::services::context_watch::ContextFeed for super::services::ConnectionSubscription {
7    fn next_reading(&mut self) -> Option<(String, Vec<u8>)> {
8        self.try_next()
9            .map(|envelope| (envelope.message_id.as_uuid().to_string(), envelope.payload))
10    }
11}
12
13/// Starts the context watch over the `context-watch` channel when the
14/// service bindings ask for it (ACCESS-003 R1).
15///
16/// One subscription whose inbox
17/// wakes the watcher on every admitted reading, bounded by the same §5
18/// limits a connection's subscription is. `Ok(None)` when the bindings do not
19/// watch.
20///
21/// # Errors
22/// The `context-watch` channel refused the subscription (it must be
23/// configured), or the watcher thread did not start.
24pub fn start_context_watch(
25    services: &dyn ConnectionServices,
26    bindings: &crate::services::production::ServiceBindings,
27    limits: &LimitsConfig,
28) -> Result<Option<crate::services::context_watch::ContextWatch>, ServerError> {
29    if !bindings.watches_context() {
30        return Ok(None);
31    }
32    let wake = crate::services::context_watch::ContextWake::new();
33    let install = liminal::channel::InboxInstall {
34        budget: liminal::channel::ConnectionInboxBudget::new(limits.max_connection_inbox_bytes),
35        depth_cap: limits.max_subscription_inbox_depth,
36        notifier: Some(wake.notifier()),
37        capacity: None,
38    };
39    let subscription = services.subscribe(
40        crate::services::ServiceKind::ContextWatch.subject_shape(),
41        &[],
42        Some(install),
43    )?;
44    crate::services::context_watch::ContextWatch::start(
45        subscription,
46        wake,
47        bindings.context_reporter(),
48    )
49    .map(Some)
50    .map_err(|error| ServerError::ConfigValidation {
51        message: error.to_string(),
52    })
53}
liminal_server/server/connection/supervisor/context_watch.rs

liminal_server/server/connection/supervisor/
context_watch.rs
