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 std::time::Duration;
96
97    use serde_json::json;
98    use tokio::sync::broadcast;
99
100    use super::FieldUpdateWatcher;
101    use crate::dynamic::event::{ConfigChangeEvent, ConfigSetting, ConfigUpdate};
102    use crate::ConfigurationLoader;
103
104    fn change_event(key: &str, new_value: serde_json::Value) -> ConfigChangeEvent {
105        ConfigChangeEvent {
106            key: key.to_string(),
107            old_value: None,
108            new_value: Some(new_value),
109        }
110    }
111
112    fn watcher_over(key: &str, capacity: usize) -> (broadcast::Sender<ConfigChangeEvent>, FieldUpdateWatcher) {
113        let (tx, rx) = broadcast::channel(capacity);
114        (
115            tx,
116            FieldUpdateWatcher {
117                key: key.to_string(),
118                rx: Some(rx),
119            },
120        )
121    }
122
123    #[tokio::test]
124    async fn changed_returns_partial_update_for_watched_key() {
125        let (cfg, sender) = ConfigurationLoader::for_tests(
126            Some(serde_json::json!({ "foobar": { "a": false, "b": "c" } })),
127            None,
128            true,
129        )
130        .await;
131        let sender = sender.expect("sender should exist");
132
133        sender.send(ConfigUpdate::snapshot([])).await.unwrap();
134        cfg.ready().await;
135
136        let mut watcher = cfg.watch_for_updates("watched_key");
137
138        sender
139            .send(ConfigUpdate::Partial(ConfigSetting::explicit(
140                "watched_key",
141                serde_json::json!("hello"),
142            )))
143            .await
144            .unwrap();
145
146        let (old, new) = tokio::time::timeout(std::time::Duration::from_secs(2), watcher.changed::<String>())
147            .await
148            .expect("timed out waiting for watched_key update");
149
150        assert_eq!(old, None);
151        assert_eq!(new, Some("hello".to_string()));
152    }
153
154    #[tokio::test]
155    async fn changed_returns_nested_key_update() {
156        let (cfg, sender) = ConfigurationLoader::for_tests(
157            Some(serde_json::json!({ "foobar": { "a": false, "b": "c" } })),
158            None,
159            true,
160        )
161        .await;
162        let sender = sender.expect("sender should exist");
163
164        sender.send(ConfigUpdate::snapshot([])).await.unwrap();
165        cfg.ready().await;
166
167        let mut watcher = cfg.watch_for_updates("foobar.a");
168
169        // Update nested value via dotted path
170        sender
171            .send(ConfigUpdate::Partial(ConfigSetting::explicit(
172                "foobar.a",
173                serde_json::json!(true),
174            )))
175            .await
176            .unwrap();
177
178        let (old, new) = tokio::time::timeout(std::time::Duration::from_secs(2), watcher.changed::<bool>())
179            .await
180            .expect("timed out waiting for foobar.a update");
181
182        assert_eq!(old, Some(false));
183        assert_eq!(new, Some(true));
184        assert!(cfg.get_typed::<bool>("foobar.a").unwrap());
185
186        // Existing nested key not updated is still present
187        assert_eq!(cfg.get_typed::<String>("foobar.b").unwrap(), "c");
188    }
189
190    #[tokio::test]
191    async fn changed_returns_update_when_parent_object_changes() {
192        let (cfg, sender) = ConfigurationLoader::for_tests(
193            Some(serde_json::json!({ "foobar": { "a": false, "b": "c" } })),
194            None,
195            true,
196        )
197        .await;
198        let sender = sender.expect("sender should exist");
199
200        sender.send(ConfigUpdate::snapshot([])).await.unwrap();
201        cfg.ready().await;
202
203        let mut watcher = cfg.watch_for_updates("foobar.a");
204
205        // Update parent object directly
206        sender
207            .send(ConfigUpdate::Partial(ConfigSetting::explicit(
208                "foobar",
209                serde_json::json!({ "a": true }),
210            )))
211            .await
212            .unwrap();
213
214        let (old, new) = tokio::time::timeout(std::time::Duration::from_secs(2), watcher.changed::<bool>())
215            .await
216            .expect("timed out waiting for foobar.a update");
217
218        assert_eq!(old, Some(false));
219        assert_eq!(new, Some(true));
220        assert!(cfg.get_typed::<bool>("foobar.a").unwrap());
221
222        // Existing nested key not updated is still present
223        assert_eq!(cfg.get_typed::<String>("foobar.b").unwrap(), "c");
224    }
225
226    #[tokio::test]
227    async fn changed_waits_forever_when_dynamic_configuration_disabled() {
228        // With no receiver (dynamic configuration disabled), `changed` must never resolve.
229        let mut watcher = FieldUpdateWatcher {
230            key: "watched_key".to_string(),
231            rx: None,
232        };
233
234        let result = tokio::time::timeout(Duration::from_millis(250), watcher.changed::<String>()).await;
235        assert!(
236            result.is_err(),
237            "changed() must stay pending when dynamic config is disabled"
238        );
239    }
240
241    #[tokio::test]
242    async fn changed_waits_forever_after_channel_closes() {
243        // Once the broadcast sender is dropped, `recv` yields `Closed`; the watcher then matches "might never fire"
244        // semantics by staying pending rather than returning.
245        let (tx, mut watcher) = watcher_over("watched_key", 4);
246        drop(tx);
247
248        let result = tokio::time::timeout(Duration::from_millis(250), watcher.changed::<String>()).await;
249        assert!(result.is_err(), "changed() must stay pending after the channel closes");
250    }
251
252    #[tokio::test]
253    async fn changed_skips_lagged_errors_and_returns_next_event() {
254        // A capacity-1 channel with three unread sends forces the receiver to lag: the first `recv` returns
255        // `Lagged`, which the watcher skips before returning the most recent retained event.
256        let (tx, mut watcher) = watcher_over("watched_key", 1);
257        tx.send(change_event("watched_key", json!("first"))).unwrap();
258        tx.send(change_event("watched_key", json!("second"))).unwrap();
259        tx.send(change_event("watched_key", json!("third"))).unwrap();
260
261        let (old, new) = tokio::time::timeout(Duration::from_secs(2), watcher.changed::<String>())
262            .await
263            .expect("changed() should resolve after skipping the lagged error");
264
265        assert_eq!(old, None);
266        assert_eq!(new, Some("third".to_string()));
267    }
268
269    #[tokio::test]
270    async fn changed_skips_events_that_fail_to_deserialize() {
271        // A value that can't be deserialized into the requested type is skipped (neither old nor new deserializes),
272        // and the watcher keeps waiting until a well-typed value for the same key arrives.
273        let (tx, mut watcher) = watcher_over("watched_key", 8);
274        tx.send(change_event("watched_key", json!("not-a-number"))).unwrap();
275        tx.send(change_event("watched_key", json!(42))).unwrap();
276
277        let (old, new) = tokio::time::timeout(Duration::from_secs(2), watcher.changed::<u32>())
278            .await
279            .expect("changed() should resolve once a well-typed value arrives");
280
281        assert_eq!(old, None);
282        assert_eq!(new, Some(42));
283    }
284}
saluki_config/dynamic/watcher.rs

saluki_config/dynamic/
watcher.rs
