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

saluki_config/dynamic/
watcher.rs
