rullst_core/server/
builder.rs

1use crate::Router;
2use crate::lifecycle::{ApplicationLifecycle, apply_lifecycle};
3use crate::scheduler::{Scheduler, SchedulerHandle};
4use crate::server::dylib_loader::load_dylib_router;
5use crate::server::hotswap::HotSwapService;
6use crate::server::server_middleware::zstd_static_middleware;
7#[cfg(feature = "orm")]
8use rullst_orm::Orm;
9use std::collections::HashMap;
10use std::net::SocketAddr;
11use std::sync::{Arc, Mutex, RwLock};
12
13/// Typed server startup and runtime failures.
14#[derive(Debug, thiserror::Error)]
15#[non_exhaustive]
16pub enum ServerError {
17    /// Application configuration is invalid or unreadable.
18    #[error("server configuration error: {0}")]
19    Configuration(String),
20
21    /// A configured database could not be initialized.
22    #[error("database initialization failed: {0}")]
23    Database(String),
24
25    /// A configured background scheduler could not be started or stopped.
26    #[error("scheduler lifecycle failed: {0}")]
27    Scheduler(#[from] crate::scheduler::SchedulerError),
28
29    /// Traffic Shield monitoring could not be started safely.
30    #[error("traffic shield lifecycle failed: {0}")]
31    TrafficShield(#[from] crate::resilience::TrafficShieldError),
32
33    /// The process readiness or graceful-drain lifecycle became invalid.
34    #[error("application lifecycle failed: {0}")]
35    Lifecycle(#[from] crate::lifecycle::ApplicationLifecycleError),
36
37    /// The requested listen address is invalid.
38    #[error("invalid server listen address `{host}:{port}`")]
39    InvalidAddress {
40        /// Configured host value.
41        host: String,
42        /// Configured TCP port.
43        port: u16,
44    },
45
46    /// Hot reload was requested outside its supported local debug mode.
47    #[error("hot reload is available only in local development debug builds")]
48    HotReloadDisabled,
49
50    /// Loading or invoking a hot-reload library failed.
51    #[error("hot reload failed: {0}")]
52    HotReload(String),
53
54    /// The private development reload channel is not configured safely.
55    #[error("hot reload configuration error: {0}")]
56    HotReloadConfiguration(String),
57
58    /// An operating-system I/O operation failed.
59    #[error("server I/O error: {0}")]
60    Io(#[from] std::io::Error),
61}
62
63#[non_exhaustive]
64/// The central application server builder for Rullst.
65///
66/// Configures and boots the Axum HTTP server, optional ORM connection pool,
67/// task scheduler, hot-reload DLL watcher, traffic shield, and rate limiter in
68/// a single fluent chain.
69///
70/// # Example
71/// ```rust,no_run
72/// use rullst_core::{Server, routes, routing::get};
73///
74/// #[tokio::main]
75/// async fn main() -> Result<(), Box<dyn std::error::Error>> {
76///     Server::new(routes![get("/" => || async { "OK" })])
77///         .run(3000)
78///         .await?;
79///     Ok(())
80/// }
81/// ```
82pub struct Server {
83    pub(crate) router: Router,
84    pub(crate) db_url: Option<String>,
85    pub(crate) scheduler: Option<Scheduler>,
86    pub(crate) hot_reload_lib: Option<String>,
87    pub(crate) shield: Option<crate::resilience::TrafficShield>,
88    pub(crate) limiter: Option<crate::resilience::RateLimiter>,
89    pub(crate) lifecycle: Option<ApplicationLifecycle>,
90    pub(crate) machine_endpoints: Option<crate::security::MachineEndpointPolicy>,
91}
92
93impl Server {
94    /// Creates a new `Server` from an already-built [`Router`].
95    /// Use [`Server::new_hot`] instead to enable hot-reload mode.
96    pub fn new(router: Router) -> Self {
97        Server {
98            router,
99            db_url: None,
100            scheduler: None,
101            hot_reload_lib: None,
102            shield: None,
103            limiter: None,
104            lifecycle: None,
105            machine_endpoints: None,
106        }
107    }
108
109    /// Creates a `Server` in **hot-reload** mode that loads the application router from
110    /// a compiled `cdylib` dynamic library at the given `lib_path`.
111    /// The background file-watcher recompiles and hot-swaps the router on source changes.
112    pub fn new_hot<S: Into<String>>(lib_path: S) -> Self {
113        Server {
114            router: Router::new(),
115            db_url: None,
116            scheduler: None,
117            hot_reload_lib: Some(lib_path.into()),
118            shield: None,
119            limiter: None,
120            lifecycle: None,
121            machine_endpoints: None,
122        }
123    }
124
125    /// Requires explicit machine authentication before CSRF exemptions on exact routes.
126    pub fn with_machine_endpoints(
127        mut self,
128        policy: crate::security::MachineEndpointPolicy,
129    ) -> Self {
130        self.machine_endpoints = Some(policy);
131        self
132    }
133
134    /// Sets a database URL to initialize the ORM connection pool at startup.
135    ///
136    /// This requires the `orm` feature. When Core is compiled without `orm`,
137    /// configuring a database remains a valid builder operation but [`Self::run`]
138    /// fails closed with [`ServerError::Database`].
139    pub fn with_db<S: Into<String>>(mut self, db_url: S) -> Self {
140        self.db_url = Some(db_url.into());
141        self
142    }
143
144    /// Attach a task scheduler that runs alongside the HTTP server.
145    ///
146    /// # Example
147    /// ```rust,no_run
148    /// use rullst_core::{Server, Scheduler, routes, routing::get};
149    ///
150    /// #[tokio::main]
151    /// async fn main() -> Result<(), Box<dyn std::error::Error>> {
152    ///     let scheduler = Scheduler::new()
153    ///         .task("0 0 * * *", || async { println!("daily cleanup"); })?;
154    ///     let router = routes![get("/" => || async { "OK" })];
155    ///
156    ///     Server::new(router)
157    ///         .schedule(scheduler)
158    ///         .run(3000)
159    ///         .await?;
160    ///     Ok(())
161    /// }
162    /// ```
163    pub fn schedule(mut self, scheduler: Scheduler) -> Self {
164        self.scheduler = Some(scheduler);
165        self
166    }
167
168    /// Attaches an adaptive TrafficShield to the server to protect against CPU/DB saturation.
169    pub fn shield(mut self, shield: crate::resilience::TrafficShield) -> Self {
170        self.shield = Some(shield);
171        self
172    }
173
174    /// Attaches a global RateLimiter to the server.
175    pub fn rate_limit(mut self, limiter: crate::resilience::RateLimiter) -> Self {
176        self.limiter = Some(limiter);
177        self
178    }
179
180    /// Attaches a shared readiness and graceful-drain coordinator.
181    ///
182    /// Static and development hot-reload requests are admitted only while this
183    /// lifecycle and all of its required components are ready. Exact health
184    /// probes remain reachable. Mount
185    /// [`crate::health::health_router_with_lifecycle`] with a clone to expose
186    /// the same aggregate state to an orchestrator.
187    pub fn with_lifecycle(mut self, lifecycle: ApplicationLifecycle) -> Self {
188        self.lifecycle = Some(lifecycle);
189        self
190    }
191
192    /// Start the HTTP server on the specified port
193    #[cfg_attr(mutants, mutants::skip)]
194    pub async fn run(self, port: u16) -> Result<(), ServerError> {
195        self.run_with_shutdown(port, shutdown_signal()).await
196    }
197
198    /// Starts the HTTP server with a caller-supplied graceful-shutdown future.
199    ///
200    /// This is useful for embedded process supervisors and deterministic tests.
201    /// Resolving the future begins lifecycle draining before Axum waits for
202    /// already accepted requests. [`Self::run`] remains the OS-signal default.
203    #[cfg_attr(mutants, mutants::skip)]
204    pub async fn run_with_shutdown<F>(self, port: u16, shutdown: F) -> Result<(), ServerError>
205    where
206        F: std::future::Future<Output = ()> + Send + 'static,
207    {
208        let lifecycle = self.lifecycle.clone();
209        let result = self.run_inner(port, shutdown).await;
210        if result.is_err()
211            && let Some(lifecycle) = lifecycle
212        {
213            lifecycle.mark_stopped();
214        }
215        result
216    }
217
218    async fn run_inner<F>(mut self, port: u16, shutdown: F) -> Result<(), ServerError>
219    where
220        F: std::future::Future<Output = ()> + Send + 'static,
221    {
222        let dotenv = Self::load_dotenv_values().await?;
223        #[cfg(feature = "orm")]
224        let _ = crate::artisan::check_and_run_artisan(vec![], vec![]).await;
225        let _ = crate::telemetry::init_telemetry();
226        let app_config = Self::load_config().await?;
227        let environment = resolve_environment(&app_config, &dotenv)?;
228
229        self.init_database(&app_config, &dotenv).await?;
230        let addr = Self::setup_networking(port, app_config.app.port, environment, &dotenv)?;
231        let scheduler_handle = self.start_scheduler()?;
232        let shield_lifecycle = self.start_traffic_shield()?;
233
234        let server_result = if let Some(lib_path) = self.hot_reload_lib.take() {
235            self.run_hot_reload(lib_path, addr, environment, shutdown)
236                .await
237        } else {
238            self.run_static(app_config, addr, environment, shutdown)
239                .await
240        };
241
242        if let Some(shield) = shield_lifecycle {
243            shield.shutdown();
244        }
245        let scheduler_result = match scheduler_handle {
246            Some(handle) => handle.shutdown().await.map_err(ServerError::from),
247            None => Ok(()),
248        };
249
250        match server_result {
251            Err(error) => Err(error),
252            Ok(()) => scheduler_result,
253        }
254    }
255
256    #[cfg_attr(mutants, mutants::skip)]
257    async fn load_dotenv_values() -> Result<HashMap<String, String>, ServerError> {
258        if !std::path::Path::new(".env").exists() {
259            return Ok(HashMap::new());
260        }
261
262        let content = tokio::fs::read_to_string(".env").await?;
263        dotenvy::from_read_iter(content.as_bytes())
264            .map(|entry| entry.map_err(|error| ServerError::Configuration(error.to_string())))
265            .collect()
266    }
267
268    #[cfg_attr(mutants, mutants::skip)]
269    async fn load_config() -> Result<crate::config::RullstConfig, ServerError> {
270        let app_config = if std::path::Path::new("Rullst.toml").exists() {
271            crate::config::RullstConfig::load_from_file("Rullst.toml")
272                .await
273                .map_err(|error| ServerError::Configuration(error.to_string()))?
274        } else {
275            crate::config::RullstConfig::new()
276        };
277
278        app_config
279            .validate()
280            .map_err(|error| ServerError::Configuration(error.to_string()))?;
281        let _ = crate::config::RullstConfig::set_global(app_config.clone());
282        Ok(app_config)
283    }
284
285    #[cfg(feature = "orm")]
286    #[cfg_attr(mutants, mutants::skip)]
287    async fn init_database(
288        &mut self,
289        app_config: &crate::config::RullstConfig,
290        dotenv: &HashMap<String, String>,
291    ) -> Result<(), ServerError> {
292        if rullst_orm::Orm::try_pool().is_ok() {
293            return Ok(());
294        }
295
296        if self.db_url.is_none() {
297            if let Some(env_db_url) = read_optional_environment_variable("DATABASE_URL")? {
298                self.db_url = Some(env_db_url);
299            } else if let Some(dotenv_db_url) = dotenv.get("DATABASE_URL") {
300                self.db_url = Some(dotenv_db_url.clone());
301            } else if let Some(ref url) = app_config.database.url {
302                self.db_url = Some(url.clone());
303            }
304        }
305
306        if let Some(db_url) = &self.db_url {
307            println!("Initializing Orm database pool...");
308            Orm::init(db_url)
309                .await
310                .map_err(|error| ServerError::Database(error.to_string()))?;
311            println!("Database initialized successfully.");
312        }
313
314        Ok(())
315    }
316
317    #[cfg(not(feature = "orm"))]
318    #[cfg_attr(mutants, mutants::skip)]
319    async fn init_database(
320        &mut self,
321        app_config: &crate::config::RullstConfig,
322        dotenv: &HashMap<String, String>,
323    ) -> Result<(), ServerError> {
324        let database_requested = self.db_url.is_some()
325            || app_config.database.url.is_some()
326            || dotenv.contains_key("DATABASE_URL")
327            || read_optional_environment_variable("DATABASE_URL")?.is_some();
328
329        if database_requested {
330            return Err(ServerError::Database(
331                "rullst-core was compiled without the `orm` feature".to_string(),
332            ));
333        }
334
335        Ok(())
336    }
337
338    #[cfg_attr(mutants, mutants::skip)]
339    fn start_scheduler(&mut self) -> Result<Option<SchedulerHandle>, ServerError> {
340        self.scheduler
341            .take()
342            .map(Scheduler::start)
343            .transpose()
344            .map_err(ServerError::from)
345    }
346
347    #[cfg_attr(mutants, mutants::skip)]
348    fn start_traffic_shield(
349        &self,
350    ) -> Result<Option<crate::resilience::TrafficShield>, ServerError> {
351        let Some(shield) = self.shield.clone() else {
352            return Ok(None);
353        };
354        shield.start().map_err(ServerError::from)?;
355        Ok(Some(shield))
356    }
357
358    #[cfg_attr(mutants, mutants::skip)]
359    fn setup_networking(
360        fallback_port: u16,
361        configured_port: Option<u16>,
362        environment: crate::config::Environment,
363        dotenv: &HashMap<String, String>,
364    ) -> Result<SocketAddr, ServerError> {
365        let host_str = read_optional_environment_variable("HOST")?
366            .or(read_optional_environment_variable("RULLST_HOST")?)
367            .or_else(|| dotenv.get("HOST").cloned())
368            .or_else(|| dotenv.get("RULLST_HOST").cloned())
369            .unwrap_or_else(|| {
370                if environment.requires_secure_defaults() {
371                    "0.0.0.0".to_string()
372                } else {
373                    "127.0.0.1".to_string()
374                }
375            });
376
377        let env_port_value =
378            read_optional_environment_variable("PORT")?.or_else(|| dotenv.get("PORT").cloned());
379        let env_port = match env_port_value {
380            Some(value) => Some(value.parse::<u16>().map_err(|_| {
381                ServerError::Configuration(format!("PORT must be a valid u16, got `{value}`"))
382            })?),
383            None => None,
384        };
385        let port = env_port.or(configured_port).unwrap_or(fallback_port);
386
387        let addr: SocketAddr =
388            format!("{host_str}:{port}")
389                .parse()
390                .map_err(|_| ServerError::InvalidAddress {
391                    host: host_str,
392                    port,
393                })?;
394
395        if environment.allows_development_tools() && addr.ip().is_unspecified() {
396            eprintln!(
397                "⚠️  Rullst Dev: Self-Healing Console mounted on /_rullst/*\n\
398                   Set RULLST_ENV=production to disable before deploying."
399            );
400        }
401
402        Ok(addr)
403    }
404
405    #[cfg_attr(mutants, mutants::skip)]
406    async fn run_hot_reload<F>(
407        self,
408        lib_path: String,
409        addr: SocketAddr,
410        environment: crate::config::Environment,
411        shutdown: F,
412    ) -> Result<(), ServerError>
413    where
414        F: std::future::Future<Output = ()> + Send + 'static,
415    {
416        if !cfg!(debug_assertions) || !environment.allows_development_tools() {
417            return Err(ServerError::HotReloadDisabled);
418        }
419
420        let is_dev = true;
421        let reload_token = resolve_hot_reload_token()?;
422
423        println!(
424            "\x1b[36mRullst: initializing the authenticated development-library reload boundary...\x1b[0m"
425        );
426
427        let (initial_router, library) = match load_dylib_router(&lib_path, is_dev) {
428            Ok(r) => r,
429            Err(e) => {
430                println!(
431                    "\x1b[31m❌ Failed to load initial dylib: {}. Make sure the dynamic library was compiled by running 'cargo build --lib'.\x1b[0m",
432                    e
433                );
434                return Err(ServerError::HotReload(e.to_string()));
435            }
436        };
437
438        let current_router = Arc::new(RwLock::new(initial_router));
439        let active_libraries = Arc::new(Mutex::new(vec![library]));
440        let (hmr_sender, _receiver) = tokio::sync::broadcast::channel(32);
441
442        let hotswap_service = HotSwapService {
443            current_router: current_router.clone(),
444            active_libraries: active_libraries.clone(),
445            hmr_sender,
446            reload_lock: Arc::new(tokio::sync::Mutex::new(())),
447            reload_token,
448            lib_path: lib_path.clone(),
449            is_dev,
450            shield: self.shield,
451            limiter: self.limiter,
452            lifecycle: self.lifecycle.clone(),
453        };
454
455        println!(
456            "Rullst framework serving on http://{} (authenticated development hot reload)",
457            addr
458        );
459        println!(
460            "🚀 Visit: http://localhost:{} to see the result!",
461            addr.port()
462        );
463
464        let listener = tokio::net::TcpListener::bind(addr).await?;
465        mark_lifecycle_ready(self.lifecycle.as_ref())?;
466        let lifecycle = self.lifecycle.clone();
467        let result = axum::serve(listener, hotswap_service)
468            .with_graceful_shutdown(shutdown_with_lifecycle(shutdown, lifecycle.clone()))
469            .await
470            .map_err(ServerError::from);
471        mark_lifecycle_stopped(lifecycle.as_ref());
472        result
473    }
474
475    #[cfg_attr(mutants, mutants::skip)]
476    async fn run_static<F>(
477        self,
478        app_config: crate::config::RullstConfig,
479        addr: SocketAddr,
480        environment: crate::config::Environment,
481        shutdown: F,
482    ) -> Result<(), ServerError>
483    where
484        F: std::future::Future<Output = ()> + Send + 'static,
485    {
486        let is_dev = environment.allows_development_tools();
487        let mut app = self.router.into_axum();
488        app = super::dev_reload::mount(app, is_dev, std::env::var("RULLST_DEV_GENERATION").ok());
489
490        app = app.layer(axum::middleware::from_fn(
491            |req: axum::extract::Request, next: axum::middleware::Next| async move {
492                let method = req.method().to_string();
493                let path = req.uri().path().to_string();
494                let start = std::time::Instant::now();
495                let res = next.run(req).await;
496                let status = res.status().as_u16();
497                let elapsed = start.elapsed().as_secs_f64() * 1000.0;
498                if !path.starts_with("/_rullst_hmr") {
499                    println!(
500                        "[HTTP] {} {} -> {} ({:.2} ms)",
501                        method, path, status, elapsed
502                    );
503                }
504                res
505            },
506        ));
507
508        if std::path::Path::new("static").exists() {
509            app = app
510                .nest_service(
511                    "/static",
512                    tower_http::services::ServeDir::new("static").precompressed_br(),
513                )
514                .layer(axum::middleware::from_fn(zstd_static_middleware));
515        }
516
517        if is_dev {
518            app = app
519                .route(
520                    "/_rullst/explain",
521                    axum::routing::get(crate::error_console::handle_explain),
522                )
523                .route(
524                    "/_rullst/autofix",
525                    axum::routing::post(crate::error_console::handle_autofix),
526                )
527                .layer(axum::middleware::from_fn(
528                    crate::error_console::catch_panic_middleware,
529                ));
530        }
531
532        if let Some(limiter) = self.limiter {
533            app = app.layer(axum::middleware::from_fn(move |req, next| {
534                crate::resilience::rate_limit_middleware(limiter.clone(), req, next)
535            }));
536        }
537
538        if let Some(shield) = self.shield {
539            app = app.layer(axum::middleware::from_fn(move |req, next| {
540                crate::resilience::backpressure_middleware(shield.clone(), req, next)
541            }));
542        }
543
544        if let Some(lifecycle) = self.lifecycle.clone() {
545            app = apply_lifecycle(app, lifecycle);
546        }
547
548        app = if let Some(policy) = self.machine_endpoints {
549            crate::security::apply_security_baseline_with_machine_endpoints(
550                app,
551                app_config.security,
552                environment,
553                policy,
554            )
555        } else {
556            crate::security::apply_security_baseline(app, app_config.security, environment)
557        }
558        .map_err(|error| ServerError::Configuration(error.to_string()))?;
559
560        println!("Rullst framework serving on http://{}", addr);
561        println!(
562            "🚀 Visit: http://localhost:{} to see the result!",
563            addr.port()
564        );
565
566        let listener = tokio::net::TcpListener::bind(addr).await?;
567        mark_lifecycle_ready(self.lifecycle.as_ref())?;
568        let lifecycle = self.lifecycle.clone();
569        let result = axum::serve(
570            listener,
571            app.into_make_service_with_connect_info::<std::net::SocketAddr>(),
572        )
573        .with_graceful_shutdown(shutdown_with_lifecycle(shutdown, lifecycle.clone()))
574        .await
575        .map_err(ServerError::from);
576        mark_lifecycle_stopped(lifecycle.as_ref());
577        result
578    }
579}
580
581fn read_optional_environment_variable(name: &str) -> Result<Option<String>, ServerError> {
582    match std::env::var(name) {
583        Ok(value) => Ok(Some(value)),
584        Err(std::env::VarError::NotPresent) => Ok(None),
585        Err(std::env::VarError::NotUnicode(_)) => Err(ServerError::Configuration(format!(
586            "{name} is not valid Unicode"
587        ))),
588    }
589}
590
591fn resolve_hot_reload_token() -> Result<Arc<str>, ServerError> {
592    let token = read_optional_environment_variable("RULLST_HMR_TOKEN")?.ok_or_else(|| {
593        ServerError::HotReloadConfiguration(
594            "RULLST_HMR_TOKEN is missing; start hot reload through `cargo rullst dev` or `cargo rullst dash`"
595                .to_string(),
596        )
597    })?;
598    if token.len() != 64 || !token.bytes().all(|byte| byte.is_ascii_hexdigit()) {
599        return Err(ServerError::HotReloadConfiguration(
600            "RULLST_HMR_TOKEN must contain exactly 64 hexadecimal characters".to_string(),
601        ));
602    }
603    Ok(Arc::from(token))
604}
605
606fn resolve_environment(
607    config: &crate::config::RullstConfig,
608    dotenv: &HashMap<String, String>,
609) -> Result<crate::config::Environment, ServerError> {
610    let rullst_env = read_optional_environment_variable("RULLST_ENV")?;
611    let app_env = read_optional_environment_variable("APP_ENV")?;
612    let fallback = dotenv
613        .get("RULLST_ENV")
614        .or_else(|| dotenv.get("APP_ENV"))
615        .map(String::as_str)
616        .or(config.app.env.as_deref());
617
618    crate::config::Environment::resolve(rullst_env.as_deref(), app_env.as_deref(), fallback)
619        .map_err(|error| ServerError::Configuration(error.to_string()))
620}
621
622/// Listens for OS termination signals (SIGINT / SIGTERM / Ctrl+C) to drain in-flight requests cleanly.
623#[cfg_attr(mutants, mutants::skip)]
624pub async fn shutdown_signal() {
625    let ctrl_c = async {
626        let _ = tokio::signal::ctrl_c().await;
627    };
628
629    #[cfg(unix)]
630    let terminate = async {
631        if let Ok(mut stream) =
632            tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
633        {
634            stream.recv().await;
635        } else {
636            std::future::pending::<()>().await;
637        }
638    };
639
640    #[cfg(not(unix))]
641    let terminate = std::future::pending::<()>();
642
643    tokio::select! {
644        _ = ctrl_c => {
645            println!("\n🛑 [Rullst Shutdown] Received SIGINT (Ctrl+C). Draining in-flight requests...");
646        },
647        _ = terminate => {
648            println!("\n🛑 [Rullst Shutdown] Received SIGTERM. Draining in-flight requests...");
649        },
650    }
651}
652
653async fn shutdown_with_lifecycle<F>(shutdown: F, lifecycle: Option<ApplicationLifecycle>)
654where
655    F: std::future::Future<Output = ()>,
656{
657    shutdown.await;
658    if let Some(lifecycle) = lifecycle {
659        let _ = lifecycle.begin_draining();
660    }
661}
662
663fn mark_lifecycle_ready(lifecycle: Option<&ApplicationLifecycle>) -> Result<(), ServerError> {
664    match lifecycle {
665        Some(lifecycle) => lifecycle.mark_ready().map_err(ServerError::from),
666        None => Ok(()),
667    }
668}
669
670fn mark_lifecycle_stopped(lifecycle: Option<&ApplicationLifecycle>) {
671    if let Some(lifecycle) = lifecycle {
672        lifecycle.mark_stopped();
673    }
674}
675
676#[cfg(test)]
677#[path = "builder_tests.rs"]
678mod tests;
rullst_core/server/builder.rs

rullst_core/server/
builder.rs
