frame_host/embedded.rs
1//! The embedded liminal component.
2//!
3//! Liminal runs INSIDE the frame server process — not as a child process and
4//! not side-by-side. Frame-host owns liminal's lifecycle (it boots the
5//! listeners, holds their handles, and drives graceful shutdown) but never its
6//! bytes: the browser connects to liminal's WebSocket directly (D3), so no feed
7//! data passes through this host.
8//!
9//! ## Why this composes liminal's primitives instead of calling `liminal_server::run`
10//!
11//! `liminal_server::run(&Path)` is a `main`-shaped wrapper: it takes a config
12//! FILE PATH (not a config value), creates its own [`ShutdownHandle`] that it
13//! never returns, and registers PROCESS-GLOBAL SIGTERM/SIGINT handlers itself.
14//! None of that is usable from inside an embedding host — the host cannot inject
15//! a config, cannot reach a shutdown handle to stop it, and cannot let liminal
16//! fight the host for signal disposition. So this module composes liminal's
17//! PUBLIC primitives directly (the same construction `run` performs internally),
18//! consuming liminal entirely as-is with zero edits to its tree:
19//!
20//! - [`start_health_server`] — the health/readiness endpoint,
21//! - [`LiminalConnectionServices::from_config`] + [`ConnectionSupervisor`] —
22//! the full-profile channel/conversation/durability stack,
23//! - [`ServerListener::bind`] — the TCP wire listener,
24//! - [`WebSocketListener::bind`] — the browser-facing WebSocket transport,
25//! - [`run_shutdown_sequence`] — liminal's own graceful drain/flush.
26//!
27//! [`ShutdownHandle`]: liminal_server::server::ShutdownHandle
28
29use std::net::SocketAddr;
30use std::sync::Arc;
31use std::time::Duration;
32
33use liminal_server::config::ServerConfig;
34use liminal_server::health::{
35 HealthServerHandle, ReadinessState, SharedReadinessState, start_health_server,
36};
37use liminal_server::server::connection::{LiminalConnectionServices, WebSocketListener};
38use liminal_server::server::shutdown::run_shutdown_sequence;
39use liminal_server::server::{ConnectionSupervisor, ServerListener};
40
41use crate::error::HostError;
42
43/// A booted, live embedded liminal component: the TCP listener, the browser
44/// WebSocket listener, and the health endpoint, all bound and serving on
45/// liminal's own threads inside the frame-host process.
46///
47/// Holding this value keeps the component alive. [`Self::shutdown`] drives
48/// liminal's own graceful teardown; dropping it without calling shutdown still
49/// stops every listener thread via each handle's `Drop` (no leaked threads).
50pub struct EmbeddedLiminal {
51 listener: ServerListener,
52 websocket: WebSocketListener,
53 health: HealthServerHandle,
54 readiness: SharedReadinessState,
55 drain_timeout: Duration,
56 tcp_addr: SocketAddr,
57 websocket_addr: SocketAddr,
58 health_addr: SocketAddr,
59 websocket_path: String,
60}
61
62impl EmbeddedLiminal {
63 /// Boots the embedded liminal component from the validated `[bus]`
64 /// config: health endpoint, full-profile services, TCP listener, and the
65 /// browser WebSocket listener — each bound before the next.
66 ///
67 /// The caller ([`crate::config::FrameConfig::load`]) has already validated
68 /// the config and refused any embedded-unsupported shape, so `[websocket]`
69 /// is guaranteed present here.
70 ///
71 /// # Errors
72 ///
73 /// Returns [`HostError::LiminalComponent`] naming the exact component that
74 /// failed to bind or boot. Every already-bound component is torn down by its
75 /// own `Drop` as the error unwinds, so a failed boot never leaves a listener
76 /// behind — there is no half-up state.
77 pub fn boot(config: &ServerConfig) -> Result<Self, HostError> {
78 let websocket_config =
79 config
80 .websocket
81 .as_ref()
82 .ok_or_else(|| HostError::EmbeddedModeUnsupported {
83 detail: "internal invariant violated: EmbeddedLiminal::boot requires \
84 [bus.websocket]; FrameConfig::load must refuse its absence first"
85 .to_owned(),
86 })?;
87
88 // Mirror standalone liminal's startup (`server/runtime.rs`): enable the
89 // process-global metrics registry and register the three server
90 // families BEFORE any listener spawns, so `/metrics` renders real data
91 // and the connection/publish/delivery recorders are live. Without this
92 // the embedded `/metrics` answers 200 with an empty body — a silent
93 // observability hole the health-proxy e2e now guards against.
94 liminal_server::metrics::init();
95
96 let readiness = SharedReadinessState::new(ReadinessState::default());
97 let health = start_health_server(config.health_listen_address, readiness.clone()).map_err(
98 |source| HostError::LiminalComponent {
99 component: "health endpoint",
100 source,
101 },
102 )?;
103 let health_addr = health.local_addr();
104
105 let auth_token = config
106 .auth
107 .as_ref()
108 .map(|auth| auth.token.clone().into_bytes());
109 let services = Arc::new(LiminalConnectionServices::from_config(config).map_err(
110 |source| HostError::LiminalComponent {
111 component: "connection services",
112 source,
113 },
114 )?);
115 let supervisor = ConnectionSupervisor::with_services_auth_and_limits(
116 services,
117 auth_token,
118 config.limits,
119 )
120 .map_err(|source| HostError::LiminalComponent {
121 component: "connection supervisor",
122 source,
123 })?;
124
125 let listener = ServerListener::bind(config, supervisor).map_err(|source| {
126 HostError::LiminalComponent {
127 component: "TCP listener",
128 source,
129 }
130 })?;
131 let tcp_addr = listener.local_addr();
132
133 let websocket =
134 WebSocketListener::bind(websocket_config, listener.supervisor()).map_err(|source| {
135 HostError::LiminalComponent {
136 component: "WebSocket listener",
137 source,
138 }
139 })?;
140 let websocket_addr = websocket.local_addr();
141
142 // Mirror standalone liminal's readiness sequencing: single-node (no
143 // cluster), config loaded, listeners bound.
144 readiness.set_cluster_configured(false);
145 readiness.set_config_loaded(true);
146 readiness.set_listener_bound(true);
147
148 Ok(Self {
149 listener,
150 websocket,
151 health,
152 readiness,
153 drain_timeout: config.drain_timeout(),
154 tcp_addr,
155 websocket_addr,
156 health_addr,
157 websocket_path: websocket_config.path.clone(),
158 })
159 }
160
161 /// The browser-facing WebSocket endpoint, derived from the ACTUAL bound
162 /// listener address plus the configured upgrade path.
163 ///
164 /// This is the single source of truth surfaced to the page as
165 /// `busEndpoint` (and, during the compatibility window, as the deprecated
166 /// duplicate `liminalEndpoint`): because it comes from the live listener's own
167 /// `local_addr`, the page's endpoint and the server's real bound address are
168 /// one value and can never drift.
169 #[must_use]
170 pub fn websocket_endpoint(&self) -> String {
171 format!("ws://{}{}", self.websocket_addr, self.websocket_path)
172 }
173
174 /// The bound TCP wire listener address (native clients, e.g. the demo
175 /// publisher, connect here).
176 #[must_use]
177 pub const fn tcp_addr(&self) -> SocketAddr {
178 self.tcp_addr
179 }
180
181 /// The bound browser WebSocket listener address.
182 #[must_use]
183 pub const fn websocket_addr(&self) -> SocketAddr {
184 self.websocket_addr
185 }
186
187 /// The bound health endpoint address.
188 #[must_use]
189 pub const fn health_addr(&self) -> SocketAddr {
190 self.health_addr
191 }
192
193 /// Drives liminal's own graceful shutdown: stop accepting on both
194 /// transports, drain/force-close active connections, flush durable state,
195 /// then stop the health endpoint.
196 ///
197 /// # Errors
198 ///
199 /// Returns [`HostError::LiminalShutdown`] if liminal's drain/flush fails, or
200 /// [`HostError::LiminalComponent`] if the health endpoint fails to stop.
201 pub fn shutdown(mut self) -> Result<(), HostError> {
202 self.readiness.set_listener_bound(false);
203 let supervisor = self.listener.supervisor();
204 let drain = run_shutdown_sequence(
205 &mut self.listener,
206 Some(&mut self.websocket),
207 &supervisor,
208 self.drain_timeout,
209 )
210 .map_err(|source| HostError::LiminalShutdown { source });
211 // Stop the health endpoint UNCONDITIONALLY, even if the drain failed —
212 // matching standalone liminal's `run`, which always calls
213 // `health_server.shutdown()` regardless of the drain result. The drain
214 // error takes precedence in the report.
215 let health = self
216 .health
217 .shutdown()
218 .map_err(|source| HostError::LiminalComponent {
219 component: "health endpoint",
220 source,
221 });
222 drain?;
223 health?;
224 Ok(())
225 }
226}