saluki_config/dynamic/
watcher.rs

1//! A watcher for a specific configuration key.
2
3use std::future::pending as pending_forever;
4
5use serde::de::DeserializeOwned;
6use tokio::sync::broadcast;
7use tracing::warn;
8
9use crate::dynamic::ConfigChangeEvent;
10
11/// A watcher for a specific configuration key.
12///
13/// It filters [`ConfigChangeEvent`]s down to the
14/// requested key.
15///
16/// If dynamic configuration is disabled, [`changed`](Self::changed) will wait indefinitely and never yield.
17pub struct FieldUpdateWatcher {
18    /// The configuration key to watch for updates.
19    pub(crate) key: String,
20    /// Receiver of global configuration change events (None when dynamic is disabled).
21    pub(crate) rx: Option<broadcast::Receiver<ConfigChangeEvent>>,
22}
23
24impl FieldUpdateWatcher {
25    /// Waits until the watched key changes and returns a typed (old, new) tuple.
26    pub async fn changed<T>(&mut self) -> (Option<T>, Option<T>)
27    where
28        T: DeserializeOwned,
29    {
30        if self.rx.is_none() {
31            pending_forever::<()>().await;
32            unreachable!();
33        }
34
35        let rx = self.rx.as_mut().unwrap();
36        loop {
37            match rx.recv().await {
38                Ok(event) if event.key == self.key => {
39                    let old_ref = event.old_value.as_ref();
40                    let new_ref = event.new_value.as_ref();
41
42                    let old_t = old_ref.and_then(|ov| serde_json::from_value::<T>(ov.clone()).ok());
43                    let new_t = new_ref.and_then(|nv| serde_json::from_value::<T>(nv.clone()).ok());
44
45                    if new_t.is_some() || old_t.is_some() {
46                        return (old_t, new_t);
47                    }
48
49                    // If a new value was present but failed to deserialize, warn so we don't silently hide updates.
50                    if let Some(new_ref) = new_ref {
51                        warn!(
52                            key = %self.key,
53                            expected = %std::any::type_name::<T>(),
54                            actual = %get_type_name(new_ref),
55                            "FieldUpdateWatcher failed to deserialize new value. Skipping update."
56                        );
57                    }
58                }
59                // Ignore other key changes.
60                Ok(_) => continue,
61                Err(broadcast::error::RecvError::Lagged(_)) => {
62                    saluki_antithesis::unreachable!(
63                        "config filter update dropped (broadcast Lagged); live filtering may stay stale",
64                        { "key": self.key.to_string() }
65                    );
66                    warn!(
67                        "FieldUpdateWatcher dropped events for key: {}. Continuing to wait for the next event.",
68                        self.key
69                    );
70                    continue;
71                }
72                Err(broadcast::error::RecvError::Closed) => {
73                    // Keep pending forever to match "might never fire" semantics.
74                    pending_forever::<()>().await;
75                    unreachable!();
76                }
77            }
78        }
79    }
80}
81
82fn get_type_name(value: &serde_json::Value) -> &'static str {
83    match value {
84        serde_json::Value::Null => "null",
85        serde_json::Value::Bool(_) => "bool",
86        serde_json::Value::Number(_) => "number",
87        serde_json::Value::String(_) => "string",
88        serde_json::Value::Array(_) => "array",
89        serde_json::Value::Object(_) => "object",
90    }
91}
92
93#[cfg(test)]
94mod tests {
95    use crate::dynamic::event::ConfigUpdate;
96    use crate::ConfigurationLoader;
97
98    #[tokio::test]
99    async fn test_basic_field_update_watcher() {
100        let (cfg, sender) = ConfigurationLoader::for_tests(
101            Some(serde_json::json!({ "foobar": { "a": false, "b": "c" } })),
102            None,
103            true,
104        )
105        .await;
106        let sender = sender.expect("sender should exist");
107
108        sender
109            .send(ConfigUpdate::Snapshot(serde_json::json!({})))
110            .await
111            .unwrap();
112        cfg.ready().await;
113
114        let mut watcher = cfg.watch_for_updates("watched_key");
115
116        sender
117            .send(ConfigUpdate::Partial {
118                key: "watched_key".to_string(),
119                value: serde_json::json!("hello"),
120            })
121            .await
122            .unwrap();
123
124        let (old, new) = tokio::time::timeout(std::time::Duration::from_secs(2), watcher.changed::<String>())
125            .await
126            .expect("timed out waiting for watched_key update");
127
128        assert_eq!(old, None);
129        assert_eq!(new, Some("hello".to_string()));
130    }
131
132    #[tokio::test]
133    async fn test_field_update_watcher_nested_key() {
134        let (cfg, sender) = ConfigurationLoader::for_tests(
135            Some(serde_json::json!({ "foobar": { "a": false, "b": "c" } })),
136            None,
137            true,
138        )
139        .await;
140        let sender = sender.expect("sender should exist");
141
142        sender
143            .send(ConfigUpdate::Snapshot(serde_json::json!({})))
144            .await
145            .unwrap();
146        cfg.ready().await;
147
148        let mut watcher = cfg.watch_for_updates("foobar.a");
149
150        // Update nested value via dotted path
151        sender
152            .send(ConfigUpdate::Partial {
153                key: "foobar.a".to_string(),
154                value: serde_json::json!(true),
155            })
156            .await
157            .unwrap();
158
159        let (old, new) = tokio::time::timeout(std::time::Duration::from_secs(2), watcher.changed::<bool>())
160            .await
161            .expect("timed out waiting for foobar.a update");
162
163        assert_eq!(old, Some(false));
164        assert_eq!(new, Some(true));
165        assert!(cfg.get_typed::<bool>("foobar.a").unwrap());
166
167        // Existing nested key not updated is still present
168        assert_eq!(cfg.get_typed::<String>("foobar.b").unwrap(), "c");
169    }
170
171    #[tokio::test]
172    async fn test_field_update_watcher_parent_update() {
173        let (cfg, sender) = ConfigurationLoader::for_tests(
174            Some(serde_json::json!({ "foobar": { "a": false, "b": "c" } })),
175            None,
176            true,
177        )
178        .await;
179        let sender = sender.expect("sender should exist");
180
181        sender
182            .send(ConfigUpdate::Snapshot(serde_json::json!({})))
183            .await
184            .unwrap();
185        cfg.ready().await;
186
187        let mut watcher = cfg.watch_for_updates("foobar.a");
188
189        // Update parent object directly
190        sender
191            .send(ConfigUpdate::Partial {
192                key: "foobar".to_string(),
193                value: serde_json::json!({ "a": true }),
194            })
195            .await
196            .unwrap();
197
198        let (old, new) = tokio::time::timeout(std::time::Duration::from_secs(2), watcher.changed::<bool>())
199            .await
200            .expect("timed out waiting for foobar.a update");
201
202        assert_eq!(old, Some(false));
203        assert_eq!(new, Some(true));
204        assert!(cfg.get_typed::<bool>("foobar.a").unwrap());
205
206        // Existing nested key not updated is still present
207        assert_eq!(cfg.get_typed::<String>("foobar.b").unwrap(), "c");
208    }
209}
saluki_config/dynamic/watcher.rs

saluki_config/dynamic/
watcher.rs
