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

liminal_server/server/
runtime.rs
