autumn_web/
state.rs

1//! Shared application state.
2//!
3//! This module defines [`AppState`], the core state object passed to all
4//! Axum route handlers. It contains framework-managed resources like the
5//! database connection pool, metrics collector, and WebSocket channels.
6//!
7//! Handlers typically don't extract `AppState` directly. Instead, they use
8//! specialized extractors like [`Db`](crate::Db) which pull what they need
9//! from the state. However, custom extractors can access the state via
10//! `crate::extract::State<AppState>`.
11
12use std::any::{Any, TypeId};
13use std::collections::HashMap;
14use std::sync::Arc;
15
16use crate::cache::Cache;
17use crate::time::{ClockSource, SystemClock};
18
19/// Newtype wrapper used to store the global cache in the extension map so that
20/// `set_cache` (called from startup hooks) is visible to all `AppState` clones.
21pub struct GlobalCacheEntry(pub Arc<dyn Cache>);
22
23use crate::actuator;
24use crate::authorization::{ForbiddenResponse, Policy, PolicyRegistry, Scope};
25#[cfg(feature = "ws")]
26use crate::channels::Channels;
27#[cfg(feature = "db")]
28use crate::db::DbState;
29use crate::middleware;
30#[cfg(feature = "presence")]
31use crate::presence::Presence;
32use crate::probe;
33#[cfg(feature = "ws")]
34use tokio_util::sync::CancellationToken;
35
36/// Shared application state passed to all route handlers.
37///
38/// Holds framework-managed resources such as the database connection pool.
39/// Axum requires handler state to be [`Clone`], so internal resources use
40/// `Arc` or are already cheaply cloneable (`deadpool::Pool` is `Arc`-wrapped
41/// internally).
42///
43/// This struct is normally constructed by [`crate::app::AppBuilder::run`] and
44/// should not need to be created manually. It is public so that custom
45/// Axum extractors can access framework resources via
46/// `State<AppState>`.
47///
48/// # Examples
49///
50/// ```rust
51/// use autumn_web::AppState;
52///
53/// // State without a database (e.g., for testing)
54/// let state = AppState::for_test().with_profile("dev");
55/// ```
56#[derive(Clone)]
57#[non_exhaustive]
58pub struct AppState {
59    /// Runtime-managed typed extensions installed by integrations after the app
60    /// state has been constructed.
61    pub(crate) extensions: Arc<std::sync::RwLock<HashMap<TypeId, Arc<dyn Any + Send + Sync>>>>,
62
63    /// Primary/write database connection pool, or `None` when no
64    /// `database.primary_url` or legacy `database.url` is configured.
65    #[cfg(feature = "db")]
66    pub(crate) pool:
67        Option<diesel_async::pooled_connection::deadpool::Pool<crate::db::RuntimeConnection>>,
68
69    /// Read-replica connection pool, or `None` when no replica role is configured.
70    #[cfg(feature = "db")]
71    pub(crate) replica_pool:
72        Option<diesel_async::pooled_connection::deadpool::Pool<crate::db::RuntimeConnection>>,
73
74    /// Configured shard set, or `None` when no `[[database.shards]]`
75    /// entries exist. The `pool`/`replica_pool` roles above are the
76    /// control topology; tenant data routes across these shards.
77    #[cfg(feature = "db")]
78    pub(crate) shards: Option<crate::sharding::ShardSet>,
79
80    /// Active profile name (e.g., "dev", "prod", "staging").
81    pub(crate) profile: Option<String>,
82
83    /// Resolved process role for this replica, after config parsing and the
84    /// `AUTUMN_ROLE` env override. This is the same value the framework uses to
85    /// gate the job runtime, scheduler, and commit-hook worker, exposed here as
86    /// a first-class accessor ([`role`](Self::role)) so `on_startup`/`on_shutdown`
87    /// hooks, plugins, and handlers can self-gate app-owned background work
88    /// without re-reading `AUTUMN_ROLE` by hand.
89    pub(crate) role: crate::config::ProcessRole,
90
91    /// When the application started. Used for uptime calculation.
92    pub(crate) started_at: std::time::Instant,
93
94    /// Whether the health endpoint should include detailed info.
95    pub(crate) health_detailed: bool,
96
97    /// Probe lifecycle state for liveness, readiness, and startup endpoints.
98    pub(crate) probes: probe::ProbeState,
99
100    /// In-memory metrics collector for the `/actuator/metrics` endpoint.
101    pub(crate) metrics: middleware::MetricsCollector,
102
103    /// Runtime log level state for the `/actuator/loggers` endpoint.
104    pub(crate) log_levels: actuator::LogLevels,
105
106    /// Scheduled task registry for the `/actuator/tasks` endpoint.
107    pub(crate) task_registry: actuator::TaskRegistry,
108    /// Job registry for the `/actuator/jobs` endpoint.
109    pub(crate) job_registry: actuator::JobRegistry,
110
111    /// Resolved config properties with source tracking for `/actuator/configprops`.
112    pub(crate) config_props: actuator::ConfigProperties,
113
114    /// Registry of plugin-contributed metrics sources, populated by
115    /// [`crate::app::AppBuilder::metrics_source`].
116    pub(crate) metrics_source_registry: actuator::MetricsSourceRegistry,
117
118    /// Registry of custom health indicators, populated by
119    /// [`crate::app::AppBuilder::health_indicator`].
120    pub(crate) health_indicator_registry: actuator::HealthIndicatorRegistry,
121
122    /// Named broadcast channel registry for real-time messaging.
123    ///
124    /// Available when the `ws` feature is enabled. Use
125    /// [`channels()`](Self::channels) for convenient access.
126    #[cfg(feature = "ws")]
127    pub(crate) channels: Channels,
128
129    /// Distributed presence tracker layered on top of [`Channels`].
130    ///
131    /// Available when the `presence` feature is enabled. Use
132    /// [`presence()`](Self::presence) for convenient access.
133    #[cfg(feature = "presence")]
134    pub(crate) presence: Presence,
135
136    /// Cancellation token signalled during graceful shutdown.
137    ///
138    /// WebSocket handlers receive a child token so they can clean up
139    /// when the server is stopping.
140    #[cfg(feature = "ws")]
141    pub(crate) shutdown: CancellationToken,
142
143    /// Per-resource policy + scope registry used by `#[authorize]`
144    /// and `#[repository(policy = ...)]`-generated handlers.
145    pub(crate) policy_registry: PolicyRegistry,
146
147    /// HTTP status returned when a [`Policy`] denies a record-level
148    /// action. Defaults to `404 Not Found` to mirror Rails / Phoenix
149    /// posture and avoid leaking record existence.
150    pub(crate) forbidden_response: ForbiddenResponse,
151
152    /// Session key the `#[authorize]` machinery reads to resolve the
153    /// authenticated user id for the
154    /// [`PolicyContext`](crate::authorization::PolicyContext).
155    /// Mirrors `[auth] session_key` (default: `"user_id"`).
156    pub(crate) auth_session_key: String,
157
158    /// Shared application cache backend. `None` means no global cache has been
159    /// registered; `#[cached]` will fall back to its per-function Moka store.
160    pub(crate) shared_cache: Option<Arc<dyn Cache>>,
161
162    /// Injected wall-clock. Defaults to [`SystemClock`] (real time).
163    /// Tests override via [`crate::test::TestApp::with_clock`].
164    pub(crate) clock: Arc<dyn ClockSource>,
165
166    /// Process-unique identity assigned once per real `AppState` construction
167    /// and preserved verbatim across `.clone()` (it is `Copy`).
168    ///
169    /// Two independently built apps that happen to share identical rate-limit
170    /// config would otherwise collide in the process-global `#[throttle]`
171    /// limiter registry (keyed only by route/name + config fingerprint), so
172    /// traffic in one app would drain the other's per-route bucket. Folding
173    /// this id into the registry key gives each app its own buckets. Sourced
174    /// from a monotonic `AtomicU64` — never reused, unlike a pointer address.
175    pub(crate) app_id: u64,
176}
177
178/// Monotonic source for [`AppState::app_id`]. Starts at 1 so `0` can never be a
179/// live app id.
180static NEXT_APP_ID: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1);
181
182impl crate::authorization::ProvideAuthorizationState for AppState {
183    fn policy_registry(&self) -> &crate::authorization::PolicyRegistry {
184        &self.policy_registry
185    }
186
187    fn auth_session_key(&self) -> &str {
188        &self.auth_session_key
189    }
190
191    fn forbidden_response(&self) -> &crate::authorization::ForbiddenResponse {
192        &self.forbidden_response
193    }
194
195    #[cfg(feature = "db")]
196    fn pool(
197        &self,
198    ) -> Option<&diesel_async::pooled_connection::deadpool::Pool<crate::db::RuntimeConnection>>
199    {
200        self.pool.as_ref()
201    }
202}
203
204impl AppState {
205    /// Install or replace a typed runtime extension.
206    ///
207    /// Integrations use this to publish typed runtime resources, such as
208    /// background-worker handles or dedicated storage pools, after startup.
209    ///
210    /// # Panics
211    ///
212    /// Panics if the internal extension map mutex is poisoned.
213    pub fn insert_extension<T>(&self, value: T)
214    where
215        T: Any + Send + Sync + 'static,
216    {
217        self.extensions
218            .write()
219            .expect("app state extension lock poisoned")
220            .insert(TypeId::of::<T>(), Arc::new(value));
221    }
222
223    /// Borrow a typed runtime extension if it has been installed.
224    ///
225    /// The returned [`Arc`] is cloned out of the internal registry so callers
226    /// do not hold the state mutex while using the value.
227    ///
228    /// # Panics
229    ///
230    /// Panics if the internal extension map mutex is poisoned.
231    #[must_use]
232    pub fn extension<T>(&self) -> Option<Arc<T>>
233    where
234        T: Any + Send + Sync + 'static,
235    {
236        self.extensions
237            .read()
238            .expect("app state extension lock poisoned")
239            .get(&TypeId::of::<T>())
240            .cloned()
241            .and_then(|value| Arc::downcast::<T>(value).ok())
242    }
243
244    /// Fetch the extension of type `T`, inserting `f()`'s result if absent.
245    /// Atomic get-or-insert under the write lock: concurrent callers share one
246    /// value. Used to lazily register process-wide registries.
247    ///
248    /// # Panics
249    ///
250    /// Panics if the internal extension map mutex is poisoned.
251    pub fn extension_or_insert_with<T>(&self, f: impl FnOnce() -> T) -> Arc<T>
252    where
253        T: Any + Send + Sync + 'static,
254    {
255        if let Some(existing) = self.extension::<T>() {
256            return existing;
257        }
258        let mut map = self
259            .extensions
260            .write()
261            .expect("app state extension lock poisoned");
262        if let Some(existing) = map
263            .get(&TypeId::of::<T>())
264            .cloned()
265            .and_then(|value| Arc::downcast::<T>(value).ok())
266        {
267            return existing;
268        }
269        let arc = Arc::new(f());
270        map.insert(TypeId::of::<T>(), arc.clone() as Arc<dyn Any + Send + Sync>);
271        arc
272    }
273
274    /// Returns the registered error reporters, if any were installed via
275    /// [`AppBuilder::with_error_reporter`](crate::app::AppBuilder::with_error_reporter).
276    ///
277    /// Returns an empty `Vec` when none are registered; the
278    /// [`ReportingLayer`](crate::reporting::ReportingLayer) then falls back to
279    /// the built-in [`LogReporter`](crate::reporting::LogReporter).
280    #[cfg(feature = "reporting")]
281    #[must_use]
282    pub(crate) fn error_reporters(
283        &self,
284    ) -> Vec<std::sync::Arc<dyn crate::reporting::ErrorReporter>> {
285        self.extension::<crate::reporting::RegisteredReporters>()
286            .map(|reporters| reporters.0.clone())
287            .unwrap_or_default()
288    }
289
290    /// Returns the database connection pool.
291    #[cfg(feature = "db")]
292    #[must_use]
293    pub const fn pool(
294        &self,
295    ) -> Option<&diesel_async::pooled_connection::deadpool::Pool<crate::db::RuntimeConnection>>
296    {
297        self.pool.as_ref()
298    }
299
300    /// Returns the read-replica database connection pool, if configured.
301    #[cfg(feature = "db")]
302    #[must_use]
303    pub const fn replica_pool(
304        &self,
305    ) -> Option<&diesel_async::pooled_connection::deadpool::Pool<crate::db::RuntimeConnection>>
306    {
307        self.replica_pool.as_ref()
308    }
309
310    /// Returns the configured shard set, when `[[database.shards]]`
311    /// entries exist.
312    ///
313    /// The control roles ([`pool`](Self::pool)/[`replica_pool`](Self::replica_pool))
314    /// are unaffected by sharding; framework state lives there.
315    #[cfg(feature = "db")]
316    #[must_use]
317    pub const fn shards(&self) -> Option<&crate::sharding::ShardSet> {
318        self.shards.as_ref()
319    }
320
321    /// Returns the pool used for read-only work.
322    #[cfg(feature = "db")]
323    #[must_use]
324    pub fn read_pool(
325        &self,
326    ) -> Option<&diesel_async::pooled_connection::deadpool::Pool<crate::db::RuntimeConnection>>
327    {
328        if self.replica_pool.is_some() && self.probes.should_route_reads_to_replica() {
329            self.replica_pool.as_ref()
330        } else if self.replica_pool.is_some() && self.probes.should_fallback_reads_to_primary() {
331            self.pool.as_ref()
332        } else if self.replica_pool.is_some() {
333            None
334        } else {
335            self.pool.as_ref()
336        }
337    }
338
339    /// Returns the metrics collector.
340    #[must_use]
341    pub const fn metrics(&self) -> &middleware::MetricsCollector {
342        &self.metrics
343    }
344
345    /// Returns the log levels configuration.
346    #[must_use]
347    pub const fn log_levels(&self) -> &actuator::LogLevels {
348        &self.log_levels
349    }
350
351    /// Returns the task registry.
352    #[must_use]
353    pub const fn task_registry(&self) -> &actuator::TaskRegistry {
354        &self.task_registry
355    }
356
357    /// Returns the job registry.
358    #[must_use]
359    pub const fn job_registry(&self) -> &actuator::JobRegistry {
360        &self.job_registry
361    }
362
363    /// Returns the config properties.
364    #[must_use]
365    pub const fn config_props(&self) -> &actuator::ConfigProperties {
366        &self.config_props
367    }
368
369    /// Returns the registry of plugin-contributed metrics sources.
370    #[must_use]
371    pub const fn metrics_source_registry(&self) -> &actuator::MetricsSourceRegistry {
372        &self.metrics_source_registry
373    }
374
375    /// Returns the registry of custom health indicators.
376    #[must_use]
377    pub const fn health_indicator_registry(&self) -> &actuator::HealthIndicatorRegistry {
378        &self.health_indicator_registry
379    }
380
381    /// Returns the resolved [`crate::config::AutumnConfig`] from the extension map.
382    ///
383    /// Falls back to a default config if no config has been installed
384    /// (typically only in tests that don't wire the full startup pipeline).
385    #[must_use]
386    pub fn config(&self) -> crate::config::AutumnConfig {
387        self.extension::<crate::config::AutumnConfig>()
388            .map_or_else(crate::config::AutumnConfig::default, |arc| (*arc).clone())
389    }
390
391    /// Allocate the next process-unique app id.
392    ///
393    /// Called exactly once per genuine `AppState` construction; clones copy the
394    /// resulting `u64` verbatim, so a cloned state (what `State<AppState>` hands
395    /// a handler) reports the same id as its origin while a separately built
396    /// state gets a fresh one.
397    pub(crate) fn next_app_id() -> u64 {
398        NEXT_APP_ID.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
399    }
400
401    /// Process-unique identity of this app, stable across clones.
402    ///
403    /// Used to scope per-route `#[throttle]` limiters so two independently
404    /// built apps with identical config never share a token bucket.
405    #[must_use]
406    pub(crate) const fn app_id(&self) -> u64 {
407        self.app_id
408    }
409
410    /// Returns the shared probe lifecycle state.
411    #[must_use]
412    pub const fn probes(&self) -> &probe::ProbeState {
413        &self.probes
414    }
415
416    /// Mark startup as complete so readiness can become healthy.
417    pub fn mark_startup_complete(&self) {
418        self.probes.mark_startup_complete();
419    }
420
421    /// Mark the application as draining so readiness flips unhealthy.
422    pub fn begin_shutdown(&self) {
423        self.probes.begin_shutdown();
424    }
425
426    /// Sets the database pool.
427    #[cfg(feature = "db")]
428    #[must_use]
429    pub fn with_pool(
430        mut self,
431        pool: diesel_async::pooled_connection::deadpool::Pool<crate::db::RuntimeConnection>,
432    ) -> Self {
433        self.pool = Some(pool);
434        self
435    }
436
437    /// Sets the read-replica database pool.
438    #[cfg(feature = "db")]
439    #[must_use]
440    pub fn with_replica_pool(
441        mut self,
442        pool: diesel_async::pooled_connection::deadpool::Pool<crate::db::RuntimeConnection>,
443    ) -> Self {
444        self.replica_pool = Some(pool);
445        self
446    }
447
448    /// Sets the shard set.
449    #[cfg(feature = "db")]
450    #[must_use]
451    pub fn with_shards(mut self, shards: crate::sharding::ShardSet) -> Self {
452        self.shards = Some(shards);
453        self
454    }
455
456    /// Install a typed runtime extension while building test or ad-hoc state.
457    #[must_use]
458    pub fn with_extension<T>(self, value: T) -> Self
459    where
460        T: Any + Send + Sync + 'static,
461    {
462        self.insert_extension(value);
463        self
464    }
465
466    /// Returns the registered global cache backend, if any.
467    ///
468    /// Checks the extension map first (populated at runtime by startup hooks
469    /// via [`Self::set_cache`]) so that a plugin replacing a build-time backend
470    /// is always visible. Falls back to `shared_cache` (set at build time via
471    /// [`Self::with_cache`]).
472    #[must_use]
473    pub fn cache(&self) -> Option<Arc<dyn Cache>> {
474        self.extension::<GlobalCacheEntry>()
475            .map(|e| e.0.clone())
476            .or_else(|| self.shared_cache.clone())
477    }
478
479    /// Register a global cache backend (builder / test helper, build-time).
480    #[must_use]
481    pub fn with_cache(mut self, cache: Arc<dyn Cache>) -> Self {
482        self.shared_cache = Some(cache);
483        self
484    }
485
486    /// Returns the active clock source wired into this state.
487    ///
488    /// Handlers should prefer the [`crate::time::Clock`] extractor; this
489    /// accessor exists for framework internals (middleware, storage) that
490    /// need the time without going through Axum's extractor machinery.
491    #[must_use]
492    pub fn clock(&self) -> &dyn ClockSource {
493        self.clock.as_ref()
494    }
495
496    /// Replace the clock (builder / test helper).
497    #[must_use]
498    pub fn with_clock(mut self, clock: Arc<dyn ClockSource>) -> Self {
499        self.clock = clock;
500        self
501    }
502
503    /// Install or replace the global cache backend at runtime (e.g. from a startup hook).
504    ///
505    /// Updates both the process-level global (used by `#[cached]` functions) and
506    /// the extension map (used by `CacheResponseLayer::from_app` and `state.cache()`).
507    pub fn set_cache(&self, cache: Arc<dyn Cache>) {
508        crate::cache::set_global_cache(cache.clone());
509        self.insert_extension(GlobalCacheEntry(cache));
510    }
511
512    /// Sets the active profile.
513    #[must_use]
514    pub fn with_profile(mut self, profile: impl Into<String>) -> Self {
515        self.profile = Some(profile.into());
516        self
517    }
518
519    /// Returns a reference to the [`PolicyRegistry`].
520    #[must_use]
521    pub const fn policy_registry(&self) -> &PolicyRegistry {
522        &self.policy_registry
523    }
524
525    /// Resolve the registered [`Policy`] for resource `R`, if any.
526    #[must_use]
527    pub fn policy<R: Send + Sync + 'static>(&self) -> Option<std::sync::Arc<dyn Policy<R>>> {
528        self.policy_registry.policy::<R>()
529    }
530
531    /// Resolve the registered [`Scope`] for resource `R`, if any.
532    #[must_use]
533    pub fn scope<R: Send + Sync + 'static>(&self) -> Option<std::sync::Arc<dyn Scope<R>>> {
534        self.policy_registry.scope::<R>()
535    }
536
537    /// Configured deny-response shape. See
538    /// [`ForbiddenResponse`] for the trade-off between `403` and
539    /// `404` defaults.
540    #[must_use]
541    pub const fn forbidden_response(&self) -> ForbiddenResponse {
542        self.forbidden_response
543    }
544
545    /// Session key used to resolve the authenticated user id for
546    /// [`PolicyContext`](crate::authorization::PolicyContext).
547    #[must_use]
548    pub fn auth_session_key(&self) -> &str {
549        &self.auth_session_key
550    }
551
552    /// Override the configured deny response (test helper).
553    #[doc(hidden)]
554    #[must_use]
555    pub const fn with_forbidden_response(mut self, value: ForbiddenResponse) -> Self {
556        self.forbidden_response = value;
557        self
558    }
559
560    /// Override the auth session key (test helper).
561    #[doc(hidden)]
562    #[must_use]
563    pub fn with_auth_session_key(mut self, value: impl Into<String>) -> Self {
564        self.auth_session_key = value.into();
565        self
566    }
567
568    /// Set the startup probe completion flag.
569    #[doc(hidden)]
570    #[must_use]
571    pub fn with_startup_complete(self, startup_complete: bool) -> Self {
572        self.probes.set_startup_complete(startup_complete);
573        self
574    }
575
576    /// Set the readiness draining flag.
577    #[doc(hidden)]
578    #[must_use]
579    pub fn with_draining(self, draining: bool) -> Self {
580        self.probes.set_draining(draining);
581        self
582    }
583
584    /// Returns the active profile name, or `"default"` if none is set.
585    #[must_use]
586    pub fn profile(&self) -> &str {
587        self.profile.as_deref().unwrap_or("default")
588    }
589
590    /// Returns the resolved [`ProcessRole`](crate::config::ProcessRole) for this
591    /// replica.
592    ///
593    /// This is the role after config parsing and the `AUTUMN_ROLE` env override
594    /// — the exact same value the framework uses to gate the job runtime,
595    /// scheduler, and commit-hook worker. Use it from `state_initializer`,
596    /// `on_startup`/`on_shutdown` hooks, plugins, and request handlers to
597    /// self-gate app-owned background work:
598    ///
599    /// ```rust
600    /// # use autumn_web::AppState;
601    /// # fn example(state: &AppState) {
602    /// if state.role().runs_workers() {
603    ///     // start an embedded worker loop only on replicas that run workers
604    /// }
605    /// # }
606    /// ```
607    ///
608    /// [`serves_http`](crate::config::ProcessRole::serves_http) and
609    /// [`runs_workers`](crate::config::ProcessRole::runs_workers) are reachable
610    /// on the returned value.
611    #[must_use]
612    pub const fn role(&self) -> crate::config::ProcessRole {
613        self.role
614    }
615
616    /// Returns how long the application has been running.
617    #[must_use]
618    pub fn uptime(&self) -> std::time::Duration {
619        self.started_at.elapsed()
620    }
621
622    /// Format uptime as a human-readable string (e.g., "2h 15m").
623    #[must_use]
624    pub fn uptime_display(&self) -> String {
625        let secs = self.started_at.elapsed().as_secs();
626        if secs < 60 {
627            format!("{secs}s")
628        } else if secs < 3600 {
629            format!("{}m {}s", secs / 60, secs % 60)
630        } else {
631            let hours = secs / 3600;
632            let mins = (secs % 3600) / 60;
633            format!("{hours}h {mins}m")
634        }
635    }
636
637    /// Returns a reference to the broadcast channel registry.
638    ///
639    /// Shorthand for accessing `self.channels` directly.
640    #[cfg(feature = "ws")]
641    #[must_use]
642    pub const fn channels(&self) -> &Channels {
643        &self.channels
644    }
645
646    /// Returns a reference to the distributed presence tracker.
647    #[cfg(feature = "presence")]
648    #[must_use]
649    pub const fn presence(&self) -> &Presence {
650        &self.presence
651    }
652
653    /// Returns a high-level broadcast facade for raw and htmx HTML payloads.
654    #[cfg(feature = "ws")]
655    #[must_use]
656    pub fn broadcast(&self) -> crate::channels::Broadcast {
657        self.channels.broadcast()
658    }
659
660    /// Returns a child cancellation token for the server shutdown signal.
661    ///
662    /// WebSocket handlers should select on this to clean up when the
663    /// server is shutting down.
664    #[cfg(feature = "ws")]
665    #[must_use]
666    pub fn shutdown_token(&self) -> CancellationToken {
667        self.shutdown.child_token()
668    }
669
670    /// Helper for integration tests to simulate a server shutdown.
671    #[cfg(feature = "ws")]
672    #[doc(hidden)]
673    pub fn trigger_shutdown_for_test(&self) {
674        self.begin_shutdown();
675        self.shutdown.cancel();
676    }
677
678    /// Update startup completion in tests after the router is already built.
679    #[doc(hidden)]
680    pub fn set_startup_complete_for_test(&self, startup_complete: bool) {
681        self.probes.set_startup_complete(startup_complete);
682    }
683
684    /// Update draining state in tests after the router is already built.
685    #[doc(hidden)]
686    pub fn set_draining_for_test(&self, draining: bool) {
687        self.probes.set_draining(draining);
688    }
689
690    /// Compatibility helper for tests that model shutdown as readiness drain.
691    #[doc(hidden)]
692    pub fn begin_shutdown_for_test(&self) {
693        self.set_draining_for_test(true);
694    }
695
696    /// Create a minimal detached `AppState` without an HTTP server.
697    ///
698    /// This is useful for background runtimes or helper processes that still
699    /// need framework-managed resources such as typed extensions, metrics, or
700    /// WebSocket channel registries.
701    #[must_use]
702    pub fn detached() -> Self {
703        #[cfg(feature = "ws")]
704        let channels = Channels::new(32);
705        Self {
706            extensions: Arc::new(std::sync::RwLock::new(HashMap::new())),
707            #[cfg(feature = "db")]
708            pool: None,
709            #[cfg(feature = "db")]
710            replica_pool: None,
711            #[cfg(feature = "db")]
712            shards: None,
713            profile: None,
714            role: crate::config::ProcessRole::Combined,
715            started_at: std::time::Instant::now(),
716            health_detailed: true,
717            probes: probe::ProbeState::ready_for_test(),
718            metrics: middleware::MetricsCollector::new(),
719            log_levels: actuator::LogLevels::new("info"),
720            task_registry: actuator::TaskRegistry::new(),
721            job_registry: actuator::JobRegistry::new(),
722            config_props: actuator::ConfigProperties::default(),
723            metrics_source_registry: actuator::MetricsSourceRegistry::new(),
724            health_indicator_registry: actuator::HealthIndicatorRegistry::new(),
725            #[cfg(feature = "presence")]
726            presence: Presence::new(channels.clone()),
727            #[cfg(feature = "ws")]
728            channels,
729            #[cfg(feature = "ws")]
730            shutdown: CancellationToken::new(),
731            policy_registry: PolicyRegistry::default(),
732            forbidden_response: ForbiddenResponse::default(),
733            auth_session_key: "user_id".to_owned(),
734            shared_cache: None,
735            clock: Arc::new(SystemClock),
736            app_id: Self::next_app_id(),
737        }
738    }
739
740    /// Create an `AppState` suitable for testing, with sensible defaults
741    /// for all fields. Database pool is `None`.
742    #[allow(dead_code)]
743    #[must_use]
744    pub fn for_test() -> Self {
745        Self::detached()
746    }
747}
748
749#[cfg(feature = "db")]
750impl DbState for AppState {
751    fn metrics(&self) -> Option<&crate::middleware::MetricsCollector> {
752        Some(&self.metrics)
753    }
754
755    fn pool(
756        &self,
757    ) -> Option<&diesel_async::pooled_connection::deadpool::Pool<crate::db::RuntimeConnection>>
758    {
759        self.pool.as_ref()
760    }
761
762    fn replica_pool(
763        &self,
764    ) -> Option<&diesel_async::pooled_connection::deadpool::Pool<crate::db::RuntimeConnection>>
765    {
766        self.replica_pool.as_ref()
767    }
768
769    fn read_pool(
770        &self,
771    ) -> Option<&diesel_async::pooled_connection::deadpool::Pool<crate::db::RuntimeConnection>>
772    {
773        Self::read_pool(self)
774    }
775
776    fn shards(&self) -> Option<&crate::sharding::ShardSet> {
777        self.shards.as_ref()
778    }
779
780    fn db_interceptors(
781        &self,
782    ) -> Vec<std::sync::Arc<dyn crate::interceptor::DbConnectionInterceptor>> {
783        self.extension::<Arc<dyn crate::interceptor::DbConnectionInterceptor>>()
784            .map(|arc| vec![(*arc).clone()])
785            .unwrap_or_default()
786    }
787    fn statement_timeout(&self) -> Option<std::time::Duration> {
788        self.extension::<crate::config::AutumnConfig>()
789            .and_then(|cfg| cfg.database.statement_timeout)
790    }
791
792    fn slow_query_threshold(&self) -> std::time::Duration {
793        self.extension::<crate::config::AutumnConfig>().map_or_else(
794            || std::time::Duration::from_millis(500),
795            |cfg| cfg.database.slow_query_threshold,
796        )
797    }
798}
799
800impl crate::probe::ProvideProbeState for AppState {
801    fn probes(&self) -> &crate::probe::ProbeState {
802        &self.probes
803    }
804
805    fn health_detailed(&self) -> bool {
806        self.health_detailed
807    }
808
809    fn profile(&self) -> &str {
810        self.profile()
811    }
812
813    fn uptime_display(&self) -> String {
814        self.uptime_display()
815    }
816
817    #[cfg(feature = "db")]
818    fn pool(
819        &self,
820    ) -> Option<&diesel_async::pooled_connection::deadpool::Pool<crate::db::RuntimeConnection>>
821    {
822        self.pool.as_ref()
823    }
824
825    #[cfg(feature = "db")]
826    fn replica_pool(
827        &self,
828    ) -> Option<&diesel_async::pooled_connection::deadpool::Pool<crate::db::RuntimeConnection>>
829    {
830        self.replica_pool.as_ref()
831    }
832
833    fn health_indicator_registry(&self) -> Option<&crate::actuator::HealthIndicatorRegistry> {
834        Some(&self.health_indicator_registry)
835    }
836}
837
838impl crate::actuator::ProvideActuatorState for AppState {
839    fn metrics(&self) -> &crate::middleware::MetricsCollector {
840        &self.metrics
841    }
842
843    fn log_levels(&self) -> &crate::actuator::LogLevels {
844        &self.log_levels
845    }
846
847    fn task_registry(&self) -> &crate::actuator::TaskRegistry {
848        &self.task_registry
849    }
850
851    fn job_registry(&self) -> &crate::actuator::JobRegistry {
852        &self.job_registry
853    }
854
855    fn config_props(&self) -> &crate::actuator::ConfigProperties {
856        &self.config_props
857    }
858
859    fn profile(&self) -> &str {
860        self.profile()
861    }
862
863    fn uptime_display(&self) -> String {
864        self.uptime_display()
865    }
866
867    fn metrics_source_registry(&self) -> Option<&crate::actuator::MetricsSourceRegistry> {
868        Some(&self.metrics_source_registry)
869    }
870
871    fn health_indicator_registry(&self) -> Option<&crate::actuator::HealthIndicatorRegistry> {
872        Some(&self.health_indicator_registry)
873    }
874
875    fn health_detailed(&self) -> bool {
876        self.health_detailed
877    }
878
879    fn deploy_version(&self) -> String {
880        self.extension::<crate::canary::CanaryState>().map_or_else(
881            || crate::canary::STABLE.to_owned(),
882            |c| c.version().to_owned(),
883        )
884    }
885
886    #[cfg(feature = "ws")]
887    fn channels(&self) -> &crate::channels::Channels {
888        &self.channels
889    }
890
891    #[cfg(feature = "ws")]
892    fn shutdown_token(&self) -> tokio_util::sync::CancellationToken {
893        self.shutdown_token()
894    }
895
896    #[cfg(feature = "db")]
897    fn pool(
898        &self,
899    ) -> Option<&diesel_async::pooled_connection::deadpool::Pool<crate::db::RuntimeConnection>>
900    {
901        self.pool.as_ref()
902    }
903
904    #[cfg(feature = "db")]
905    fn shards(&self) -> Option<&crate::sharding::ShardSet> {
906        self.shards.as_ref()
907    }
908    // a11y_posture() uses the trait default (all-false) intentionally: AppState
909    // cannot know whether the application's layout is accessible.  Override this
910    // method on your own state type — or in a custom ProvideActuatorState impl —
911    // once you have verified that your pages include lang, a skip link, and
912    // landmark regions.  See docs/guide/accessibility.md for details.
913
914    #[cfg(feature = "http-client")]
915    fn webhook_outbound(&self) -> Option<crate::webhook_outbound::WebhookOutboundManager> {
916        self.extension::<crate::webhook_outbound::WebhookOutboundManager>()
917            .map(|x| (*x).clone())
918    }
919
920    fn log_buffer(&self) -> Option<crate::log::capture::LogBuffer> {
921        self.extension::<crate::log::capture::LogBuffer>()
922            .map(|x| (*x).clone())
923    }
924}
925
926impl std::fmt::Debug for AppState {
927    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
928        let mut s = f.debug_struct("AppState");
929        #[cfg(feature = "db")]
930        s.field(
931            "pool",
932            &self
933                .pool
934                .as_ref()
935                .map(|p| format!("Pool(max={})", p.status().max_size)),
936        );
937        s.field(
938            "extensions",
939            &self
940                .extensions
941                .read()
942                .map_or(0, |extensions| extensions.len()),
943        );
944        s.field("profile", &self.profile)
945            .field("started_at", &self.started_at)
946            .field("health_detailed", &self.health_detailed)
947            .field("probes", &self.probes)
948            .field("metrics", &"MetricsCollector")
949            .field("log_levels", &"LogLevels")
950            .field("task_registry", &"TaskRegistry")
951            .finish_non_exhaustive()
952    }
953}
954
955#[cfg(test)]
956mod tests {
957    use super::*;
958    #[cfg(feature = "db")]
959    use crate::config;
960    #[cfg(feature = "db")]
961    use crate::db;
962
963    #[test]
964    fn app_state_debug_without_pool() {
965        let state = AppState::for_test().with_profile("dev");
966        let debug = format!("{state:?}");
967        assert!(debug.contains("AppState"));
968        assert!(debug.contains("dev"));
969    }
970
971    #[cfg(feature = "db")]
972    #[test]
973    fn app_state_debug_with_pool() {
974        let config = config::DatabaseConfig {
975            url: Some("postgres://localhost/test".into()),
976            pool_size: 5,
977            ..Default::default()
978        };
979        let pool = db::create_pool(&config).unwrap().unwrap();
980        let state = AppState::for_test().with_pool(pool);
981        let debug = format!("{state:?}");
982        assert!(debug.contains("Pool(max=5)"));
983    }
984
985    #[cfg(feature = "db")]
986    #[test]
987    fn database_topology_state_exposes_replica_as_read_pool() {
988        let primary_config = config::DatabaseConfig {
989            url: Some("postgres://localhost/primary".into()),
990            pool_size: 5,
991            ..Default::default()
992        };
993        let replica_config = config::DatabaseConfig {
994            url: Some("postgres://localhost/replica".into()),
995            pool_size: 2,
996            ..Default::default()
997        };
998        let primary = db::create_pool(&primary_config).unwrap().unwrap();
999        let replica = db::create_pool(&replica_config).unwrap().unwrap();
1000
1001        let state = AppState::for_test()
1002            .with_pool(primary)
1003            .with_replica_pool(replica);
1004
1005        assert_eq!(state.pool().expect("primary pool").status().max_size, 5);
1006        assert_eq!(
1007            state
1008                .replica_pool()
1009                .expect("replica pool")
1010                .status()
1011                .max_size,
1012            2
1013        );
1014        assert_eq!(state.read_pool().expect("read pool").status().max_size, 2);
1015    }
1016
1017    #[cfg(feature = "db")]
1018    #[test]
1019    fn read_pool_uses_primary_when_replica_is_unready_and_policy_allows_fallback() {
1020        let primary_config = config::DatabaseConfig {
1021            url: Some("postgres://localhost/primary".into()),
1022            pool_size: 5,
1023            ..Default::default()
1024        };
1025        let replica_config = config::DatabaseConfig {
1026            url: Some("postgres://localhost/replica".into()),
1027            pool_size: 2,
1028            ..Default::default()
1029        };
1030        let primary = db::create_pool(&primary_config).unwrap().unwrap();
1031        let replica = db::create_pool(&replica_config).unwrap().unwrap();
1032
1033        let state = AppState::for_test()
1034            .with_pool(primary)
1035            .with_replica_pool(replica);
1036        state
1037            .probes()
1038            .configure_replica_dependency(config::ReplicaFallback::Primary);
1039        state
1040            .probes()
1041            .mark_replica_unready("replica migrations lag primary");
1042
1043        assert_eq!(state.read_pool().expect("read pool").status().max_size, 5);
1044        assert_eq!(
1045            db::DbState::read_pool(&state)
1046                .expect("trait read pool")
1047                .status()
1048                .max_size,
1049            5
1050        );
1051    }
1052
1053    #[cfg(feature = "db")]
1054    #[test]
1055    fn read_pool_does_not_route_to_unready_replica_when_policy_fails_readiness() {
1056        let primary_config = config::DatabaseConfig {
1057            url: Some("postgres://localhost/primary".into()),
1058            pool_size: 5,
1059            ..Default::default()
1060        };
1061        let replica_config = config::DatabaseConfig {
1062            url: Some("postgres://localhost/replica".into()),
1063            pool_size: 2,
1064            ..Default::default()
1065        };
1066        let primary = db::create_pool(&primary_config).unwrap().unwrap();
1067        let replica = db::create_pool(&replica_config).unwrap().unwrap();
1068
1069        let state = AppState::for_test()
1070            .with_pool(primary)
1071            .with_replica_pool(replica);
1072        state
1073            .probes()
1074            .configure_replica_dependency(config::ReplicaFallback::FailReadiness);
1075        state
1076            .probes()
1077            .mark_replica_unready("replica connection failed");
1078
1079        assert!(state.read_pool().is_none());
1080    }
1081
1082    #[cfg(feature = "db")]
1083    #[tokio::test]
1084    async fn readiness_fails_when_app_state_replica_is_unready_and_policy_is_fail_readiness() {
1085        let primary_config = config::DatabaseConfig {
1086            url: Some("postgres://localhost/primary".into()),
1087            pool_size: 5,
1088            ..Default::default()
1089        };
1090        let replica_config = config::DatabaseConfig {
1091            url: Some("postgres://localhost/replica".into()),
1092            pool_size: 2,
1093            ..Default::default()
1094        };
1095        let primary = db::create_pool(&primary_config).unwrap().unwrap();
1096        let replica = db::create_pool(&replica_config).unwrap().unwrap();
1097
1098        let state = AppState::for_test()
1099            .with_pool(primary)
1100            .with_replica_pool(replica);
1101        state
1102            .probes()
1103            .configure_replica_dependency(config::ReplicaFallback::FailReadiness);
1104        state
1105            .probes()
1106            .mark_replica_unready("replica migrations lag primary");
1107
1108        let (status, _) = crate::probe::readiness_response(&state).await;
1109
1110        assert_eq!(status, http::StatusCode::SERVICE_UNAVAILABLE);
1111    }
1112
1113    #[test]
1114    fn detached_state_starts_without_profile() {
1115        let state = AppState::detached();
1116
1117        assert_eq!(state.profile(), "default");
1118    }
1119
1120    fn require_clone<T: Clone>(t: &T) -> T {
1121        t.clone()
1122    }
1123
1124    #[test]
1125    fn app_state_is_clone() {
1126        let state = AppState::for_test();
1127        let _cloned = require_clone(&state);
1128    }
1129
1130    #[test]
1131    fn app_state_profile_accessor() {
1132        let state = AppState::for_test().with_profile("staging");
1133        assert_eq!(state.profile(), "staging");
1134    }
1135
1136    #[test]
1137    fn app_state_deploy_version_defaults_to_stable() {
1138        use crate::actuator::ProvideActuatorState;
1139        let state = AppState::for_test();
1140        assert_eq!(state.deploy_version(), crate::canary::STABLE);
1141    }
1142
1143    #[test]
1144    fn app_state_deploy_version_reads_canary_extension() {
1145        use crate::actuator::ProvideActuatorState;
1146        let state = AppState::for_test();
1147        state.insert_extension(crate::canary::CanaryState::new(crate::canary::CANARY));
1148        assert_eq!(state.deploy_version(), crate::canary::CANARY);
1149    }
1150
1151    #[test]
1152    fn app_state_profile_default() {
1153        let state = AppState::for_test();
1154        assert_eq!(state.profile(), "default");
1155    }
1156
1157    #[test]
1158    fn app_state_uptime_display() {
1159        let state = AppState::for_test();
1160        let display = state.uptime_display();
1161        assert!(
1162            display.contains('s'),
1163            "uptime should contain 's': {display}"
1164        );
1165    }
1166
1167    #[test]
1168    fn app_state_accessors() {
1169        let state = AppState::for_test();
1170
1171        // Exercise the new getters to ensure they compile and return the expected types
1172        let _metrics = state.metrics();
1173        let _log_levels = state.log_levels();
1174        let _task_registry = state.task_registry();
1175        let _config_props = state.config_props();
1176
1177        #[cfg(feature = "db")]
1178        {
1179            let _pool = state.pool();
1180        }
1181        let _missing = state.extension::<String>();
1182    }
1183
1184    #[test]
1185    fn app_state_runtime_extensions_round_trip() {
1186        let state = AppState::for_test();
1187        state.insert_extension(String::from("haunted"));
1188
1189        let stored = state
1190            .extension::<String>()
1191            .expect("runtime extension should be installed");
1192
1193        assert_eq!(stored.as_str(), "haunted");
1194    }
1195}
autumn_web/state.rs

autumn_web/
state.rs
