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#[derive(Debug, thiserror::Error)]
15#[non_exhaustive]
16pub enum ServerError {
17 #[error("server configuration error: {0}")]
19 Configuration(String),
20
21 #[error("database initialization failed: {0}")]
23 Database(String),
24
25 #[error("scheduler lifecycle failed: {0}")]
27 Scheduler(#[from] crate::scheduler::SchedulerError),
28
29 #[error("traffic shield lifecycle failed: {0}")]
31 TrafficShield(#[from] crate::resilience::TrafficShieldError),
32
33 #[error("application lifecycle failed: {0}")]
35 Lifecycle(#[from] crate::lifecycle::ApplicationLifecycleError),
36
37 #[error("invalid server listen address `{host}:{port}`")]
39 InvalidAddress {
40 host: String,
42 port: u16,
44 },
45
46 #[error("hot reload is available only in local development debug builds")]
48 HotReloadDisabled,
49
50 #[error("hot reload failed: {0}")]
52 HotReload(String),
53
54 #[error("hot reload configuration error: {0}")]
56 HotReloadConfiguration(String),
57
58 #[error("server I/O error: {0}")]
60 Io(#[from] std::io::Error),
61}
62
63#[non_exhaustive]
64pub 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 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 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 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 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 pub fn schedule(mut self, scheduler: Scheduler) -> Self {
169 self.scheduler = Some(scheduler);
170 self
171 }
172
173 pub fn shield(mut self, shield: crate::resilience::TrafficShield) -> Self {
177 self.shield = Some(shield);
178 self
179 }
180
181 pub fn rate_limit(mut self, limiter: crate::resilience::RateLimiter) -> Self {
185 self.limiter = Some(limiter);
186 self
187 }
188
189 pub fn with_lifecycle(mut self, lifecycle: ApplicationLifecycle) -> Self {
197 self.lifecycle = Some(lifecycle);
198 self
199 }
200
201 #[cfg_attr(mutants, mutants::skip)]
203 pub async fn run(self, port: u16) -> Result<(), ServerError> {
204 self.run_with_shutdown(port, shutdown_signal()).await
205 }
206
207 #[cfg_attr(mutants, mutants::skip)]
213 pub async fn run_with_shutdown<F>(self, port: u16, shutdown: F) -> Result<(), ServerError>
214 where
215 F: std::future::Future<Output = ()> + Send + 'static,
216 {
217 let lifecycle = self.lifecycle.clone();
218 let result = self.run_inner(port, shutdown).await;
219 if result.is_err()
220 && let Some(lifecycle) = lifecycle
221 {
222 lifecycle.mark_stopped();
223 }
224 result
225 }
226
227 async fn run_inner<F>(mut self, port: u16, shutdown: F) -> Result<(), ServerError>
228 where
229 F: std::future::Future<Output = ()> + Send + 'static,
230 {
231 let dotenv = Self::load_dotenv_values().await?;
232 #[cfg(feature = "orm")]
233 crate::artisan::runner::intercept_artisan_command(None, self.db_url.as_deref()).await;
234 let _ = crate::telemetry::init_telemetry();
235 let app_config = Self::load_config().await?;
236 let environment = resolve_environment(&app_config, &dotenv)?;
237
238 self.init_database(&app_config, &dotenv).await?;
239 let addr = Self::setup_networking(port, app_config.app.port, environment, &dotenv)?;
240 let mut scheduler_handle = self.start_scheduler()?;
241 let shield_lifecycle = self.start_traffic_shield()?;
242
243 let server = async move {
244 if let Some(lib_path) = self.hot_reload_lib.take() {
245 self.run_hot_reload(lib_path, addr, environment, shutdown)
246 .await
247 } else {
248 self.run_static(app_config, addr, environment, shutdown)
249 .await
250 }
251 };
252 let server_result =
255 super::scheduler_supervision::serve_while_draining(server, scheduler_handle.as_mut())
256 .await;
257
258 if let Some(shield) = shield_lifecycle {
259 shield.shutdown();
260 }
261 let scheduler_result = super::scheduler_supervision::stop_scheduler(scheduler_handle).await;
262
263 match server_result {
264 Err(error) => Err(error),
265 Ok(()) => scheduler_result,
266 }
267 }
268
269 #[cfg_attr(mutants, mutants::skip)]
270 async fn load_dotenv_values() -> Result<HashMap<String, String>, ServerError> {
271 super::database_url::load_dotenv_file(std::path::Path::new(".env")).await
272 }
273
274 #[cfg_attr(mutants, mutants::skip)]
275 async fn load_config() -> Result<crate::config::RullstConfig, ServerError> {
276 let app_config =
277 super::database_url::load_config_file(std::path::Path::new("Rullst.toml")).await?;
278
279 app_config
280 .validate()
281 .map_err(|error| ServerError::Configuration(error.to_string()))?;
282 let _ = crate::config::RullstConfig::set_global(app_config.clone());
283 Ok(app_config)
284 }
285
286 #[cfg(feature = "orm")]
287 #[cfg_attr(mutants, mutants::skip)]
288 async fn init_database(
289 &mut self,
290 app_config: &crate::config::RullstConfig,
291 dotenv: &HashMap<String, String>,
292 ) -> Result<(), ServerError> {
293 if rullst_orm::Orm::try_pool().is_ok() {
294 return Ok(());
295 }
296
297 if self.db_url.is_none() {
298 self.db_url = super::database_url::resolve_database_url(
299 None,
300 read_optional_environment_variable,
301 dotenv,
302 app_config,
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 = super::database_url::resolve_database_url(
325 self.db_url.as_deref(),
326 read_optional_environment_variable,
327 dotenv,
328 app_config,
329 )?
330 .is_some();
331
332 if database_requested {
333 return Err(ServerError::Database(
334 "rullst-core was compiled without the `orm` feature".to_string(),
335 ));
336 }
337
338 Ok(())
339 }
340
341 #[cfg_attr(mutants, mutants::skip)]
342 fn start_scheduler(&mut self) -> Result<Option<SchedulerHandle>, ServerError> {
343 self.scheduler
344 .take()
345 .map(Scheduler::start)
346 .transpose()
347 .map_err(ServerError::from)
348 }
349
350 #[cfg_attr(mutants, mutants::skip)]
351 fn start_traffic_shield(
352 &self,
353 ) -> Result<Option<crate::resilience::TrafficShield>, ServerError> {
354 let Some(shield) = self.shield.clone() else {
355 return Ok(None);
356 };
357 shield.start().map_err(ServerError::from)?;
358 Ok(Some(shield))
359 }
360
361 #[cfg_attr(mutants, mutants::skip)]
362 fn setup_networking(
363 fallback_port: u16,
364 configured_port: Option<u16>,
365 environment: crate::config::Environment,
366 dotenv: &HashMap<String, String>,
367 ) -> Result<SocketAddr, ServerError> {
368 let host_str = read_optional_environment_variable("HOST")?
369 .or(read_optional_environment_variable("RULLST_HOST")?)
370 .or_else(|| dotenv.get("HOST").cloned())
371 .or_else(|| dotenv.get("RULLST_HOST").cloned())
372 .unwrap_or_else(|| {
373 if environment.requires_secure_defaults() {
374 "0.0.0.0".to_string()
375 } else {
376 "127.0.0.1".to_string()
377 }
378 });
379
380 let env_port_value =
381 read_optional_environment_variable("PORT")?.or_else(|| dotenv.get("PORT").cloned());
382 let env_port = match env_port_value {
383 Some(value) => Some(value.parse::<u16>().map_err(|_| {
384 ServerError::Configuration(format!("PORT must be a valid u16, got `{value}`"))
385 })?),
386 None => None,
387 };
388 let port = env_port.or(configured_port).unwrap_or(fallback_port);
389
390 let addr: SocketAddr =
391 format!("{host_str}:{port}")
392 .parse()
393 .map_err(|_| ServerError::InvalidAddress {
394 host: host_str,
395 port,
396 })?;
397
398 if development_console_enabled(cfg!(debug_assertions), environment)
399 && addr.ip().is_unspecified()
400 {
401 eprintln!(
402 "⚠️ Rullst Dev: Self-Healing Console mounted on /_rullst/*\n\
403 Set RULLST_ENV=production to disable before deploying."
404 );
405 }
406
407 Ok(addr)
408 }
409
410 #[cfg_attr(mutants, mutants::skip)]
411 async fn run_hot_reload<F>(
412 self,
413 lib_path: String,
414 addr: SocketAddr,
415 environment: crate::config::Environment,
416 shutdown: F,
417 ) -> Result<(), ServerError>
418 where
419 F: std::future::Future<Output = ()> + Send + 'static,
420 {
421 if !cfg!(debug_assertions) || !environment.allows_development_tools() {
422 return Err(ServerError::HotReloadDisabled);
423 }
424
425 let is_dev = true;
426 let reload_token = resolve_hot_reload_token()?;
427
428 println!(
429 "\x1b[36mRullst: initializing the authenticated development-library reload boundary...\x1b[0m"
430 );
431
432 let (initial_router, library) = match load_dylib_router(&lib_path, is_dev) {
433 Ok(r) => r,
434 Err(e) => {
435 println!(
436 "\x1b[31m❌ Failed to load initial dylib: {}. Make sure the dynamic library was compiled by running 'cargo build --lib'.\x1b[0m",
437 e
438 );
439 return Err(ServerError::HotReload(e.to_string()));
440 }
441 };
442
443 let current_router = Arc::new(RwLock::new(initial_router));
444 let active_libraries = Arc::new(Mutex::new(vec![library]));
445 let (hmr_sender, _receiver) = tokio::sync::broadcast::channel(32);
446
447 let hotswap_service = HotSwapService {
448 current_router: current_router.clone(),
449 active_libraries: active_libraries.clone(),
450 hmr_sender,
451 reload_lock: Arc::new(tokio::sync::Mutex::new(())),
452 reload_token,
453 lib_path: lib_path.clone(),
454 is_dev,
455 shield: self.shield,
456 limiter: self.limiter,
457 lifecycle: self.lifecycle.clone(),
458 };
459
460 println!(
461 "Rullst framework serving on http://{} (authenticated development hot reload)",
462 addr
463 );
464 println!(
465 "🚀 Visit: http://localhost:{} to see the result!",
466 addr.port()
467 );
468
469 let listener = tokio::net::TcpListener::bind(addr).await?;
470 mark_lifecycle_ready(self.lifecycle.as_ref())?;
471 let lifecycle = self.lifecycle.clone();
472 let result = axum::serve(listener, hotswap_service)
473 .with_graceful_shutdown(shutdown_with_lifecycle(shutdown, lifecycle.clone()))
474 .await
475 .map_err(ServerError::from);
476 mark_lifecycle_stopped(lifecycle.as_ref());
477 result
478 }
479
480 #[cfg_attr(mutants, mutants::skip)]
481 async fn run_static<F>(
482 self,
483 app_config: crate::config::RullstConfig,
484 addr: SocketAddr,
485 environment: crate::config::Environment,
486 shutdown: F,
487 ) -> Result<(), ServerError>
488 where
489 F: std::future::Future<Output = ()> + Send + 'static,
490 {
491 let is_dev = environment.allows_development_tools();
492 let mut app = self.router.into_axum();
493 app = super::dev_reload::mount(app, is_dev, std::env::var("RULLST_DEV_GENERATION").ok());
494
495 app = app.layer(axum::middleware::from_fn(
496 super::console::access_log_middleware,
497 ));
498
499 if std::path::Path::new("static").exists() {
500 app = app
501 .nest_service(
502 "/static",
503 tower_http::services::ServeDir::new("static").precompressed_br(),
504 )
505 .layer(axum::middleware::from_fn(zstd_static_middleware));
506 }
507
508 if development_console_enabled(cfg!(debug_assertions), environment) {
509 app = app
510 .route(
511 "/_rullst/explain",
512 axum::routing::get(crate::error_console::handle_explain),
513 )
514 .route(
515 "/_rullst/autofix",
516 axum::routing::post(crate::error_console::handle_autofix),
517 )
518 .layer(axum::middleware::from_fn(
519 crate::error_console::catch_panic_middleware,
520 ));
521 }
522
523 app = super::traffic::apply_traffic_controls(app, self.limiter, self.shield);
524
525 if let Some(lifecycle) = self.lifecycle.clone() {
526 app = apply_lifecycle(app, lifecycle);
527 }
528
529 app = if let Some(policy) = self.machine_endpoints {
530 crate::security::apply_security_baseline_with_machine_endpoints(
531 app,
532 app_config.security,
533 environment,
534 policy,
535 )
536 } else {
537 crate::security::apply_security_baseline(app, app_config.security, environment)
538 }
539 .map_err(|error| ServerError::Configuration(error.to_string()))?;
540
541 println!("Rullst framework serving on http://{}", addr);
542 println!(
543 "🚀 Visit: http://localhost:{} to see the result!",
544 addr.port()
545 );
546
547 let listener = tokio::net::TcpListener::bind(addr).await?;
548 mark_lifecycle_ready(self.lifecycle.as_ref())?;
549 let lifecycle = self.lifecycle.clone();
550 let result = axum::serve(
551 listener,
552 app.into_make_service_with_connect_info::<std::net::SocketAddr>(),
553 )
554 .with_graceful_shutdown(shutdown_with_lifecycle(shutdown, lifecycle.clone()))
555 .await
556 .map_err(ServerError::from);
557 mark_lifecycle_stopped(lifecycle.as_ref());
558 result
559 }
560}
561
562pub(crate) fn read_optional_environment_variable(
563 name: &str,
564) -> Result<Option<String>, ServerError> {
565 match std::env::var(name) {
566 Ok(value) => Ok(Some(value)),
567 Err(std::env::VarError::NotPresent) => Ok(None),
568 Err(std::env::VarError::NotUnicode(_)) => Err(ServerError::Configuration(format!(
569 "{name} is not valid Unicode"
570 ))),
571 }
572}
573
574fn development_console_enabled(debug_build: bool, environment: crate::config::Environment) -> bool {
579 debug_build && environment.allows_development_tools()
580}
581
582fn resolve_hot_reload_token() -> Result<Arc<str>, ServerError> {
583 let token = read_optional_environment_variable("RULLST_HMR_TOKEN")?.ok_or_else(|| {
584 ServerError::HotReloadConfiguration(
585 "RULLST_HMR_TOKEN is missing; start hot reload through `cargo rullst dev` or `cargo rullst dash`"
586 .to_string(),
587 )
588 })?;
589 if token.len() != 64 || !token.bytes().all(|byte| byte.is_ascii_hexdigit()) {
590 return Err(ServerError::HotReloadConfiguration(
591 "RULLST_HMR_TOKEN must contain exactly 64 hexadecimal characters".to_string(),
592 ));
593 }
594 Ok(Arc::from(token))
595}
596
597fn resolve_environment(
598 config: &crate::config::RullstConfig,
599 dotenv: &HashMap<String, String>,
600) -> Result<crate::config::Environment, ServerError> {
601 let rullst_env = read_optional_environment_variable("RULLST_ENV")?;
602 let app_env = read_optional_environment_variable("APP_ENV")?;
603 let fallback = dotenv
604 .get("RULLST_ENV")
605 .or_else(|| dotenv.get("APP_ENV"))
606 .map(String::as_str)
607 .or(config.app.env.as_deref());
608
609 crate::config::Environment::resolve(rullst_env.as_deref(), app_env.as_deref(), fallback)
610 .map_err(|error| ServerError::Configuration(error.to_string()))
611}
612
613#[cfg_attr(mutants, mutants::skip)]
615pub async fn shutdown_signal() {
616 let ctrl_c = async {
617 let _ = tokio::signal::ctrl_c().await;
618 };
619
620 #[cfg(unix)]
621 let terminate = async {
622 if let Ok(mut stream) =
623 tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
624 {
625 stream.recv().await;
626 } else {
627 std::future::pending::<()>().await;
628 }
629 };
630
631 #[cfg(not(unix))]
632 let terminate = std::future::pending::<()>();
633
634 tokio::select! {
635 _ = ctrl_c => {
636 super::console::stdout_line(format_args!("\n🛑 [Rullst Shutdown] Received SIGINT (Ctrl+C). Draining in-flight requests..."));
637 },
638 _ = terminate => {
639 super::console::stdout_line(format_args!("\n🛑 [Rullst Shutdown] Received SIGTERM. Draining in-flight requests..."));
640 },
641 }
642}
643
644async fn shutdown_with_lifecycle<F>(shutdown: F, lifecycle: Option<ApplicationLifecycle>)
645where
646 F: std::future::Future<Output = ()>,
647{
648 shutdown.await;
649 if let Some(lifecycle) = lifecycle {
650 let _ = lifecycle.begin_draining();
651 }
652}
653
654fn mark_lifecycle_ready(lifecycle: Option<&ApplicationLifecycle>) -> Result<(), ServerError> {
655 match lifecycle {
656 Some(lifecycle) => lifecycle.mark_ready().map_err(ServerError::from),
657 None => Ok(()),
658 }
659}
660
661fn mark_lifecycle_stopped(lifecycle: Option<&ApplicationLifecycle>) {
662 if let Some(lifecycle) = lifecycle {
663 lifecycle.mark_stopped();
664 }
665}
666
667pub(super) fn parse_dotenv(content: &str) -> Result<HashMap<String, String>, ServerError> {
670 let mut values = HashMap::new();
671 for (index, entry) in dotenvy::from_read_iter(content.as_bytes()).enumerate() {
672 let (name, value) = entry.map_err(|error| {
673 ServerError::Configuration(match error {
674 dotenvy::Error::LineParse(..) => {
675 format!("invalid .env syntax in entry {}", index + 1)
676 }
677 dotenvy::Error::Io(error) => format!("failed to read .env: {}", error.kind()),
678 _ => "invalid .env file".to_string(),
679 })
680 })?;
681 values.insert(name, value);
682 }
683 Ok(values)
684}
685
686#[cfg(test)]
687#[path = "builder_tests.rs"]
688mod tests;