| 1 | //! Process-wide shared state. |
| 2 | |
| 3 | use std::path::PathBuf; |
| 4 | use std::sync::Arc; |
| 5 | use std::sync::atomic::{AtomicU64, Ordering}; |
| 6 | |
| 7 | use arc_swap::ArcSwap; |
| 8 | use tokio::sync::Semaphore; |
| 9 | |
| 10 | use crate::cache::Cache; |
| 11 | use crate::config::{Config, ConfigError, Listen}; |
| 12 | use crate::index::Indexer; |
| 13 | use crate::server::assets::Assets; |
| 14 | |
| 15 | pub struct AppState { |
| 16 | /// Current config; swapped atomically on reload. |
| 17 | pub config: ArcSwap<Config>, |
| 18 | /// Bumped on every successful reload; part of page ETags. |
| 19 | pub config_generation: AtomicU64, |
| 20 | pub assets: Assets, |
| 21 | pub cache: Cache, |
| 22 | /// Bounds concurrent git work (`limits.git_parallel`). |
| 23 | pub git_permits: Semaphore, |
| 24 | pub index: Indexer, |
| 25 | /// Wakes the index watcher for an immediate full rescan. |
| 26 | pub rescan: tokio::sync::Notify, |
| 27 | /// Command-line overrides re-applied after every reload. |
| 28 | pub overrides: Overrides, |
| 29 | } |
| 30 | |
| 31 | #[derive(Debug, Clone, Default)] |
| 32 | pub struct Overrides { |
| 33 | pub config_path: Option<PathBuf>, |
| 34 | pub listen: Option<Listen>, |
| 35 | } |
| 36 | |
| 37 | impl Overrides { |
| 38 | pub fn load(&self) -> Result<Config, ConfigError> { |
| 39 | let mut config = Config::load(self.config_path.as_deref())?; |
| 40 | if let Some(listen) = &self.listen { |
| 41 | config.server.listen = listen.clone(); |
| 42 | } |
| 43 | Ok(config) |
| 44 | } |
| 45 | } |
| 46 | |
| 47 | impl AppState { |
| 48 | pub fn new(config: Config, overrides: Overrides) -> Arc<Self> { |
| 49 | Arc::new(AppState { |
| 50 | cache: Cache::new(config.cache.size_mb), |
| 51 | git_permits: Semaphore::new(config.git_parallel()), |
| 52 | config_generation: AtomicU64::new(0), |
| 53 | config: ArcSwap::from_pointee(config), |
| 54 | assets: Assets::new(), |
| 55 | index: Indexer::new(), |
| 56 | rescan: tokio::sync::Notify::new(), |
| 57 | overrides, |
| 58 | }) |
| 59 | } |
| 60 | |
| 61 | /// Re-read the config file and swap it in. On error the running config |
| 62 | /// stays in place. Returns the new config. |
| 63 | pub fn reload(&self) -> Result<Arc<Config>, ConfigError> { |
| 64 | let mut new = self.overrides.load()?; |
| 65 | let old = self.config.load(); |
| 66 | // Settings baked into the listener and middleware stack keep their |
| 67 | // running values so the config always describes the live server. |
| 68 | if new.server.listen != old.server.listen { |
| 69 | tracing::warn!( |
| 70 | "server.listen changed to {}; restart required to take effect", |
| 71 | new.server.listen |
| 72 | ); |
| 73 | new.server.listen = old.server.listen.clone(); |
| 74 | } |
| 75 | if new.server.base_path != old.server.base_path { |
| 76 | tracing::warn!("server.base_path changed; restart required to take effect"); |
| 77 | new.server.base_path = old.server.base_path.clone(); |
| 78 | } |
| 79 | if new.server.compression != old.server.compression |
| 80 | || new.limits.request_timeout_secs != old.limits.request_timeout_secs |
| 81 | { |
| 82 | tracing::warn!( |
| 83 | "server.compression / limits.request_timeout_secs changed; restart required" |
| 84 | ); |
| 85 | new.server.compression = old.server.compression; |
| 86 | new.limits.request_timeout_secs = old.limits.request_timeout_secs; |
| 87 | } |
| 88 | if new.cache.size_mb != old.cache.size_mb || new.git_parallel() != old.git_parallel() { |
| 89 | tracing::warn!("cache.size_mb / limits.git_parallel changed; restart required"); |
| 90 | new.cache.size_mb = old.cache.size_mb; |
| 91 | new.limits.git_parallel = old.limits.git_parallel; |
| 92 | } |
| 93 | let new = Arc::new(new); |
| 94 | self.config.store(new.clone()); |
| 95 | self.config_generation.fetch_add(1, Ordering::Relaxed); |
| 96 | Ok(new) |
| 97 | } |
| 98 | |
| 99 | /// Run blocking git work on the blocking pool, at most |
| 100 | /// `limits.git_parallel` jobs at a time. |
| 101 | pub async fn blocking<T: Send + 'static>( |
| 102 | self: &Arc<Self>, |
| 103 | work: impl FnOnce(&AppState) -> T + Send + 'static, |
| 104 | ) -> anyhow::Result<T> { |
| 105 | let _permit = self.git_permits.acquire().await?; |
| 106 | let state = self.clone(); |
| 107 | Ok(tokio::task::spawn_blocking(move || work(&state)).await?) |
| 108 | } |
| 109 | } |