aster_forge_config/notification/
runtime.rs

1use std::future::Future;
2use tokio_util::sync::CancellationToken;
3
4use crate::Result;
5
6use super::message::{ConfigNotificationSource, ConfigReloadMessage, ConfigReloadWorkerConfig};
7use super::notifier::SharedConfigChangeNotifier;
8use super::supervisor::{
9    ConfigReloadObserver, ConfigSyncConnectionObserver, run_config_reload_supervisor,
10    run_config_reload_supervisor_with_observers, run_config_reload_worker,
11    run_config_reload_worker_with_observer,
12};
13
14/// Namespaced runtime handle for cross-process config synchronization.
15///
16/// This type is the product-facing boundary for config sync. It keeps the
17/// namespace, runtime identity, backend notifier, publish helper, and subscription
18/// worker wiring together so product crates only provide their authoritative
19/// reload callback.
20#[derive(Clone)]
21pub struct ConfigSyncRuntime {
22    namespace: String,
23    runtime_id: String,
24    notifier: Option<SharedConfigChangeNotifier>,
25}
26
27impl ConfigSyncRuntime {
28    /// Creates an enabled runtime from a namespace, runtime ID, and notifier.
29    pub fn new(
30        namespace: impl Into<String>,
31        runtime_id: impl Into<String>,
32        notifier: impl Into<SharedConfigChangeNotifier>,
33    ) -> Self {
34        Self {
35            namespace: namespace.into(),
36            runtime_id: runtime_id.into(),
37            notifier: Some(notifier.into()),
38        }
39    }
40
41    /// Creates a disabled runtime with a generated runtime ID.
42    pub fn disabled(namespace: impl Into<String>) -> Self {
43        Self::disabled_with_runtime_id(namespace, aster_forge_utils::id::new_runtime_id())
44    }
45
46    /// Creates a disabled runtime with an explicit runtime ID.
47    pub fn disabled_with_runtime_id(
48        namespace: impl Into<String>,
49        runtime_id: impl Into<String>,
50    ) -> Self {
51        Self {
52            namespace: namespace.into(),
53            runtime_id: runtime_id.into(),
54            notifier: None,
55        }
56    }
57
58    /// Creates a disabled runtime for tests and single-process defaults.
59    pub fn disabled_for_test(namespace: impl Into<String>) -> Self {
60        Self::disabled_with_runtime_id(namespace, "test-runtime")
61    }
62
63    /// Creates an enabled runtime from an explicit notifier for tests.
64    pub fn with_notifier_for_test(
65        namespace: impl Into<String>,
66        runtime_id: impl Into<String>,
67        notifier: impl Into<SharedConfigChangeNotifier>,
68    ) -> Self {
69        Self::new(namespace, runtime_id, notifier)
70    }
71
72    /// Returns the product namespace this runtime accepts and publishes.
73    #[must_use]
74    pub fn namespace(&self) -> &str {
75        &self.namespace
76    }
77
78    /// Returns the process runtime ID.
79    #[must_use]
80    pub fn runtime_id(&self) -> &str {
81        &self.runtime_id
82    }
83
84    /// Returns the configured notifier, if cross-process sync is enabled.
85    #[must_use]
86    pub fn notifier(&self) -> Option<&SharedConfigChangeNotifier> {
87        self.notifier.as_ref()
88    }
89
90    /// Returns whether config sync is enabled.
91    #[must_use]
92    pub fn enabled(&self) -> bool {
93        self.notifier.is_some()
94    }
95
96    /// Converts this runtime into the reload-worker filter configuration.
97    #[must_use]
98    pub fn worker_config(&self) -> ConfigReloadWorkerConfig {
99        ConfigReloadWorkerConfig::new(self.namespace(), self.runtime_id())
100    }
101
102    /// Publishes a reload hint after a local config mutation.
103    ///
104    /// # Errors
105    ///
106    /// Returns [`ConfigError`] when publishing, subscription, reconciliation, or reload handling fails.
107    pub async fn publish_reload(
108        &self,
109        keys: impl IntoIterator<Item = impl Into<String>>,
110        source: ConfigNotificationSource,
111    ) -> Result<()> {
112        let Some(notifier) = self.notifier() else {
113            return Ok(());
114        };
115
116        notifier
117            .publish_reload(ConfigReloadMessage::new(
118                self.namespace(),
119                self.runtime_id(),
120                keys,
121                source,
122            ))
123            .await
124    }
125
126    /// Runs this runtime's reload subscription worker until shutdown.
127    ///
128    /// Disabled runtimes simply wait for shutdown, which lets callers spawn the
129    /// same task unconditionally if that is more convenient.
130    /// Failed reload hints are coalesced and retried without requiring another
131    /// notification. Callbacks must tolerate repeated reads and cancellation.
132    ///
133    /// # Errors
134    ///
135    /// Returns [`ConfigError`] when publishing, subscription, reconciliation, or reload handling fails.
136    pub async fn run_reload_subscription<F, Fut>(
137        &self,
138        shutdown: CancellationToken,
139        reload: F,
140    ) -> Result<()>
141    where
142        F: FnMut(ConfigReloadMessage) -> Fut,
143        Fut: Future<Output = Result<()>>,
144    {
145        let Some(notifier) = self.notifier().cloned() else {
146            shutdown.cancelled().await;
147            return Ok(());
148        };
149        run_config_reload_worker(notifier, self.worker_config(), shutdown, reload).await
150    }
151
152    /// Runs this runtime's reload subscription worker and reports observations.
153    ///
154    /// # Errors
155    ///
156    /// Returns [`ConfigError`] when publishing, subscription, reconciliation, or reload handling fails.
157    pub async fn run_reload_subscription_with_observer<F, Fut>(
158        &self,
159        shutdown: CancellationToken,
160        reload: F,
161        observer: Option<&dyn ConfigReloadObserver>,
162    ) -> Result<()>
163    where
164        F: FnMut(ConfigReloadMessage) -> Fut,
165        Fut: Future<Output = Result<()>>,
166    {
167        let Some(notifier) = self.notifier().cloned() else {
168            shutdown.cancelled().await;
169            return Ok(());
170        };
171        run_config_reload_worker_with_observer(
172            notifier,
173            self.worker_config(),
174            shutdown,
175            reload,
176            observer,
177        )
178        .await
179    }
180
181    /// Runs a reconnecting subscription with an authoritative reconcile callback.
182    ///
183    /// `reconcile` runs after each successful subscription. Product code should
184    /// reload its full snapshot and invalidate all derived configuration caches.
185    /// Failed reload or reconcile callbacks schedule a bounded reconcile retry,
186    /// even while the transport stays connected. Shutdown drops in-flight
187    /// callbacks, so reconciliation must be safe to cancel and repeat.
188    ///
189    /// # Errors
190    ///
191    /// Returns [`ConfigError`] when publishing, subscription, reconciliation, or reload handling fails.
192    pub async fn run_reload_subscription_with_reconcile<R, RFut, F, Fut>(
193        &self,
194        shutdown: CancellationToken,
195        reconcile: R,
196        reload: F,
197    ) -> Result<()>
198    where
199        R: FnMut() -> RFut,
200        RFut: Future<Output = Result<()>>,
201        F: FnMut(ConfigReloadMessage) -> Fut,
202        Fut: Future<Output = Result<()>>,
203    {
204        let Some(notifier) = self.notifier().cloned() else {
205            shutdown.cancelled().await;
206            return Ok(());
207        };
208        run_config_reload_supervisor(notifier, self.worker_config(), shutdown, reconcile, reload)
209            .await
210    }
211
212    /// Runs a reconnecting subscription and reports reload and connection observations.
213    ///
214    /// # Errors
215    ///
216    /// Returns [`ConfigError`] when publishing, subscription, reconciliation, or reload handling fails.
217    pub async fn run_reload_subscription_with_reconcile_and_observers<R, RFut, F, Fut>(
218        &self,
219        shutdown: CancellationToken,
220        reconcile: R,
221        reload: F,
222        reload_observer: Option<&dyn ConfigReloadObserver>,
223        connection_observer: Option<&dyn ConfigSyncConnectionObserver>,
224    ) -> Result<()>
225    where
226        R: FnMut() -> RFut,
227        RFut: Future<Output = Result<()>>,
228        F: FnMut(ConfigReloadMessage) -> Fut,
229        Fut: Future<Output = Result<()>>,
230    {
231        let Some(notifier) = self.notifier().cloned() else {
232            shutdown.cancelled().await;
233            return Ok(());
234        };
235        run_config_reload_supervisor_with_observers(
236            notifier,
237            self.worker_config(),
238            shutdown,
239            reconcile,
240            reload,
241            reload_observer,
242            connection_observer,
243        )
244        .await
245    }
246}
aster_forge_config/notification/runtime.rs

aster_forge_config/notification/
runtime.rs
