aster_forge_config/notification/
message.rs

1use serde::{Deserialize, Serialize};
2use std::future::Future;
3
4use crate::Result;
5
6/// Source that emitted a configuration reload notification.
7#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
8#[serde(rename_all = "snake_case")]
9pub enum ConfigNotificationSource {
10    /// Notification was emitted by an API mutation.
11    Api,
12    /// Notification was emitted by a CLI operation.
13    Cli,
14    /// Notification was emitted by a startup/bootstrap path.
15    Startup,
16    /// Notification was emitted by an unspecified or product-specific source.
17    Other(String),
18}
19
20/// Notification payload published after configuration changes.
21#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
22pub struct ConfigReloadMessage {
23    /// Product or service namespace, for example `aster_yggdrasil`.
24    pub namespace: String,
25    /// Runtime instance ID that emitted the message. Receivers can use this to ignore
26    /// their own message after already applying the local change.
27    pub origin_runtime_id: String,
28    /// Changed keys. Empty means receivers should reload all runtime config.
29    pub keys: Vec<String>,
30    /// Source of the change.
31    pub source: ConfigNotificationSource,
32}
33
34impl ConfigReloadMessage {
35    /// Creates a reload message and sorts/deduplicates keys.
36    pub fn new(
37        namespace: impl Into<String>,
38        origin_runtime_id: impl Into<String>,
39        keys: impl IntoIterator<Item = impl Into<String>>,
40        source: ConfigNotificationSource,
41    ) -> Self {
42        let mut keys = keys.into_iter().map(Into::into).collect::<Vec<_>>();
43        keys.sort();
44        keys.dedup();
45        Self {
46            namespace: namespace.into(),
47            origin_runtime_id: origin_runtime_id.into(),
48            keys,
49            source,
50        }
51    }
52
53    /// Serializes the message for transport.
54    pub fn encode(&self) -> Result<String> {
55        serde_json::to_string(self).map_err(Into::into)
56    }
57
58    /// Decodes a transport payload.
59    pub fn decode(payload: &str) -> Result<Self> {
60        serde_json::from_str(payload).map_err(Into::into)
61    }
62}
63
64/// Local event delivered by a notifier.
65#[derive(Debug, Clone, PartialEq, Eq)]
66pub enum ConfigChangeEvent {
67    /// Receivers should reload from storage.
68    Reload(ConfigReloadMessage),
69}
70
71impl ConfigChangeEvent {
72    /// Returns the reload message carried by this event.
73    pub const fn reload_message(&self) -> &ConfigReloadMessage {
74        match self {
75            Self::Reload(message) => message,
76        }
77    }
78}
79
80/// Result of handling one reload notification.
81#[derive(Debug, Clone, Copy, PartialEq, Eq)]
82pub enum ConfigReloadDecision {
83    /// The notification matched this process and triggered a reload.
84    Reloaded,
85    /// The notification belongs to another product namespace.
86    IgnoredNamespace,
87    /// The notification came from this process and should not be replayed.
88    IgnoredOrigin,
89}
90
91impl ConfigReloadDecision {
92    /// Returns the stable metrics label for this decision.
93    pub const fn as_label(self) -> &'static str {
94        match self {
95            Self::Reloaded => "reloaded",
96            Self::IgnoredNamespace => "ignored_namespace",
97            Self::IgnoredOrigin => "ignored_origin",
98        }
99    }
100}
101
102/// Runtime reload worker configuration.
103#[derive(Debug, Clone, PartialEq, Eq)]
104pub struct ConfigReloadWorkerConfig {
105    /// Product or service namespace accepted by this worker.
106    pub namespace: String,
107    /// Runtime instance ID for the current process.
108    pub runtime_id: String,
109}
110
111impl ConfigReloadWorkerConfig {
112    /// Creates a worker config.
113    pub fn new(namespace: impl Into<String>, runtime_id: impl Into<String>) -> Self {
114        Self {
115            namespace: namespace.into(),
116            runtime_id: runtime_id.into(),
117        }
118    }
119
120    /// Returns whether a message belongs to this worker namespace.
121    pub fn accepts_namespace(&self, message: &ConfigReloadMessage) -> bool {
122        message.namespace == self.namespace
123    }
124
125    /// Returns whether a message was emitted by this process.
126    pub fn is_local_origin(&self, message: &ConfigReloadMessage) -> bool {
127        message.origin_runtime_id == self.runtime_id
128    }
129}
130
131/// Decodes one transport payload into a config reload event.
132///
133/// Transport adapters should use this helper before forwarding data into the common notifier path.
134/// Malformed payloads are returned as errors so listeners can log and continue instead of ending the
135/// subscription loop.
136pub fn decode_config_reload_transport_payload(payload: &str) -> Result<ConfigChangeEvent> {
137    ConfigReloadMessage::decode(payload).map(ConfigChangeEvent::Reload)
138}
139
140/// Handles one reload notification by filtering namespace/origin and invoking `reload`.
141pub async fn handle_config_reload_notification<F, Fut>(
142    config: &ConfigReloadWorkerConfig,
143    message: ConfigReloadMessage,
144    reload: F,
145) -> Result<ConfigReloadDecision>
146where
147    F: FnOnce(ConfigReloadMessage) -> Fut,
148    Fut: Future<Output = Result<()>>,
149{
150    if !config.accepts_namespace(&message) {
151        return Ok(ConfigReloadDecision::IgnoredNamespace);
152    }
153    if config.is_local_origin(&message) {
154        return Ok(ConfigReloadDecision::IgnoredOrigin);
155    }
156
157    reload(message).await?;
158    Ok(ConfigReloadDecision::Reloaded)
159}
aster_forge_config/notification/message.rs

aster_forge_config/notification/
message.rs
