folk_core/
server.rs

1//! `FolkServer`: the lifecycle owner for the Folk application server.
2//!
3//! Constructed by `folk-ext` (PHP extension) or in tests with `MockRuntime`.
4
5use std::sync::Arc;
6
7use anyhow::{Context, Result};
8use folk_api::{Plugin, PluginContext, RpcRegistrar};
9use tokio::sync::watch;
10use tracing::{info, warn};
11
12use crate::config::FolkConfig;
13use crate::health_registry::HealthRegistryImpl;
14use crate::logging;
15use crate::metrics_registry::MetricsRegistryImpl;
16use crate::plugin_registry::PluginRegistry;
17use crate::runtime::Runtime;
18use crate::worker_pool::WorkerPool;
19
20/// The Folk application server.
21pub struct FolkServer {
22    config: FolkConfig,
23    runtime: Arc<dyn Runtime>,
24    plugins: PluginRegistry,
25    rpc_registrar: Option<Arc<dyn RpcRegistrar>>,
26}
27
28impl FolkServer {
29    /// Construct a server with the given config and runtime.
30    pub fn new(config: FolkConfig, runtime: Arc<dyn Runtime>) -> Self {
31        Self {
32            config,
33            runtime,
34            plugins: PluginRegistry::new(),
35            rpc_registrar: None,
36        }
37    }
38
39    /// Set an in-process RPC registrar for plugin method registration.
40    /// Plugins will register their handlers here during boot.
41    pub fn set_rpc_registrar(&mut self, registrar: Arc<dyn RpcRegistrar>) {
42        self.rpc_registrar = Some(registrar);
43    }
44
45    /// Register a plugin. Call before `run`.
46    pub fn register_plugin(&mut self, plugin: Box<dyn Plugin>) {
47        self.plugins.register(plugin);
48    }
49
50    /// Run the server until SIGTERM/SIGINT.
51    pub async fn run(mut self) -> Result<()> {
52        let _ = logging::init(&self.config.log);
53
54        info!(
55            version = folk_api::FOLK_API_VERSION,
56            workers = self.config.workers.count,
57            "folk server starting"
58        );
59
60        let (shutdown_tx, shutdown_rx) = watch::channel(false);
61
62        let health_registry = HealthRegistryImpl::new();
63        let metrics_registry = MetricsRegistryImpl::new();
64
65        let pool = WorkerPool::new(self.runtime.clone(), self.config.workers.clone())
66            .context("failed to start worker pool")?;
67
68        info!("worker pool started");
69
70        let ctx = PluginContext {
71            executor: pool.clone(),
72            shutdown: shutdown_rx.clone(),
73            rpc_registrar: self.rpc_registrar.clone(),
74            health_registry: Some(health_registry.clone()),
75            metrics_registry: Some(metrics_registry.clone()),
76        };
77
78        self.plugins
79            .boot_all(&ctx)
80            .await
81            .context("plugin boot failed")?;
82
83        info!(plugins = ?self.plugins.names(), "all plugins booted");
84
85        // Wait for shutdown signal.
86        wait_for_signal().await;
87        info!("shutdown signal received; draining");
88
89        let _ = shutdown_tx.send(true);
90
91        let timeout = self.config.server.shutdown_timeout;
92        let shutdown_result =
93            tokio::time::timeout(timeout, async { self.plugins.shutdown_all().await }).await;
94
95        if shutdown_result.is_err() {
96            warn!(?timeout, "graceful shutdown timed out; forcing");
97        }
98
99        info!("folk server stopped");
100        Ok(())
101    }
102}
103
104#[cfg(not(target_os = "windows"))]
105async fn wait_for_signal() {
106    use tokio::signal::unix::{SignalKind, signal};
107    let mut sigterm = signal(SignalKind::terminate()).expect("SIGTERM handler");
108    let mut sigint = signal(SignalKind::interrupt()).expect("SIGINT handler");
109    tokio::select! {
110        _ = sigterm.recv() => { info!("received SIGTERM"); },
111        _ = sigint.recv() => { info!("received SIGINT"); },
112    }
113}
114
115#[cfg(target_os = "windows")]
116async fn wait_for_signal() {
117    tokio::signal::ctrl_c().await.expect("ctrl-c");
118}
folk_core/server.rs

folk_core/
server.rs
