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