liminal_server/server/
runtime.rs

1use std::path::Path;
2use std::sync::Arc;
3
4use crate::ServerError;
5use crate::cluster::{self, ClusterHandle};
6use crate::config::file::load_config;
7use crate::config::types::{ClusterConfig, ServiceProfile};
8use crate::health::{ReadinessState, SharedReadinessState, start_health_server};
9use crate::server::connection::ConnectionSupervisor;
10use crate::server::connection::WebSocketListener;
11use crate::server::connection::services::{
12    ChannelCluster, LiminalConnectionServices, build_connection_services,
13};
14use crate::server::listener::ServerListener;
15use crate::server::shutdown::{ShutdownHandle, register_signal_handlers, run_shutdown_sequence};
16
17/// Starts the server deployment wrapper for the supplied configuration path.
18///
19/// # Errors
20///
21/// Returns [`ServerError`] when a later server lifecycle phase fails.
22pub fn run(config_path: &Path) -> Result<(), ServerError> {
23    if config_path.as_os_str().is_empty() {
24        return Err(ServerError::ConfigLoad {
25            message: "configuration path is empty".to_owned(),
26        });
27    }
28
29    let config = load_config(config_path)?;
30
31    // Enable metrics for this process before the health server accepts scrapes,
32    // so `/metrics` renders the server families. Standalone liminal library users
33    // never call this, so the registry gate stays off for them.
34    crate::metrics::init();
35
36    let readiness = SharedReadinessState::new(ReadinessState::default());
37    let health_server = start_health_server(config.health_listen_address, readiness.clone())?;
38    let shutdown_handle = ShutdownHandle::new();
39    let signal_registration = register_signal_handlers(shutdown_handle.clone())?;
40
41    // The configured [auth] token must ride along here: these call sites build
42    // services themselves (full mode reaches the shared channel cluster first;
43    // the worker front door builds no cluster at all) and so cannot use
44    // `from_config`, which is the only other place the token is wired.
45    let auth_token = config
46        .auth
47        .as_ref()
48        .map(|auth| auth.token.clone().into_bytes());
49
50    // D2: the service profile selects which connection-services stack is built.
51    // Full mode is byte-for-byte the previous construction path (build services,
52    // reach the shared channel cluster, start clustering when configured). The
53    // worker front door constructs the connection supervisor over the
54    // capability-scoped adapter and NOTHING else — no channel/conversation/haematite
55    // services, and therefore no distribution cluster (config validation rejects a
56    // `[cluster]` section under this profile, so none can be present here).
57    let (connection_supervisor, cluster_handle) = match config.services.profile()? {
58        ServiceProfile::Full => {
59            let services = Arc::new(LiminalConnectionServices::from_config(&config)?);
60            let channel_cluster = services.channel_cluster().clone();
61            let connection_supervisor = ConnectionSupervisor::with_fatal_shutdown(
62                services,
63                auth_token,
64                config.limits,
65                shutdown_handle.clone(),
66            )?;
67
68            // SRV-005: start clustering on the channel-supervisor scheduler when a
69            // [cluster] section is configured. The returned handle owns the inbound
70            // distribution listener and the membership poll loop; it must outlive the
71            // server and is torn down in the shutdown sequence below.
72            readiness.set_cluster_configured(config.cluster.is_some());
73            let cluster_handle = match config.cluster.as_ref() {
74                Some(cluster_config) => {
75                    Some(start_cluster(&channel_cluster, cluster_config, &readiness)?)
76                }
77                None => None,
78            };
79            (connection_supervisor, cluster_handle)
80        }
81        ServiceProfile::WorkerFrontDoor => {
82            let services = build_connection_services(&config)?;
83            let connection_supervisor = ConnectionSupervisor::with_fatal_shutdown(
84                services,
85                auth_token,
86                config.limits,
87                shutdown_handle.clone(),
88            )?;
89            readiness.set_cluster_configured(false);
90            (connection_supervisor, None)
91        }
92    };
93
94    let mut listener = ServerListener::bind(&config, connection_supervisor)?;
95    // LP-WS-TRANSPORT R1.1: the sibling WebSocket acceptor is explicit opt-in.
96    // Absent `[websocket]` binds nothing — no HTTP surface exists at all — and
97    // a bind failure fails startup BEFORE readiness reports the listeners
98    // bound, exactly like the main listener.
99    let mut websocket_listener = match config.websocket.as_ref() {
100        Some(websocket_config) => Some(WebSocketListener::bind(
101            websocket_config,
102            listener.supervisor(),
103        )?),
104        None => None,
105    };
106    readiness.set_config_loaded(true);
107    readiness.set_listener_bound(true);
108
109    tracing::debug!(
110        config_path = %config_path.display(),
111        listen_address = %config.listen_address,
112        health_listen_address = %health_server.local_addr(),
113        "liminal server configuration validated"
114    );
115
116    tracing::info!(
117        listen_address = %listener.local_addr(),
118        health_listen_address = %health_server.local_addr(),
119        "liminal server started"
120    );
121
122    shutdown_handle.wait();
123    readiness.set_listener_bound(false);
124
125    // Tear the cluster down before draining connections: stop accepting peer
126    // links and halt the membership poll loop. Each node shuts down independently
127    // (no cluster-wide coordinated shutdown — that boundary belongs to SRV-004).
128    if let Some(mut cluster_handle) = cluster_handle {
129        cluster_handle.shutdown();
130    }
131
132    let supervisor = listener.supervisor();
133    let shutdown_result = run_shutdown_sequence(
134        &mut listener,
135        websocket_listener.as_mut(),
136        &supervisor,
137        config.drain_timeout(),
138    );
139    let participant_fatal = supervisor.participant_service_fatal();
140    drop(websocket_listener);
141    drop(signal_registration);
142    health_server.shutdown()?;
143    shutdown_result?;
144    participant_fatal?.map_or(Ok(()), |fatal| {
145        Err(ServerError::ParticipantServiceFatal { fatal })
146    })
147}
148
149/// Starts clustering on the shared channel supervisor's scheduler (SRV-005).
150///
151/// Installs the cluster `sync` as the supervisor's [`ClusterObserver`] so channel
152/// subscribe/unsubscribe/publish events drive process-group membership and
153/// cross-node fan-out.
154///
155/// On the success path this marks cluster membership as established on `readiness`
156/// (G2) via [`cluster::start`]'s `on_established` hook, so a clustered server's
157/// `/ready` endpoint transitions from 503 to 200 once the cluster stack is up.
158/// Every early return here (missing resolver, listener bind failure, no reachable
159/// seed) leaves the flag unset, so `/ready` stays 503.
160fn start_cluster(
161    channel_cluster: &ChannelCluster,
162    cluster_config: &ClusterConfig,
163    readiness: &SharedReadinessState,
164) -> Result<ClusterHandle, ServerError> {
165    let resolver = channel_cluster
166        .resolver()
167        .cloned()
168        .ok_or_else(|| ServerError::ClusterJoin {
169            message: "clustering configured but channel supervisor has no distribution resolver"
170                .to_owned(),
171        })?;
172    let scheduler = channel_cluster.supervisor().scheduler();
173    let supervisor = channel_cluster.supervisor().clone();
174    let readiness = readiness.clone();
175    cluster::start(
176        &scheduler,
177        resolver,
178        cluster_config,
179        move |sync| {
180            supervisor.install_observer(Arc::new(sync));
181        },
182        move || readiness.set_cluster_membership_established(true),
183    )
184}
185
186#[cfg(test)]
187mod tests {
188    use std::net::SocketAddr;
189
190    use super::{ChannelCluster, ClusterConfig, SharedReadinessState, start_cluster};
191    use crate::ServerError;
192    use crate::health::{ClusterReadiness, ReadinessCondition, ReadinessState, readiness_check};
193    use crate::server::connection::services::LiminalConnectionServices;
194
195    /// A channel cluster with NO distribution resolver — the shape produced when a
196    /// server was built without a `[cluster]` section. `start_cluster` must reject
197    /// it before touching `cluster::start`, so its `on_established` hook never runs.
198    fn unclustered_channel_cluster() -> Result<ChannelCluster, ServerError> {
199        Ok(LiminalConnectionServices::empty()?
200            .channel_cluster()
201            .clone())
202    }
203
204    fn clustered_but_unmet_readiness() -> SharedReadinessState {
205        SharedReadinessState::new(ReadinessState::new(
206            true,
207            true,
208            ClusterReadiness::Configured {
209                membership_established: false,
210            },
211        ))
212    }
213
214    fn sample_cluster_config() -> Result<ClusterConfig, Box<dyn std::error::Error>> {
215        let listen_address: SocketAddr = "127.0.0.1:0".parse()?;
216        Ok(ClusterConfig {
217            node_name: "node-under-test@127.0.0.1".to_owned(),
218            listen_address,
219            seed_nodes: Vec::new(),
220            cookie: "runtime-test-cookie".to_owned(),
221        })
222    }
223
224    #[test]
225    fn failed_cluster_start_leaves_membership_unestablished()
226    -> Result<(), Box<dyn std::error::Error>> {
227        let readiness = clustered_but_unmet_readiness();
228        let channel_cluster = unclustered_channel_cluster()?;
229        let config = sample_cluster_config()?;
230
231        // Missing-resolver failure path: start_cluster returns Err before the
232        // established hook can fire.
233        let result = start_cluster(&channel_cluster, &config, &readiness);
234        assert!(
235            result.is_err(),
236            "start_cluster must fail without a distribution resolver"
237        );
238
239        // The readiness flag stays unset, so /ready still lists the unmet gate.
240        let status = readiness_check(&readiness.snapshot());
241        assert!(
242            !status.ready,
243            "readiness must remain not-ready after a failed start"
244        );
245        assert!(
246            status
247                .unmet_conditions
248                .contains(&ReadinessCondition::ClusterMembershipEstablished),
249            "cluster membership gate must stay unmet after a failed start"
250        );
251
252        Ok(())
253    }
254}
liminal_server/server/runtime.rs

liminal_server/server/
runtime.rs
