liminal_server/server/connection/
worker_front_door.rs

1//! Capability-scoped worker front door services (D2).
2//!
3//! [`WorkerFrontDoorServices`] is the [`ConnectionServices`] adapter for
4//! worker-only ("worker front door") deployments — aion-style hosts that use the
5//! bus for worker registration, correlated server push/reply, and
6//! notifier-consumed reserved publishes, and nothing else. It constructs no
7//! channel supervisor, conversation supervisor, haematite database/store, dedup
8//! cache, or temporary directory: it is a stateless adapter, so the only runtime
9//! the worker-front-door construction path spins up is the connection supervisor's
10//! own scheduler (owned by [`super::supervisor`], shared by both profiles).
11//!
12//! Worker registration, correlated push/reply, and the reserved observability tap
13//! do NOT flow through this adapter at all — they are served by the connection
14//! supervisor and the [`ConnectionNotifier`](super::notifier::ConnectionNotifier)
15//! independently of which `ConnectionServices` is installed, so they behave exactly
16//! as in full mode. This adapter carries only the channel/conversation operations,
17//! every one of which it rejects with a typed error frame.
18
19use liminal::protocol::{MessageEnvelope, SchemaId as ProtocolSchemaId};
20
21use super::conversation::ConnectionConversation;
22use super::services::{ConnectionServices, ConnectionSubscription, PublishOutcome};
23use crate::ServerError;
24use crate::config::types::ServiceProfile;
25
26/// [`ConnectionServices`] for the worker front door: registration, correlated
27/// push/reply, and notifier-consumed reserved publishes only.
28///
29/// The type holds no state and constructs nothing, so [`Self::new`] is an
30/// infallible `const fn` — a structural signal that this path starts no scheduler
31/// and opens no store (both of which are fallible in full mode). Ordinary
32/// publish/subscribe/conversation operations return a typed [`ServerError`], which
33/// [`super::apply`] renders as the matching typed error frame
34/// (`PublishError`/`SubscribeError`/`ConversationError`); the connection stays
35/// healthy across the rejection.
36#[derive(Debug, Default, Clone, Copy)]
37pub struct WorkerFrontDoorServices;
38
39impl WorkerFrontDoorServices {
40    /// Creates the worker-front-door adapter. Constructs nothing.
41    #[must_use]
42    pub const fn new() -> Self {
43        Self
44    }
45
46    /// Builds the typed rejection returned for an unsupported channel/conversation
47    /// operation. Rendered by [`super::apply`] as the operation's typed error frame.
48    fn unsupported(operation: &str) -> ServerError {
49        ServerError::UnsupportedOperation {
50            operation: operation.to_owned(),
51            profile: ServiceProfile::WORKER_FRONT_DOOR,
52        }
53    }
54}
55
56impl ConnectionServices for WorkerFrontDoorServices {
57    fn publish(
58        &self,
59        _connection: u64,
60        channel: &str,
61        _envelope: &MessageEnvelope,
62        _idempotency_key: Option<&str>,
63    ) -> Result<PublishOutcome, ServerError> {
64        // Reserved observability publishes never reach here: `apply_frame` offers
65        // every publish to the notifier's observability tap FIRST and only falls
66        // through to `services.publish` when the tap did not consume it. So an
67        // ordinary (untapped) publish is the only thing that lands here, and the
68        // front door serves no channels — reject it explicitly.
69        Err(Self::unsupported(&format!(
70            "publish to channel '{channel}'"
71        )))
72    }
73
74    fn subscribe(
75        &self,
76        channel: &str,
77        _accepted_schemas: &[ProtocolSchemaId],
78        _install: Option<liminal::channel::InboxInstall>,
79    ) -> Result<ConnectionSubscription, ServerError> {
80        Err(Self::unsupported(&format!(
81            "subscribe to channel '{channel}'"
82        )))
83    }
84
85    fn unsubscribe(&self, subscription: ConnectionSubscription) -> Result<(), ServerError> {
86        // Unreachable in practice: `subscribe` never yields a `ConnectionSubscription`,
87        // so no owned subscription can exist to pass here, and `apply_frame` rejects
88        // an `Unsubscribe` frame before this method via `supports_channel_operations`.
89        // Release the resource and succeed rather than invent a second rejection
90        // surface for a value that cannot be constructed in this profile.
91        subscription.unsubscribe()
92    }
93
94    fn open_conversation(
95        &self,
96        _conversation_id: u64,
97        subject: &str,
98    ) -> Result<ConnectionConversation, ServerError> {
99        Err(Self::unsupported(&format!(
100            "opening conversation '{subject}'"
101        )))
102    }
103
104    fn conversation_message(
105        &self,
106        _conversation: &ConnectionConversation,
107        _envelope: &MessageEnvelope,
108        _op_id: Option<u64>,
109    ) -> Result<(), ServerError> {
110        // Unreachable: no `ConnectionConversation` can exist because
111        // `open_conversation` always rejects. `apply_frame` rejects a
112        // `ConversationMessage` frame before this method via
113        // `supports_channel_operations`.
114        Err(Self::unsupported("conversation message"))
115    }
116
117    fn close_conversation(&self, conversation: ConnectionConversation) -> Result<(), ServerError> {
118        // Unreachable for the same reason as `unsubscribe`; release and succeed.
119        conversation.close()
120    }
121
122    fn flush_durable_state(&self) -> Result<(), ServerError> {
123        // No durable channels exist, so a graceful-shutdown flush is a total no-op —
124        // never an error, so shutting a worker-front-door server down stays clean.
125        Ok(())
126    }
127
128    fn supports_channel_operations(&self) -> bool {
129        false
130    }
131}
132
133#[cfg(test)]
134mod tests {
135    use super::WorkerFrontDoorServices;
136    use crate::server::connection::services::ConnectionServices;
137
138    /// Structural adjunct to the §9 D2 gate: the front-door adapter constructs
139    /// nothing, so its constructor is infallible and starts no scheduler. A
140    /// `Scheduler::new` (channel/conversation/haematite) is fallible and returns
141    /// `Result`; this returning a bare `Self` is the in-language signal that no
142    /// beamr scheduler is created by the adapter itself.
143    ///
144    /// The thread half of the §9 gate is the record-by-construction SEAM CENSUS,
145    /// not this test: the profile-aware construction path reaches every
146    /// scheduler-owning subsystem only through `services::SubsystemFactory`, whose
147    /// recording test implementation cannot skip a record without also failing to
148    /// construct (asserted with a full-profile positive control in
149    /// `services::durable_store_tests` and `supervisor::tests`). A true OS-level
150    /// thread census still belongs to the beamr composition lane's upcoming
151    /// scheduler-inventory API; the census assertions upgrade to it when it lands.
152    /// It is deliberately NOT faked with a platform-specific OS thread count.
153    #[test]
154    fn construction_is_infallible_and_starts_no_scheduler() {
155        let services = WorkerFrontDoorServices::new();
156        // The type carries no channel/conversation/store/dedup fields — that absence
157        // is compile-time enforced by the struct definition. This adapter reports it
158        // serves no channel/conversation operations.
159        assert!(
160            !services.supports_channel_operations(),
161            "the worker front door must report that it serves no channel operations"
162        );
163    }
164}
liminal_server/server/connection/worker_front_door.rs

liminal_server/server/connection/
worker_front_door.rs
