rust_zero_core/
config_center.rs

1use crate::{parse_config, ConfigError, ConfigFormat};
2use serde::de::DeserializeOwned;
3use std::{
4    fmt,
5    sync::{
6        atomic::{AtomicU64, Ordering},
7        Arc, RwLock,
8    },
9};
10use tokio::sync::watch;
11
12/// An immutable, versioned view of a dynamic configuration value.
13#[derive(Debug)]
14pub struct ConfigSnapshot<T> {
15    generation: u64,
16    raw: Arc<str>,
17    value: Arc<T>,
18}
19
20impl<T> Clone for ConfigSnapshot<T> {
21    fn clone(&self) -> Self {
22        Self {
23            generation: self.generation,
24            raw: Arc::clone(&self.raw),
25            value: Arc::clone(&self.value),
26        }
27    }
28}
29
30impl<T> ConfigSnapshot<T> {
31    pub fn generation(&self) -> u64 {
32        self.generation
33    }
34
35    pub fn raw(&self) -> &str {
36        &self.raw
37    }
38
39    pub fn value(&self) -> &Arc<T> {
40        &self.value
41    }
42}
43
44struct DynamicConfigState<T> {
45    snapshot: RwLock<ConfigSnapshot<T>>,
46    generation: AtomicU64,
47    updates: watch::Sender<ConfigSnapshot<T>>,
48}
49
50/// A typed configuration-center primitive with atomic updates and subscriptions.
51///
52/// Backend adapters feed new serialized values through [`Self::update`]. Invalid
53/// updates are rejected without disturbing the last known-good snapshot.
54pub struct DynamicConfig<T> {
55    format: ConfigFormat,
56    state: Arc<DynamicConfigState<T>>,
57}
58
59impl<T> Clone for DynamicConfig<T> {
60    fn clone(&self) -> Self {
61        Self {
62            format: self.format,
63            state: Arc::clone(&self.state),
64        }
65    }
66}
67
68impl<T> DynamicConfig<T>
69where
70    T: DeserializeOwned + Send + Sync + 'static,
71{
72    pub fn new(contents: &str, format: ConfigFormat) -> Result<Self, ConfigCenterError> {
73        let value = Arc::new(parse_config(contents, format)?);
74        let snapshot = ConfigSnapshot {
75            generation: 1,
76            raw: Arc::from(contents),
77            value,
78        };
79        let (updates, _) = watch::channel(snapshot.clone());
80
81        Ok(Self {
82            format,
83            state: Arc::new(DynamicConfigState {
84                snapshot: RwLock::new(snapshot),
85                generation: AtomicU64::new(1),
86                updates,
87            }),
88        })
89    }
90
91    pub fn snapshot(&self) -> ConfigSnapshot<T> {
92        self.state
93            .snapshot
94            .read()
95            .expect("dynamic configuration lock poisoned")
96            .clone()
97    }
98
99    pub fn current(&self) -> Arc<T> {
100        Arc::clone(self.snapshot().value())
101    }
102
103    /// Subscribes to future changes. The receiver starts with the current value.
104    pub fn subscribe(&self) -> watch::Receiver<ConfigSnapshot<T>> {
105        self.state.updates.subscribe()
106    }
107
108    /// Atomically installs a new value after parsing it successfully.
109    pub fn update(&self, contents: &str) -> Result<ConfigSnapshot<T>, ConfigCenterError> {
110        let value = Arc::new(parse_config(contents, self.format)?);
111        let generation = self.state.generation.fetch_add(1, Ordering::AcqRel) + 1;
112        let snapshot = ConfigSnapshot {
113            generation,
114            raw: Arc::from(contents),
115            value,
116        };
117
118        *self
119            .state
120            .snapshot
121            .write()
122            .expect("dynamic configuration lock poisoned") = snapshot.clone();
123        self.state.updates.send_replace(snapshot.clone());
124        Ok(snapshot)
125    }
126}
127
128#[derive(Debug)]
129pub struct ConfigCenterError(ConfigError);
130
131impl fmt::Display for ConfigCenterError {
132    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
133        write!(formatter, "dynamic configuration update failed: {}", self.0)
134    }
135}
136
137impl std::error::Error for ConfigCenterError {
138    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
139        Some(&self.0)
140    }
141}
142
143impl From<ConfigError> for ConfigCenterError {
144    fn from(error: ConfigError) -> Self {
145        Self(error)
146    }
147}
148
149#[cfg(test)]
150mod tests {
151    use super::*;
152    use serde::Deserialize;
153
154    #[derive(Debug, Deserialize, PartialEq, Eq)]
155    struct Limits {
156        requests: u64,
157    }
158
159    #[tokio::test]
160    async fn publishes_valid_atomic_updates() {
161        let config = DynamicConfig::<Limits>::new("requests = 10", ConfigFormat::Toml).unwrap();
162        let mut changes = config.subscribe();
163
164        let snapshot = config.update("requests = 20").unwrap();
165        changes.changed().await.unwrap();
166
167        assert_eq!(snapshot.generation(), 2);
168        assert_eq!(changes.borrow().value().requests, 20);
169        assert_eq!(config.current().requests, 20);
170    }
171
172    #[test]
173    fn retains_last_known_good_value() {
174        let config = DynamicConfig::<Limits>::new("requests = 10", ConfigFormat::Toml).unwrap();
175
176        assert!(config.update("requests = \"invalid\"").is_err());
177        assert_eq!(config.snapshot().generation(), 1);
178        assert_eq!(config.current().requests, 10);
179    }
180}
rust_zero_core/config_center.rs

rust_zero_core/
config_center.rs
