1#![cfg_attr(not(test), deny(clippy::disallowed_methods))]
19
20use std::any::{Any, TypeId};
21use std::collections::HashMap;
22use std::sync::Arc;
23
24use crate::cache::Cache;
25use crate::time::{ClockSource, MonotonicInstant, SystemClock};
26
27pub struct GlobalCacheEntry(pub Arc<dyn Cache>);
33
34use crate::actuator;
35use crate::authorization::{ForbiddenResponse, Policy, PolicyRegistry, Scope};
36#[cfg(feature = "ws")]
37use crate::channels::Channels;
38#[cfg(all(feature = "collab", feature = "presence"))]
39use crate::collab::CollabHub;
40#[cfg(feature = "db")]
41use crate::db::DbState;
42use crate::middleware;
43#[cfg(feature = "presence")]
44use crate::presence::Presence;
45use crate::probe;
46use tokio_util::sync::CancellationToken;
47
48#[derive(Clone)]
61pub struct LateExtensions(Arc<std::sync::RwLock<HashMap<TypeId, Arc<dyn Any + Send + Sync>>>>);
62
63impl LateExtensions {
64 #[must_use]
72 pub fn get<T>(&self) -> Option<Arc<T>>
73 where
74 T: Any + Send + Sync + 'static,
75 {
76 self.0
77 .read()
78 .expect("app state extension lock poisoned")
79 .get(&TypeId::of::<T>())
80 .cloned()
81 .and_then(|value| Arc::downcast::<T>(value).ok())
82 }
83}
84
85impl std::fmt::Debug for LateExtensions {
86 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
87 f.write_str("LateExtensions")
89 }
90}
91
92#[derive(Clone)]
113#[non_exhaustive]
114pub struct AppState {
115 pub(crate) extensions: Arc<std::sync::RwLock<HashMap<TypeId, Arc<dyn Any + Send + Sync>>>>,
118
119 #[cfg(feature = "db")]
122 pub(crate) pool:
123 Option<diesel_async::pooled_connection::deadpool::Pool<crate::db::RuntimeConnection>>,
124
125 #[cfg(feature = "db")]
127 pub(crate) replica_pool:
128 Option<diesel_async::pooled_connection::deadpool::Pool<crate::db::RuntimeConnection>>,
129
130 #[cfg(feature = "db")]
134 pub(crate) shards: Option<crate::sharding::ShardSet>,
135
136 #[cfg(all(feature = "db", feature = "reporting"))]
141 pub(crate) db_capture_gap: Option<Arc<str>>,
142
143 pub(crate) profile: Option<Arc<str>>,
150
151 pub(crate) role: crate::config::ProcessRole,
158
159 pub(crate) started_at: MonotonicInstant,
168
169 pub(crate) health_detailed: bool,
171
172 pub(crate) probes: probe::ProbeState,
174
175 pub(crate) metrics: middleware::MetricsCollector,
177
178 pub(crate) log_levels: actuator::LogLevels,
180
181 pub(crate) task_registry: actuator::TaskRegistry,
183 pub(crate) job_registry: actuator::JobRegistry,
185
186 pub(crate) config_props: actuator::ConfigProperties,
188
189 pub(crate) metrics_source_registry: actuator::MetricsSourceRegistry,
192
193 pub(crate) health_indicator_registry: actuator::HealthIndicatorRegistry,
196
197 #[cfg(feature = "ws")]
202 pub(crate) channels: Channels,
203
204 #[cfg(feature = "presence")]
209 pub(crate) presence: Presence,
210
211 #[cfg(all(feature = "collab", feature = "presence"))]
216 pub(crate) collab: CollabHub,
217
218 pub(crate) policy_registry: PolicyRegistry,
221
222 pub(crate) forbidden_response: ForbiddenResponse,
226
227 pub(crate) auth_session_key: Arc<str>,
235
236 pub(crate) shared_cache: Option<Arc<dyn Cache>>,
239
240 pub(crate) clock: Arc<dyn ClockSource>,
243
244 pub(crate) entropy: Arc<dyn crate::entropy::Entropy>,
250
251 pub(crate) app_id: u64,
261}
262
263static NEXT_APP_ID: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1);
266
267static DEFAULT_CONFIG: std::sync::OnceLock<Arc<crate::config::AutumnConfig>> =
278 std::sync::OnceLock::new();
279
280fn default_config() -> &'static Arc<crate::config::AutumnConfig> {
282 DEFAULT_CONFIG.get_or_init(|| Arc::new(crate::config::AutumnConfig::default()))
283}
284
285impl crate::authorization::ProvideAuthorizationState for AppState {
286 fn policy_registry(&self) -> &crate::authorization::PolicyRegistry {
287 &self.policy_registry
288 }
289
290 fn auth_session_key(&self) -> &str {
291 &self.auth_session_key
292 }
293
294 fn forbidden_response(&self) -> &crate::authorization::ForbiddenResponse {
295 &self.forbidden_response
296 }
297
298 #[cfg(feature = "db")]
299 fn pool(
300 &self,
301 ) -> Option<&diesel_async::pooled_connection::deadpool::Pool<crate::db::RuntimeConnection>>
302 {
303 self.pool.as_ref()
304 }
305}
306
307impl AppState {
308 pub fn insert_extension<T>(&self, value: T)
317 where
318 T: Any + Send + Sync + 'static,
319 {
320 self.extensions
321 .write()
322 .expect("app state extension lock poisoned")
323 .insert(TypeId::of::<T>(), Arc::new(value));
324 }
325
326 #[must_use]
335 pub fn session_rotation_hooks(&self) -> crate::session::SessionRotationHooks {
336 let entry = {
337 let mut extensions = self
338 .extensions
339 .write()
340 .expect("app state extension lock poisoned");
341 Arc::clone(
342 extensions
343 .entry(TypeId::of::<crate::session::SessionRotationHooks>())
344 .or_insert_with(|| Arc::new(crate::session::SessionRotationHooks::default())),
345 )
346 };
347 entry
348 .downcast::<crate::session::SessionRotationHooks>()
349 .map(|hooks| (*hooks).clone())
350 .expect("rotation hooks extension has the wrong type")
351 }
352
353 #[must_use]
362 pub fn extension<T>(&self) -> Option<Arc<T>>
363 where
364 T: Any + Send + Sync + 'static,
365 {
366 self.extensions
367 .read()
368 .expect("app state extension lock poisoned")
369 .get(&TypeId::of::<T>())
370 .cloned()
371 .and_then(|value| Arc::downcast::<T>(value).ok())
372 }
373
374 #[must_use]
378 pub fn tenant_cell_registry(&self) -> Option<Arc<crate::tenant_cell::TenantCellRegistry>> {
379 self.extension::<crate::tenant_cell::TenantCellRegistry>()
380 }
381
382 #[must_use]
387 pub(crate) fn late_extensions(&self) -> LateExtensions {
388 LateExtensions(Arc::clone(&self.extensions))
389 }
390
391 #[must_use]
413 pub fn live_state<T>(&self) -> Option<crate::upgrade::LiveStateHandle<T>>
414 where
415 T: crate::upgrade::LiveState,
416 {
417 self.extension::<crate::upgrade::LiveStateHandle<T>>()
420 .map(|handle| (*handle).clone())
421 }
422
423 pub fn extension_or_insert_with<T>(&self, f: impl FnOnce() -> T) -> Arc<T>
431 where
432 T: Any + Send + Sync + 'static,
433 {
434 if let Some(existing) = self.extension::<T>() {
435 return existing;
436 }
437 let mut map = self
438 .extensions
439 .write()
440 .expect("app state extension lock poisoned");
441 if let Some(existing) = map
442 .get(&TypeId::of::<T>())
443 .cloned()
444 .and_then(|value| Arc::downcast::<T>(value).ok())
445 {
446 return existing;
447 }
448 let arc = Arc::new(f());
449 map.insert(TypeId::of::<T>(), arc.clone() as Arc<dyn Any + Send + Sync>);
450 arc
451 }
452
453 #[cfg(feature = "reporting")]
460 #[must_use]
461 pub(crate) fn error_reporters(
462 &self,
463 ) -> Vec<std::sync::Arc<dyn crate::reporting::ErrorReporter>> {
464 self.extension::<crate::reporting::RegisteredReporters>()
465 .map(|reporters| reporters.0.clone())
466 .unwrap_or_default()
467 }
468
469 #[cfg(feature = "db")]
471 #[must_use]
472 pub const fn pool(
473 &self,
474 ) -> Option<&diesel_async::pooled_connection::deadpool::Pool<crate::db::RuntimeConnection>>
475 {
476 self.pool.as_ref()
477 }
478
479 #[cfg(feature = "db")]
481 #[must_use]
482 pub const fn replica_pool(
483 &self,
484 ) -> Option<&diesel_async::pooled_connection::deadpool::Pool<crate::db::RuntimeConnection>>
485 {
486 self.replica_pool.as_ref()
487 }
488
489 #[cfg(feature = "db")]
495 #[must_use]
496 pub const fn shards(&self) -> Option<&crate::sharding::ShardSet> {
497 self.shards.as_ref()
498 }
499
500 #[cfg(feature = "db")]
502 #[must_use]
503 pub fn read_pool(
504 &self,
505 ) -> Option<&diesel_async::pooled_connection::deadpool::Pool<crate::db::RuntimeConnection>>
506 {
507 if self.replica_pool.is_some() && self.probes.should_route_reads_to_replica() {
508 self.replica_pool.as_ref()
509 } else if self.replica_pool.is_some() && self.probes.should_fallback_reads_to_primary() {
510 self.pool.as_ref()
511 } else if self.replica_pool.is_some() {
512 None
513 } else {
514 self.pool.as_ref()
515 }
516 }
517
518 #[must_use]
520 pub const fn metrics(&self) -> &middleware::MetricsCollector {
521 &self.metrics
522 }
523
524 #[must_use]
526 pub const fn log_levels(&self) -> &actuator::LogLevels {
527 &self.log_levels
528 }
529
530 #[must_use]
532 pub const fn task_registry(&self) -> &actuator::TaskRegistry {
533 &self.task_registry
534 }
535
536 #[must_use]
538 pub const fn job_registry(&self) -> &actuator::JobRegistry {
539 &self.job_registry
540 }
541
542 #[must_use]
544 pub const fn config_props(&self) -> &actuator::ConfigProperties {
545 &self.config_props
546 }
547
548 #[must_use]
550 pub const fn metrics_source_registry(&self) -> &actuator::MetricsSourceRegistry {
551 &self.metrics_source_registry
552 }
553
554 #[must_use]
556 pub const fn health_indicator_registry(&self) -> &actuator::HealthIndicatorRegistry {
557 &self.health_indicator_registry
558 }
559
560 #[must_use]
574 pub fn config(&self) -> crate::config::AutumnConfig {
575 (*self.config_arc()).clone()
576 }
577
578 #[must_use]
597 pub fn config_arc(&self) -> Arc<crate::config::AutumnConfig> {
598 self.extension::<crate::config::AutumnConfig>()
599 .unwrap_or_else(|| Arc::clone(default_config()))
600 }
601
602 pub(crate) fn next_app_id() -> u64 {
609 NEXT_APP_ID.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
610 }
611
612 #[must_use]
617 pub(crate) const fn app_id(&self) -> u64 {
618 self.app_id
619 }
620
621 #[must_use]
623 pub const fn probes(&self) -> &probe::ProbeState {
624 &self.probes
625 }
626
627 pub(crate) fn apply_health_config(&self, health: &crate::config::HealthConfig) {
631 self.health_indicator_registry
632 .set_cache_ttl(health.cache_ttl());
633 #[cfg(feature = "db")]
634 {
635 self.probes
636 .configure_db_check(health.cache_ttl(), health.ping_timeout());
637 self.probes.set_db_readiness(health.db_readiness);
638 }
639 }
640
641 pub fn mark_startup_complete(&self) {
643 self.probes.mark_startup_complete();
644 }
645
646 pub fn begin_shutdown(&self) {
648 self.probes.begin_shutdown();
649 }
650
651 #[cfg(feature = "db")]
653 #[must_use]
654 pub fn with_pool(
655 mut self,
656 pool: diesel_async::pooled_connection::deadpool::Pool<crate::db::RuntimeConnection>,
657 ) -> Self {
658 self.pool = Some(pool);
659 self
660 }
661
662 #[cfg(feature = "db")]
664 #[must_use]
665 pub fn with_replica_pool(
666 mut self,
667 pool: diesel_async::pooled_connection::deadpool::Pool<crate::db::RuntimeConnection>,
668 ) -> Self {
669 self.replica_pool = Some(pool);
670 self
671 }
672
673 #[cfg(feature = "db")]
675 #[must_use]
676 pub fn with_shards(mut self, shards: crate::sharding::ShardSet) -> Self {
677 self.shards = Some(shards);
678 self
679 }
680
681 #[must_use]
683 pub fn with_extension<T>(self, value: T) -> Self
684 where
685 T: Any + Send + Sync + 'static,
686 {
687 self.insert_extension(value);
688 self
689 }
690
691 #[must_use]
698 pub fn cache(&self) -> Option<Arc<dyn Cache>> {
699 self.extension::<GlobalCacheEntry>()
700 .map(|e| e.0.clone())
701 .or_else(|| self.shared_cache.clone())
702 }
703
704 #[cfg(feature = "reporting")]
709 pub(crate) fn has_cache(&self) -> bool {
710 self.cache().is_some()
711 }
712
713 #[must_use]
715 pub fn with_cache(mut self, cache: Arc<dyn Cache>) -> Self {
716 self.shared_cache = Some(crate::cache::with_capsule_seam(cache));
717 self
718 }
719
720 #[must_use]
726 pub fn clock(&self) -> &dyn ClockSource {
727 self.clock.as_ref()
728 }
729
730 #[must_use]
760 pub fn with_clock(mut self, clock: Arc<dyn ClockSource>) -> Self {
761 let already_initialized = self.job_registry.is_initialized();
762 debug_assert!(
763 !already_initialized,
764 "AppState::with_clock is a construction-time builder: by the time job \
765 names are registered a runtime holds its own clone of the registry, \
766 and re-clocking the state can no longer reach it. Install the clock \
767 before starting the job runtime."
768 );
769 if already_initialized {
770 tracing::error!(
771 "AppState::with_clock called after the job registry was populated; the \
772 job runtime holds its own clone and keeps recording there, so this \
773 state's actuator gauges will stay empty. Install the clock before \
774 starting the job runtime."
775 );
776 }
777 self.started_at = clock.monotonic();
778 self.job_registry = actuator::JobRegistry::new().with_clock(Arc::clone(&clock));
793 self.clock = clock;
794 self
795 }
796
797 #[must_use]
807 pub fn monotonic(&self) -> MonotonicInstant {
808 self.clock.monotonic()
809 }
810
811 #[must_use]
816 pub(crate) fn clock_arc(&self) -> Arc<dyn ClockSource> {
817 Arc::clone(&self.clock)
818 }
819
820 #[must_use]
826 pub fn entropy(&self) -> &dyn crate::entropy::Entropy {
827 self.entropy.as_ref()
828 }
829
830 #[must_use]
833 pub(crate) fn entropy_arc(&self) -> Arc<dyn crate::entropy::Entropy> {
834 self.entropy.clone()
835 }
836
837 #[must_use]
843 pub fn with_entropy(mut self, entropy: Arc<dyn crate::entropy::Entropy>) -> Self {
844 self.entropy = entropy;
845 self
846 }
847
848 pub fn set_cache(&self, cache: Arc<dyn Cache>) {
853 let cache = crate::cache::with_capsule_seam(cache);
855 crate::cache::set_global_cache(cache.clone());
856 self.insert_extension(GlobalCacheEntry(cache));
857 }
858
859 #[must_use]
861 pub fn with_profile(mut self, profile: impl Into<String>) -> Self {
862 self.profile = Some(Arc::from(profile.into()));
863 self
864 }
865
866 #[must_use]
868 pub const fn policy_registry(&self) -> &PolicyRegistry {
869 &self.policy_registry
870 }
871
872 #[must_use]
874 pub fn policy<R: Send + Sync + 'static>(&self) -> Option<std::sync::Arc<dyn Policy<R>>> {
875 self.policy_registry.policy::<R>()
876 }
877
878 #[must_use]
880 pub fn scope<R: Send + Sync + 'static>(&self) -> Option<std::sync::Arc<dyn Scope<R>>> {
881 self.policy_registry.scope::<R>()
882 }
883
884 #[must_use]
888 pub const fn forbidden_response(&self) -> ForbiddenResponse {
889 self.forbidden_response
890 }
891
892 #[must_use]
895 pub fn auth_session_key(&self) -> &str {
896 &self.auth_session_key
897 }
898
899 #[doc(hidden)]
901 #[must_use]
902 pub const fn with_forbidden_response(mut self, value: ForbiddenResponse) -> Self {
903 self.forbidden_response = value;
904 self
905 }
906
907 #[doc(hidden)]
909 #[must_use]
910 pub fn with_auth_session_key(mut self, value: impl Into<String>) -> Self {
911 self.auth_session_key = Arc::from(value.into());
912 self
913 }
914
915 #[doc(hidden)]
917 #[must_use]
918 pub fn with_startup_complete(self, startup_complete: bool) -> Self {
919 self.probes.set_startup_complete(startup_complete);
920 self
921 }
922
923 #[doc(hidden)]
925 #[must_use]
926 pub fn with_draining(self, draining: bool) -> Self {
927 self.probes.set_draining(draining);
928 self
929 }
930
931 #[must_use]
933 pub fn profile(&self) -> &str {
934 self.profile.as_deref().unwrap_or("default")
935 }
936
937 #[must_use]
959 pub const fn role(&self) -> crate::config::ProcessRole {
960 self.role
961 }
962
963 #[must_use]
970 pub fn uptime(&self) -> std::time::Duration {
971 self.monotonic().saturating_duration_since(self.started_at)
972 }
973
974 #[must_use]
976 pub fn uptime_display(&self) -> String {
977 let secs = self.uptime().as_secs();
978 if secs < 60 {
979 format!("{secs}s")
980 } else if secs < 3600 {
981 format!("{}m {}s", secs / 60, secs % 60)
982 } else {
983 let hours = secs / 3600;
984 let mins = (secs % 3600) / 60;
985 format!("{hours}h {mins}m")
986 }
987 }
988
989 #[cfg(feature = "ws")]
993 #[must_use]
994 pub const fn channels(&self) -> &Channels {
995 &self.channels
996 }
997
998 #[cfg(feature = "presence")]
1000 #[must_use]
1001 pub const fn presence(&self) -> &Presence {
1002 &self.presence
1003 }
1004
1005 #[cfg(all(feature = "collab", feature = "presence"))]
1009 #[must_use]
1010 pub const fn collab(&self) -> &CollabHub {
1011 &self.collab
1012 }
1013
1014 #[cfg(feature = "ws")]
1016 #[must_use]
1017 pub fn broadcast(&self) -> crate::channels::Broadcast {
1018 self.channels.broadcast()
1019 }
1020
1021 #[must_use]
1026 pub fn shutdown_token(&self) -> CancellationToken {
1027 self.probes.shutdown_signal().child_token()
1028 }
1029
1030 #[doc(hidden)]
1032 pub fn trigger_shutdown_for_test(&self) {
1033 self.begin_shutdown();
1034 self.probes.shutdown_signal().cancel();
1035 }
1036
1037 #[doc(hidden)]
1039 pub fn set_startup_complete_for_test(&self, startup_complete: bool) {
1040 self.probes.set_startup_complete(startup_complete);
1041 }
1042
1043 #[doc(hidden)]
1045 pub fn set_draining_for_test(&self, draining: bool) {
1046 self.probes.set_draining(draining);
1047 }
1048
1049 #[doc(hidden)]
1051 pub fn begin_shutdown_for_test(&self) {
1052 self.set_draining_for_test(true);
1053 }
1054
1055 #[must_use]
1061 pub fn detached() -> Self {
1062 #[cfg(feature = "ws")]
1063 let channels = Channels::new(32);
1064 #[cfg(feature = "presence")]
1068 let presence = Presence::new(channels.clone());
1069 Self {
1070 extensions: Arc::new(std::sync::RwLock::new(HashMap::new())),
1071 #[cfg(feature = "db")]
1072 pool: None,
1073 #[cfg(feature = "db")]
1074 replica_pool: None,
1075 #[cfg(feature = "db")]
1076 shards: None,
1077 #[cfg(all(feature = "db", feature = "reporting"))]
1078 db_capture_gap: None,
1079 profile: None,
1080 role: crate::config::ProcessRole::Combined,
1081 started_at: crate::time::monotonic_now(),
1082 health_detailed: true,
1083 probes: probe::ProbeState::ready_for_test(),
1084 metrics: middleware::MetricsCollector::new(),
1085 log_levels: actuator::LogLevels::new("info"),
1086 task_registry: actuator::TaskRegistry::new(),
1087 job_registry: actuator::JobRegistry::new(),
1088 config_props: actuator::ConfigProperties::default(),
1089 metrics_source_registry: actuator::MetricsSourceRegistry::new(),
1090 health_indicator_registry: actuator::HealthIndicatorRegistry::new(),
1091 #[cfg(all(feature = "collab", feature = "presence"))]
1092 collab: CollabHub::new(channels.clone(), presence.clone()),
1093 #[cfg(feature = "presence")]
1094 presence,
1095 #[cfg(feature = "ws")]
1096 channels,
1097 policy_registry: PolicyRegistry::default(),
1098 forbidden_response: ForbiddenResponse::default(),
1099 auth_session_key: "user_id".into(),
1100 shared_cache: None,
1101 clock: Arc::new(SystemClock),
1102 entropy: std::sync::Arc::new(crate::entropy::OsEntropy),
1103 app_id: Self::next_app_id(),
1104 }
1105 }
1106
1107 #[allow(dead_code)]
1110 #[must_use]
1111 pub fn for_test() -> Self {
1112 Self::detached()
1113 }
1114}
1115
1116#[cfg(feature = "db")]
1117impl DbState for AppState {
1118 fn clock(&self) -> Arc<dyn ClockSource> {
1119 self.clock_arc()
1120 }
1121
1122 fn metrics(&self) -> Option<&crate::middleware::MetricsCollector> {
1123 Some(&self.metrics)
1124 }
1125
1126 fn pool(
1127 &self,
1128 ) -> Option<&diesel_async::pooled_connection::deadpool::Pool<crate::db::RuntimeConnection>>
1129 {
1130 self.pool.as_ref()
1131 }
1132
1133 fn replica_pool(
1134 &self,
1135 ) -> Option<&diesel_async::pooled_connection::deadpool::Pool<crate::db::RuntimeConnection>>
1136 {
1137 self.replica_pool.as_ref()
1138 }
1139
1140 #[cfg(feature = "reporting")]
1141 fn db_capture_gap(&self) -> Option<Arc<str>> {
1142 self.db_capture_gap.clone()
1143 }
1144
1145 fn read_pool(
1146 &self,
1147 ) -> Option<&diesel_async::pooled_connection::deadpool::Pool<crate::db::RuntimeConnection>>
1148 {
1149 Self::read_pool(self)
1150 }
1151
1152 fn shards(&self) -> Option<&crate::sharding::ShardSet> {
1153 self.shards.as_ref()
1154 }
1155
1156 fn db_interceptors(
1157 &self,
1158 ) -> Vec<std::sync::Arc<dyn crate::interceptor::DbConnectionInterceptor>> {
1159 self.extension::<Arc<dyn crate::interceptor::DbConnectionInterceptor>>()
1160 .map(|arc| vec![(*arc).clone()])
1161 .unwrap_or_default()
1162 }
1163 fn statement_timeout(&self) -> Option<std::time::Duration> {
1164 self.extension::<crate::config::AutumnConfig>()
1165 .and_then(|cfg| cfg.database.statement_timeout)
1166 }
1167 fn idle_in_transaction_timeout(&self) -> Option<std::time::Duration> {
1168 self.extension::<crate::config::AutumnConfig>()
1169 .and_then(|cfg| cfg.database.idle_in_transaction_timeout)
1170 }
1171
1172 fn slow_query_threshold(&self) -> std::time::Duration {
1173 self.extension::<crate::config::AutumnConfig>().map_or_else(
1174 || std::time::Duration::from_millis(500),
1175 |cfg| cfg.database.slow_query_threshold,
1176 )
1177 }
1178}
1179
1180impl crate::probe::ProvideProbeState for AppState {
1181 fn probes(&self) -> &crate::probe::ProbeState {
1182 &self.probes
1183 }
1184
1185 fn health_detailed(&self) -> bool {
1186 self.health_detailed
1187 }
1188
1189 fn profile(&self) -> &str {
1190 self.profile()
1191 }
1192
1193 fn uptime_display(&self) -> String {
1194 self.uptime_display()
1195 }
1196
1197 #[cfg(feature = "db")]
1198 fn pool(
1199 &self,
1200 ) -> Option<&diesel_async::pooled_connection::deadpool::Pool<crate::db::RuntimeConnection>>
1201 {
1202 self.pool.as_ref()
1203 }
1204
1205 #[cfg(feature = "db")]
1206 fn replica_pool(
1207 &self,
1208 ) -> Option<&diesel_async::pooled_connection::deadpool::Pool<crate::db::RuntimeConnection>>
1209 {
1210 self.replica_pool.as_ref()
1211 }
1212
1213 fn health_indicator_registry(&self) -> Option<&crate::actuator::HealthIndicatorRegistry> {
1214 Some(&self.health_indicator_registry)
1215 }
1216}
1217
1218impl crate::actuator::ProvideActuatorState for AppState {
1219 fn metrics(&self) -> &crate::middleware::MetricsCollector {
1220 &self.metrics
1221 }
1222
1223 fn log_levels(&self) -> &crate::actuator::LogLevels {
1224 &self.log_levels
1225 }
1226
1227 fn task_registry(&self) -> &crate::actuator::TaskRegistry {
1228 &self.task_registry
1229 }
1230
1231 fn job_registry(&self) -> &crate::actuator::JobRegistry {
1232 &self.job_registry
1233 }
1234
1235 fn config_props(&self) -> &crate::actuator::ConfigProperties {
1236 &self.config_props
1237 }
1238
1239 fn profile(&self) -> &str {
1240 self.profile()
1241 }
1242
1243 fn uptime_display(&self) -> String {
1244 self.uptime_display()
1245 }
1246
1247 fn metrics_source_registry(&self) -> Option<&crate::actuator::MetricsSourceRegistry> {
1248 Some(&self.metrics_source_registry)
1249 }
1250
1251 fn health_indicator_registry(&self) -> Option<&crate::actuator::HealthIndicatorRegistry> {
1252 Some(&self.health_indicator_registry)
1253 }
1254
1255 fn health_detailed(&self) -> bool {
1256 self.health_detailed
1257 }
1258
1259 fn shadow(&self) -> Option<crate::shadow::ShadowHandle> {
1267 self.extension::<crate::shadow::ShadowHandle>()
1268 .map(|handle| (*handle).clone())
1269 }
1270
1271 fn architecture_graph_json(&self) -> Option<bytes::Bytes> {
1272 crate::graph::served_json(self)
1273 }
1274
1275 fn data_capsules(
1276 &self,
1277 ) -> Result<crate::gdpr::portability::CapsuleService, crate::gdpr::portability::DataCapsuleError>
1278 {
1279 crate::gdpr::portability::CapsuleService::from_state(self)
1280 }
1281
1282 fn cost_signal(&self) -> Option<crate::cost::CostSignal> {
1283 self.extension::<crate::cost::CostSignal>()
1284 .map(|signal| (*signal).clone())
1285 }
1286
1287 fn cost_accountant(&self) -> Option<crate::cost::CostAccountant> {
1288 if !self.config_arc().cost.enabled {
1291 return None;
1292 }
1293 self.extension::<crate::cost::CostAccountant>()
1294 .map(|accountant| (*accountant).clone())
1295 }
1296
1297 fn deploy_version(&self) -> String {
1298 self.extension::<crate::canary::CanaryState>().map_or_else(
1299 || crate::canary::STABLE.to_owned(),
1300 |c| c.version().to_owned(),
1301 )
1302 }
1303
1304 #[cfg(feature = "ws")]
1305 fn channels(&self) -> &crate::channels::Channels {
1306 &self.channels
1307 }
1308
1309 #[cfg(feature = "ws")]
1310 fn shutdown_token(&self) -> tokio_util::sync::CancellationToken {
1311 self.shutdown_token()
1312 }
1313
1314 #[cfg(feature = "db")]
1315 fn pool(
1316 &self,
1317 ) -> Option<&diesel_async::pooled_connection::deadpool::Pool<crate::db::RuntimeConnection>>
1318 {
1319 self.pool.as_ref()
1320 }
1321
1322 #[cfg(feature = "db")]
1323 fn probe_state(&self) -> Option<&crate::probe::ProbeState> {
1324 Some(&self.probes)
1325 }
1326
1327 #[cfg(feature = "db")]
1328 fn shards(&self) -> Option<&crate::sharding::ShardSet> {
1329 self.shards.as_ref()
1330 }
1331
1332 #[cfg(feature = "db")]
1333 fn replica_pool(
1334 &self,
1335 ) -> Option<&diesel_async::pooled_connection::deadpool::Pool<crate::db::RuntimeConnection>>
1336 {
1337 self.replica_pool.as_ref()
1338 }
1339 #[cfg(feature = "http-client")]
1346 fn webhook_outbound(&self) -> Option<crate::webhook_outbound::WebhookOutboundManager> {
1347 self.extension::<crate::webhook_outbound::WebhookOutboundManager>()
1348 .map(|x| (*x).clone())
1349 }
1350
1351 fn log_buffer(&self) -> Option<crate::log::capture::LogBuffer> {
1352 self.extension::<crate::log::capture::LogBuffer>()
1353 .map(|x| (*x).clone())
1354 }
1355}
1356
1357impl std::fmt::Debug for AppState {
1358 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1359 let mut s = f.debug_struct("AppState");
1360 #[cfg(feature = "db")]
1361 s.field(
1362 "pool",
1363 &self
1364 .pool
1365 .as_ref()
1366 .map(|p| format!("Pool(max={})", p.status().max_size)),
1367 );
1368 s.field(
1369 "extensions",
1370 &self
1371 .extensions
1372 .read()
1373 .map_or(0, |extensions| extensions.len()),
1374 );
1375 s.field("profile", &self.profile)
1376 .field("started_at", &self.started_at)
1377 .field("health_detailed", &self.health_detailed)
1378 .field("probes", &self.probes)
1379 .field("metrics", &"MetricsCollector")
1380 .field("log_levels", &"LogLevels")
1381 .field("task_registry", &"TaskRegistry")
1382 .finish_non_exhaustive()
1383 }
1384}
1385
1386#[cfg(test)]
1387impl AppState {
1388 pub(crate) fn test_default() -> Self {
1398 #[cfg(feature = "ws")]
1403 let channels = crate::channels::Channels::new(32);
1404 #[cfg(feature = "presence")]
1405 let presence = crate::presence::Presence::new(channels.clone());
1406 Self {
1407 extensions: std::sync::Arc::new(std::sync::RwLock::new(
1408 std::collections::HashMap::new(),
1409 )),
1410 #[cfg(feature = "db")]
1411 pool: None,
1412 #[cfg(feature = "db")]
1413 replica_pool: None,
1414 #[cfg(feature = "db")]
1415 shards: None,
1416 #[cfg(all(feature = "db", feature = "reporting"))]
1417 db_capture_gap: None,
1418 profile: None,
1419 role: crate::config::ProcessRole::Combined,
1420 started_at: crate::time::monotonic_now(),
1421 health_detailed: false,
1422 probes: crate::probe::ProbeState::ready_for_test(),
1423 metrics: crate::middleware::MetricsCollector::new(),
1424 log_levels: crate::actuator::LogLevels::new("info"),
1425 task_registry: crate::actuator::TaskRegistry::new(),
1426 job_registry: crate::actuator::JobRegistry::new(),
1427 config_props: crate::actuator::ConfigProperties::default(),
1428 metrics_source_registry: crate::actuator::MetricsSourceRegistry::new(),
1429 health_indicator_registry: crate::actuator::HealthIndicatorRegistry::new(),
1430 #[cfg(all(feature = "collab", feature = "presence"))]
1431 collab: CollabHub::new(channels.clone(), presence.clone()),
1432 #[cfg(feature = "presence")]
1433 presence,
1434 #[cfg(feature = "ws")]
1435 channels,
1436 policy_registry: crate::authorization::PolicyRegistry::default(),
1437 forbidden_response: crate::authorization::ForbiddenResponse::default(),
1438 auth_session_key: "user_id".into(),
1439 shared_cache: None,
1440 clock: std::sync::Arc::new(crate::time::SystemClock),
1441 entropy: std::sync::Arc::new(crate::entropy::OsEntropy),
1442 app_id: Self::next_app_id(),
1443 }
1444 }
1445}
1446
1447#[cfg(test)]
1448mod tests {
1449 struct Counted(std::sync::atomic::AtomicUsize);
1451
1452 impl crate::actuator::HealthIndicator for Counted {
1453 fn check(&self) -> futures::future::BoxFuture<'_, crate::actuator::HealthCheckOutput> {
1454 self.0.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
1455 Box::pin(async { crate::actuator::HealthCheckOutput::up() })
1456 }
1457 }
1458
1459 async fn checks_after_two_runs(health: &crate::config::HealthConfig) -> usize {
1460 let state = AppState::for_test();
1461 state.apply_health_config(health);
1462 let counted = Arc::new(Counted(std::sync::atomic::AtomicUsize::new(0)));
1463 state
1464 .health_indicator_registry
1465 .register(
1466 "counted",
1467 crate::actuator::IndicatorGroup::Readiness,
1468 counted.clone(),
1469 )
1470 .expect("register");
1471 state.health_indicator_registry.run_readiness().await;
1472 state.health_indicator_registry.run_readiness().await;
1473 counted.0.load(std::sync::atomic::Ordering::SeqCst)
1474 }
1475
1476 #[tokio::test]
1477 async fn apply_health_config_sets_the_indicator_cache_ttl() {
1478 let cached = crate::config::HealthConfig::default();
1479 assert_eq!(checks_after_two_runs(&cached).await, 1);
1480
1481 let uncached = crate::config::HealthConfig {
1482 cache_ttl_ms: 0,
1483 ..crate::config::HealthConfig::default()
1484 };
1485 assert_eq!(checks_after_two_runs(&uncached).await, 2);
1486 }
1487
1488 #[test]
1495 #[should_panic(expected = "construction-time builder")]
1496 fn re_clocking_an_initialized_state_is_refused() {
1497 use chrono::{TimeZone, Utc};
1498
1499 let state = AppState::detached();
1500 state.job_registry().register_on_queue("probe", "default");
1502
1503 let epoch = Utc.with_ymd_and_hms(2020, 1, 1, 0, 0, 0).unwrap();
1504 let _ = state.with_clock(Arc::new(crate::time::FixedClock::at(epoch)));
1505 }
1506
1507 #[test]
1517 fn re_clocking_a_state_keeps_its_gauges_on_its_own_clock() {
1518 use chrono::{TimeZone, Utc};
1519
1520 let real_state = AppState::detached();
1521 let cloned_before = real_state.clone();
1522
1523 let epoch = Utc.with_ymd_and_hms(2020, 1, 1, 0, 0, 0).unwrap();
1526 let virtual_state = real_state.with_clock(Arc::new(crate::time::FixedClock::at(epoch)));
1527
1528 let ready_at = u64::try_from(epoch.timestamp_millis()).unwrap() + 60_000;
1531 virtual_state
1532 .job_registry()
1533 .register_on_queue("probe", "default");
1534 virtual_state
1535 .job_registry()
1536 .record_enqueue_scheduled("probe", ready_at);
1537
1538 let virtual_depth = virtual_state.job_registry().queue_snapshot()["default"].depth;
1539 assert_eq!(
1540 virtual_depth, 0,
1541 "the state that stamped the mark must judge it as scheduled"
1542 );
1543
1544 let clone_snapshot = cloned_before.job_registry().queue_snapshot();
1549 assert!(
1550 !clone_snapshot.contains_key("default"),
1551 "a state cloned before the clock swap must not share gauges with a \
1552 state on a different clock; it saw {clone_snapshot:?}"
1553 );
1554 }
1555
1556 use super::*;
1557 #[cfg(feature = "db")]
1558 use crate::config;
1559 #[cfg(feature = "db")]
1560 use crate::db;
1561
1562 #[test]
1563 fn app_state_debug_without_pool() {
1564 let state = AppState::for_test().with_profile("dev");
1565 let debug = format!("{state:?}");
1566 assert!(debug.contains("AppState"));
1567 assert!(debug.contains("dev"));
1568 }
1569
1570 #[cfg(feature = "db")]
1571 #[test]
1572 fn app_state_debug_with_pool() {
1573 let config = config::DatabaseConfig {
1574 url: Some(crate::test_urls::primary("test")),
1575 pool_size: 5,
1576 ..Default::default()
1577 };
1578 let pool = db::create_pool(&config).unwrap().unwrap();
1579 let state = AppState::for_test().with_pool(pool);
1580 let debug = format!("{state:?}");
1581 assert!(debug.contains("Pool(max=5)"));
1582 }
1583
1584 #[cfg(feature = "db")]
1585 #[test]
1586 fn database_topology_state_exposes_replica_as_read_pool() {
1587 let primary_config = config::DatabaseConfig {
1588 url: Some(crate::test_urls::primary("primary")),
1589 pool_size: 5,
1590 ..Default::default()
1591 };
1592 let replica_config = config::DatabaseConfig {
1593 url: Some(crate::test_urls::replica("primary", "replica")),
1594 pool_size: 2,
1595 ..Default::default()
1596 };
1597 let primary = db::create_pool(&primary_config).unwrap().unwrap();
1598 let replica = db::create_pool(&replica_config).unwrap().unwrap();
1599
1600 let state = AppState::for_test()
1601 .with_pool(primary)
1602 .with_replica_pool(replica);
1603
1604 assert_eq!(state.pool().expect("primary pool").status().max_size, 5);
1605 assert_eq!(
1606 state
1607 .replica_pool()
1608 .expect("replica pool")
1609 .status()
1610 .max_size,
1611 2
1612 );
1613 assert_eq!(state.read_pool().expect("read pool").status().max_size, 2);
1614 }
1615
1616 #[cfg(feature = "db")]
1617 #[test]
1618 fn read_pool_uses_primary_when_replica_is_unready_and_policy_allows_fallback() {
1619 let primary_config = config::DatabaseConfig {
1620 url: Some(crate::test_urls::primary("primary")),
1621 pool_size: 5,
1622 ..Default::default()
1623 };
1624 let replica_config = config::DatabaseConfig {
1625 url: Some(crate::test_urls::replica("primary", "replica")),
1626 pool_size: 2,
1627 ..Default::default()
1628 };
1629 let primary = db::create_pool(&primary_config).unwrap().unwrap();
1630 let replica = db::create_pool(&replica_config).unwrap().unwrap();
1631
1632 let state = AppState::for_test()
1633 .with_pool(primary)
1634 .with_replica_pool(replica);
1635 state
1636 .probes()
1637 .configure_replica_dependency(config::ReplicaFallback::Primary);
1638 state
1639 .probes()
1640 .mark_replica_unready("replica migrations lag primary");
1641
1642 assert_eq!(state.read_pool().expect("read pool").status().max_size, 5);
1643 assert_eq!(
1644 db::DbState::read_pool(&state)
1645 .expect("trait read pool")
1646 .status()
1647 .max_size,
1648 5
1649 );
1650 }
1651
1652 #[cfg(feature = "db")]
1653 #[test]
1654 fn read_pool_does_not_route_to_unready_replica_when_policy_fails_readiness() {
1655 let primary_config = config::DatabaseConfig {
1656 url: Some(crate::test_urls::primary("primary")),
1657 pool_size: 5,
1658 ..Default::default()
1659 };
1660 let replica_config = config::DatabaseConfig {
1661 url: Some(crate::test_urls::replica("primary", "replica")),
1662 pool_size: 2,
1663 ..Default::default()
1664 };
1665 let primary = db::create_pool(&primary_config).unwrap().unwrap();
1666 let replica = db::create_pool(&replica_config).unwrap().unwrap();
1667
1668 let state = AppState::for_test()
1669 .with_pool(primary)
1670 .with_replica_pool(replica);
1671 state
1672 .probes()
1673 .configure_replica_dependency(config::ReplicaFallback::FailReadiness);
1674 state
1675 .probes()
1676 .mark_replica_unready("replica connection failed");
1677
1678 assert!(state.read_pool().is_none());
1679 }
1680
1681 #[cfg(feature = "db")]
1682 #[tokio::test]
1683 async fn readiness_fails_when_app_state_replica_is_unready_and_policy_is_fail_readiness() {
1684 let primary_config = config::DatabaseConfig {
1685 url: Some(crate::test_urls::primary("primary")),
1686 pool_size: 5,
1687 ..Default::default()
1688 };
1689 let replica_config = config::DatabaseConfig {
1690 url: Some(crate::test_urls::replica("primary", "replica")),
1691 pool_size: 2,
1692 ..Default::default()
1693 };
1694 let primary = db::create_pool(&primary_config).unwrap().unwrap();
1695 let replica = db::create_pool(&replica_config).unwrap().unwrap();
1696
1697 let state = AppState::for_test()
1698 .with_pool(primary)
1699 .with_replica_pool(replica);
1700 state
1701 .probes()
1702 .configure_replica_dependency(config::ReplicaFallback::FailReadiness);
1703 state
1704 .probes()
1705 .mark_replica_unready("replica migrations lag primary");
1706
1707 let (status, _) = crate::probe::readiness_response(&state).await;
1708
1709 assert_eq!(status, http::StatusCode::SERVICE_UNAVAILABLE);
1710 }
1711
1712 #[test]
1713 fn detached_state_starts_without_profile() {
1714 let state = AppState::detached();
1715
1716 assert_eq!(state.profile(), "default");
1717 }
1718
1719 fn require_clone<T: Clone>(t: &T) -> T {
1720 t.clone()
1721 }
1722
1723 #[test]
1724 fn app_state_is_clone() {
1725 let state = AppState::for_test();
1726 let _cloned = require_clone(&state);
1727 }
1728
1729 #[test]
1730 fn app_state_profile_accessor() {
1731 let state = AppState::for_test().with_profile("staging");
1732 assert_eq!(state.profile(), "staging");
1733 }
1734
1735 #[test]
1736 fn app_state_deploy_version_defaults_to_stable() {
1737 use crate::actuator::ProvideActuatorState;
1738 let state = AppState::for_test();
1739 assert_eq!(state.deploy_version(), crate::canary::STABLE);
1740 }
1741
1742 #[test]
1743 fn app_state_deploy_version_reads_canary_extension() {
1744 use crate::actuator::ProvideActuatorState;
1745 let state = AppState::for_test();
1746 state.insert_extension(crate::canary::CanaryState::new(crate::canary::CANARY));
1747 assert_eq!(state.deploy_version(), crate::canary::CANARY);
1748 }
1749
1750 #[test]
1751 fn app_state_profile_default() {
1752 let state = AppState::for_test();
1753 assert_eq!(state.profile(), "default");
1754 }
1755
1756 #[test]
1757 fn app_state_uptime_display() {
1758 let state = AppState::for_test();
1759 let display = state.uptime_display();
1760 assert!(
1761 display.contains('s'),
1762 "uptime should contain 's': {display}"
1763 );
1764 }
1765
1766 #[test]
1767 fn app_state_accessors() {
1768 let state = AppState::for_test();
1769
1770 let _metrics = state.metrics();
1772 let _log_levels = state.log_levels();
1773 let _task_registry = state.task_registry();
1774 let _config_props = state.config_props();
1775
1776 #[cfg(feature = "db")]
1777 {
1778 let _pool = state.pool();
1779 }
1780 let _missing = state.extension::<String>();
1781 }
1782
1783 #[test]
1784 fn app_state_runtime_extensions_round_trip() {
1785 let state = AppState::for_test();
1786 state.insert_extension(String::from("haunted"));
1787
1788 let stored = state
1789 .extension::<String>()
1790 .expect("runtime extension should be installed");
1791
1792 assert_eq!(stored.as_str(), "haunted");
1793 }
1794
1795 #[test]
1801 fn config_arc_returns_the_installed_arc_without_deep_cloning() {
1802 let state = AppState::for_test();
1803 state.insert_extension(crate::config::AutumnConfig {
1804 profile: Some("staging".to_owned()),
1805 ..Default::default()
1806 });
1807
1808 let installed = state
1809 .extension::<crate::config::AutumnConfig>()
1810 .expect("config extension should be installed");
1811 let first = state.config_arc();
1812 let second = state.config_arc();
1813
1814 assert!(
1815 Arc::ptr_eq(&first, &installed),
1816 "config_arc must return the extension map's Arc, not a fresh allocation"
1817 );
1818 assert!(
1819 Arc::ptr_eq(&first, &second),
1820 "two config_arc calls must share one allocation"
1821 );
1822
1823 state.insert_extension(crate::config::AutumnConfig {
1824 profile: Some("prod".to_owned()),
1825 ..Default::default()
1826 });
1827 let after_reinsert = state.config_arc();
1828
1829 assert_eq!(
1830 after_reinsert.profile.as_deref(),
1831 Some("prod"),
1832 "config_arc must observe a config re-inserted after construction"
1833 );
1834 assert!(
1835 !Arc::ptr_eq(&first, &after_reinsert),
1836 "a re-inserted config must replace the handle, not alias the old one"
1837 );
1838 }
1839
1840 #[test]
1841 fn config_arc_falls_back_to_default_without_extension() {
1842 let state = AppState::for_test();
1843 let defaults = crate::config::AutumnConfig::default();
1844
1845 let fallback = state.config_arc();
1846
1847 assert_eq!(fallback.profile, defaults.profile);
1848 assert_eq!(fallback.server.port, defaults.server.port);
1849 assert_eq!(fallback.server.host, defaults.server.host);
1850 assert!(
1854 state.extension::<crate::config::AutumnConfig>().is_none(),
1855 "the fallback default must not be written into the extension map"
1856 );
1857 }
1858
1859 #[test]
1862 fn config_matches_config_arc_by_value() {
1863 let absent = AppState::for_test();
1864 assert_eq!(
1865 format!("{:?}", absent.config()),
1866 format!("{:?}", *absent.config_arc()),
1867 "the two accessors must agree in the no-extension fallback case"
1868 );
1869
1870 let present = AppState::for_test();
1871 present.insert_extension(crate::config::AutumnConfig {
1872 profile: Some("staging".to_owned()),
1873 ..Default::default()
1874 });
1875 assert_eq!(
1876 format!("{:?}", present.config()),
1877 format!("{:?}", *present.config_arc()),
1878 "the two accessors must agree when a config is installed"
1879 );
1880 }
1881
1882 #[test]
1891 fn config_signature_is_unchanged() {
1892 let _: fn(&AppState) -> crate::config::AutumnConfig = AppState::config;
1893 let _: fn(&AppState) -> Arc<crate::config::AutumnConfig> = AppState::config_arc;
1894 }
1895}