faucet_cli/serve/
state.rs

1//! Shared, cheaply-cloneable server state handed to every handler via
2//! `axum::extract::State`. Holds auth, the Prometheus render handle, the
3//! server-wide shutdown token, the run registry, the execution semaphore, the
4//! run-history backend, and the `--default-config` merge base.
5
6use crate::serve::config::{AuthMode, ServeConfig};
7use crate::serve::history::RunHistory;
8use crate::serve::logs::LogHub;
9use crate::serve::registry::Registry;
10use metrics_exporter_prometheus::PrometheusHandle;
11use serde_json::Value;
12use std::path::PathBuf;
13use std::sync::{Arc, RwLock};
14use std::time::Duration;
15use tokio::sync::Semaphore;
16use tokio_util::sync::CancellationToken;
17
18#[derive(Clone)]
19pub struct ServerState {
20    inner: Arc<Inner>,
21}
22
23struct Inner {
24    auth: AuthMode,
25    prometheus: Option<PrometheusHandle>,
26    shutdown: CancellationToken,
27    registry: Registry,
28    semaphore: Arc<Semaphore>,
29    history: Arc<dyn RunHistory>,
30    log_hub: LogHub,
31    /// The `--default-config` merge base, hot-reloadable via `POST /v1/reload`
32    /// (#198). An `RwLock` so a reload can swap it while runs read it.
33    default_base: RwLock<Option<Value>>,
34    /// Path the default-config was loaded from, so a reload can re-read it.
35    default_config_path: Option<PathBuf>,
36    idempotency_retention: Duration,
37    probe_timeout: Duration,
38    cluster: crate::serve::cluster::ClusterHandle,
39    #[cfg(feature = "triggers")]
40    triggers: crate::serve::triggers::health::TriggersHandle,
41}
42
43impl ServerState {
44    #[allow(clippy::too_many_arguments)]
45    pub fn new(
46        config: &ServeConfig,
47        prometheus: Option<PrometheusHandle>,
48        shutdown: CancellationToken,
49        history: Arc<dyn RunHistory>,
50        log_hub: LogHub,
51        default_base: Option<Value>,
52        #[cfg(feature = "triggers")] triggers: crate::serve::triggers::health::TriggersHandle,
53    ) -> Self {
54        Self {
55            inner: Arc::new(Inner {
56                auth: config.auth.clone(),
57                prometheus,
58                shutdown,
59                registry: Registry::new(config.max_queued_runs),
60                semaphore: Arc::new(Semaphore::new(config.max_concurrent_runs)),
61                history,
62                log_hub,
63                default_base: RwLock::new(default_base),
64                default_config_path: config.default_config_path.clone(),
65                idempotency_retention: config.idempotency_retention,
66                probe_timeout: config.probe_timeout,
67                cluster: crate::serve::cluster::ClusterHandle::from_config(config),
68                #[cfg(feature = "triggers")]
69                triggers,
70            }),
71        }
72    }
73
74    pub fn auth_token(&self) -> Option<&str> {
75        match &self.inner.auth {
76            AuthMode::Token(t) => Some(t),
77            AuthMode::Rbac(_) | AuthMode::None => None,
78        }
79    }
80
81    /// The configured authentication mode (bearer resolution + RBAC).
82    pub fn auth_mode(&self) -> &AuthMode {
83        &self.inner.auth
84    }
85
86    pub fn render_metrics(&self) -> Option<String> {
87        self.inner.prometheus.as_ref().map(|h| h.render())
88    }
89
90    pub fn shutdown_token(&self) -> CancellationToken {
91        self.inner.shutdown.clone()
92    }
93
94    pub fn registry(&self) -> &Registry {
95        &self.inner.registry
96    }
97
98    pub fn semaphore(&self) -> Arc<Semaphore> {
99        Arc::clone(&self.inner.semaphore)
100    }
101
102    pub fn history(&self) -> Arc<dyn RunHistory> {
103        Arc::clone(&self.inner.history)
104    }
105
106    /// The per-run log buffer registry shared with the tracing [`LogHub`] layer.
107    pub fn log_hub(&self) -> &LogHub {
108        &self.inner.log_hub
109    }
110
111    /// A snapshot of the `--default-config` merge base (cloned under the read
112    /// lock, so a concurrent hot reload can't tear it).
113    pub fn default_base(&self) -> Option<Value> {
114        self.inner.default_base.read().unwrap().clone()
115    }
116
117    /// Path the default-config was loaded from (`None` when `--default-config`
118    /// was not passed).
119    pub fn default_config_path(&self) -> Option<&PathBuf> {
120        self.inner.default_config_path.as_ref()
121    }
122
123    /// Atomically swap the `--default-config` merge base (hot reload, #198).
124    pub fn set_default_base(&self, base: Option<Value>) {
125        *self.inner.default_base.write().unwrap() = base;
126    }
127
128    pub fn idempotency_retention(&self) -> Duration {
129        self.inner.idempotency_retention
130    }
131
132    pub fn probe_timeout(&self) -> Duration {
133        self.inner.probe_timeout
134    }
135
136    pub fn cluster(&self) -> &crate::serve::cluster::ClusterHandle {
137        &self.inner.cluster
138    }
139
140    #[cfg(feature = "triggers")]
141    pub fn triggers(&self) -> &crate::serve::triggers::health::TriggersHandle {
142        &self.inner.triggers
143    }
144}
145
146#[cfg(test)]
147mod tests {
148    use super::*;
149    use crate::serve::config::HistoryBackendSpec;
150    use crate::serve::history::memory::MemoryHistory;
151
152    fn cfg(auth: AuthMode) -> ServeConfig {
153        ServeConfig {
154            listen: "127.0.0.1:0".parse().unwrap(),
155            auth,
156            max_concurrent_runs: 4,
157            max_queued_runs: 32,
158            default_config_path: None,
159            history: HistoryBackendSpec::Memory,
160            cors_origins: vec![],
161            body_limit_bytes: 1_048_576,
162            shutdown_grace: Duration::from_secs(60),
163            retain_terminal_runs: Duration::from_secs(60),
164            idempotency_retention: Duration::from_secs(60),
165            lease_ttl: Duration::from_secs(30),
166            probe_timeout: Duration::from_secs(10),
167            env_file: None,
168            no_env_file: false,
169            log_level: "info".into(),
170            ui_enabled: true,
171            cluster: crate::serve::cluster::ClusterConfig::disabled(),
172            triggers_path: None,
173        }
174    }
175
176    fn state(auth: AuthMode) -> ServerState {
177        use crate::serve::logs::LogHub;
178        let history = Arc::new(MemoryHistory::new(Duration::from_secs(60))) as Arc<dyn RunHistory>;
179        ServerState::new(
180            &cfg(auth),
181            None,
182            CancellationToken::new(),
183            history,
184            LogHub::new(),
185            None,
186            #[cfg(feature = "triggers")]
187            crate::serve::triggers::health::TriggersHandle::empty(),
188        )
189    }
190
191    #[test]
192    fn auth_token_reflects_mode() {
193        assert_eq!(state(AuthMode::Token("x".into())).auth_token(), Some("x"));
194        assert_eq!(state(AuthMode::None).auth_token(), None);
195    }
196
197    #[test]
198    fn render_metrics_none_without_handle() {
199        assert!(state(AuthMode::None).render_metrics().is_none());
200    }
201
202    #[test]
203    fn default_base_swaps_atomically() {
204        let s = state(AuthMode::None);
205        assert!(s.default_base().is_none());
206        assert!(s.default_config_path().is_none());
207        s.set_default_base(Some(serde_json::json!({"version": 1})));
208        assert_eq!(s.default_base(), Some(serde_json::json!({"version": 1})));
209        s.set_default_base(None);
210        assert!(s.default_base().is_none());
211    }
212
213    // The reload handler is a no-op (200 `reloaded:false`) when the server was
214    // started without `--default-config`.
215    #[tokio::test]
216    async fn reload_handler_noop_without_default_config() {
217        let s = state(AuthMode::None);
218        let actor = crate::serve::rbac::AuthContext {
219            principal: "admin".into(),
220            role: crate::serve::rbac::Role::Admin,
221            source_ip: None,
222        };
223        let axum::Json(body) = crate::serve::handlers::reload::reload(
224            axum::extract::State(s),
225            axum::extract::Extension(actor),
226        )
227        .await
228        .expect("reload ok");
229        assert_eq!(body["reloaded"], serde_json::json!(false));
230    }
231
232    #[test]
233    fn registry_starts_empty() {
234        let s = state(AuthMode::None);
235        assert_eq!(s.registry().queued(), 0);
236        assert_eq!(s.registry().in_flight(), 0);
237    }
238}
faucet_cli/serve/state.rs

faucet_cli/serve/
state.rs
