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

saluki_config/dynamic/
watcher.rs
