syncweb_core/daemon/
supervisor.rs

1use std::time::Duration;
2use std::{
3    collections::HashMap,
4    sync::{Arc, Mutex},
5};
6
7use iroh_docs::NamespaceId;
8use n0_future::StreamExt;
9use tokio::{
10    sync::{broadcast, oneshot},
11    time::sleep,
12};
13
14use crate::{
15    daemon::current_timestamp,
16    error::{Result, SyncwebError},
17    filter::FilterEngine,
18    sync::{ActiveSession, IntentControl, IntentHandle, SubscribeParams, SyncEngine, SyncEvent},
19};
20
21/// A synchronization intent together with the information needed to restart it.
22#[non_exhaustive]
23pub struct SupervisedIntent {
24    pub namespace: NamespaceId,
25    pub handle: Option<IntentHandle>,
26    pub retry_count: u32,
27    pub last_error: Option<String>,
28    pub last_started_at: Option<u64>,
29    session: Option<ActiveSession>,
30}
31
32impl std::fmt::Debug for SupervisedIntent {
33    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
34        formatter
35            .debug_struct("SupervisedIntent")
36            .field("namespace", &self.namespace)
37            .field("handle", &self.handle)
38            .field("retry_count", &self.retry_count)
39            .field("last_error", &self.last_error)
40            .field("last_started_at", &self.last_started_at)
41            .finish_non_exhaustive()
42    }
43}
44
45/// Starts synchronization intents and restarts failed intents with bounded
46/// exponential backoff.
47#[derive(Clone, Copy, Debug)]
48pub struct IntentSupervisor {
49    max_retries: u32,
50    backoff_base: Duration,
51    backoff_max: Duration,
52}
53
54/// Shared controls for intents currently owned by the daemon.
55pub type IntentControls = Arc<Mutex<HashMap<NamespaceId, IntentControl>>>;
56
57/// Internal startup options for a supervised daemon intent.
58pub struct SupervisionOptions {
59    controls: Option<IntentControls>,
60    filter: Option<FilterEngine>,
61    ready: Option<oneshot::Sender<()>>,
62}
63
64impl SupervisionOptions {
65    pub const fn with_ready(
66        controls: IntentControls,
67        filter: Option<FilterEngine>,
68        ready: oneshot::Sender<()>,
69    ) -> Self {
70        Self {
71            controls: Some(controls),
72            filter,
73            ready: Some(ready),
74        }
75    }
76}
77
78impl IntentSupervisor {
79    #[must_use]
80    pub const fn new(max_retries: u32, backoff_base: Duration, backoff_max: Duration) -> Self {
81        Self {
82            max_retries,
83            backoff_base,
84            backoff_max,
85        }
86    }
87
88    /// Start one continuous intent and record any startup failure.
89    pub async fn run_intent(
90        &self,
91        sync: &SyncEngine,
92        namespace: NamespaceId,
93        params: SubscribeParams,
94    ) -> SupervisedIntent {
95        self.run_intent_with_filter(sync, namespace, params, None).await
96    }
97
98    /// Start one filtered synchronization intent.
99    pub async fn run_intent_with_filter(
100        &self,
101        sync: &SyncEngine,
102        namespace: NamespaceId,
103        params: SubscribeParams,
104        filter_engine: Option<FilterEngine>,
105    ) -> SupervisedIntent {
106        let result = match filter_engine {
107            Some(engine) => {
108                sync.sync_with_filter(namespace, crate::sync::SessionMode::Continuous, params, engine)
109                    .await
110            }
111            None => sync.subscribe(namespace, params).await,
112        };
113        match result {
114            Ok(handle) => {
115                let session = ActiveSession::register(namespace, handle.cancel_sender());
116                SupervisedIntent {
117                    namespace,
118                    handle: Some(handle),
119                    retry_count: 0,
120                    last_error: None,
121                    last_started_at: Some(current_timestamp()),
122                    session: Some(session),
123                }
124            }
125            Err(error) => SupervisedIntent {
126                namespace,
127                handle: None,
128                retry_count: 0,
129                last_error: Some(error.to_string()),
130                last_started_at: None,
131                session: None,
132            },
133        }
134    }
135
136    /// Supervise one intent until shutdown or until its retry budget is spent.
137    ///
138    /// A normal `Finished` or `Cancelled` event ends supervision. A `Failed`
139    /// event, an unexpectedly closed stream, or a startup error consumes one
140    /// retry and restarts the intent after backoff.
141    ///
142    /// # Errors
143    ///
144    /// Returns an error if supervision infrastructure fails while waiting for
145    /// shutdown or retrying an intent.
146    pub async fn supervise(
147        &self,
148        sync: &SyncEngine,
149        namespace: NamespaceId,
150        params: SubscribeParams,
151        shutdown: broadcast::Receiver<()>,
152    ) -> Result<SupervisedIntent> {
153        self.supervise_inner(
154            sync,
155            namespace,
156            params,
157            shutdown,
158            SupervisionOptions {
159                controls: None,
160                filter: None,
161                ready: None,
162            },
163        )
164        .await
165    }
166
167    /// Supervise an intent while exposing pause/resume controls to its owner.
168    ///
169    /// # Errors
170    ///
171    /// Returns an error if the shared control registry is poisoned or the
172    /// intent supervision infrastructure cannot wait for shutdown.
173    pub async fn supervise_with_controls(
174        &self,
175        sync: &SyncEngine,
176        namespace: NamespaceId,
177        params: SubscribeParams,
178        shutdown: broadcast::Receiver<()>,
179        controls: IntentControls,
180        filter: Option<FilterEngine>,
181    ) -> Result<SupervisedIntent> {
182        self.supervise_inner(
183            sync,
184            namespace,
185            params,
186            shutdown,
187            SupervisionOptions {
188                controls: Some(controls),
189                filter,
190                ready: None,
191            },
192        )
193        .await
194    }
195
196    pub(crate) async fn supervise_with_controls_and_ready(
197        &self,
198        sync: &SyncEngine,
199        namespace: NamespaceId,
200        params: SubscribeParams,
201        shutdown: broadcast::Receiver<()>,
202        options: SupervisionOptions,
203    ) -> Result<SupervisedIntent> {
204        self.supervise_inner(sync, namespace, params, shutdown, options).await
205    }
206
207    async fn supervise_inner(
208        &self,
209        sync: &SyncEngine,
210        namespace: NamespaceId,
211        params: SubscribeParams,
212        mut shutdown: broadcast::Receiver<()>,
213        mut options: SupervisionOptions,
214    ) -> Result<SupervisedIntent> {
215        let mut supervised = self
216            .run_intent_with_filter(sync, namespace, params.clone(), options.filter.clone())
217            .await;
218        let mut ready = options.ready.take();
219
220        loop {
221            let Some(mut handle) = supervised.handle.take() else {
222                if let Some(sender) = ready.take() {
223                    let _ = sender.send(());
224                }
225                Self::remove_control(options.controls.as_ref(), namespace)?;
226                if supervised.retry_count >= self.max_retries {
227                    return Ok(supervised);
228                }
229                let retry_number = supervised.retry_count.saturating_add(1);
230                if !self
231                    .wait_for_retry(self.backoff_delay(retry_number), &mut shutdown)
232                    .await
233                {
234                    return Ok(supervised);
235                }
236                supervised = self
237                    .run_intent_with_filter(sync, namespace, params.clone(), options.filter.clone())
238                    .await;
239                supervised.retry_count = retry_number;
240                continue;
241            };
242
243            Self::insert_control(options.controls.as_ref(), namespace, handle.control())?;
244            if let Some(sender) = ready.take() {
245                let _ = sender.send(());
246            }
247            let session = supervised.session.take();
248            let event = loop {
249                tokio::select! {
250                    shutdown_result = shutdown.recv() => {
251                        match shutdown_result {
252                            Ok(()) | Err(broadcast::error::RecvError::Closed) => {
253                                if let Err(error) = handle.cancel() {
254                                    supervised.last_error = Some(format!("failed to cancel intent: {error}"));
255                                }
256                                Self::remove_control(options.controls.as_ref(), namespace)?;
257                                drop(session);
258                                supervised.handle = None;
259                                return Ok(supervised);
260                            }
261                            Err(broadcast::error::RecvError::Lagged(_)) => {}
262                        }
263                    }
264                    next_event = handle.next() => break next_event,
265                }
266            };
267
268            Self::remove_control(options.controls.as_ref(), namespace)?;
269            match event {
270                Some(SyncEvent::Started) => {
271                    supervised.last_error = None;
272                    supervised.last_started_at = Some(current_timestamp());
273                    supervised.handle = Some(handle);
274                    supervised.session = session;
275                }
276                Some(SyncEvent::Cancelled | SyncEvent::Finished) => {
277                    supervised.handle = None;
278                    drop(session);
279                    return Ok(supervised);
280                }
281                Some(SyncEvent::Failed(error)) => {
282                    supervised.last_error = Some(error);
283                    supervised.handle = None;
284                    drop(session);
285                }
286                Some(SyncEvent::Progress { .. } | SyncEvent::Stats(_) | SyncEvent::Paused | SyncEvent::Resumed) => {
287                    supervised.handle = Some(handle);
288                    supervised.session = session;
289                }
290                None => {
291                    supervised.last_error = Some("synchronization intent ended unexpectedly".to_owned());
292                    supervised.handle = None;
293                    drop(session);
294                }
295            }
296        }
297    }
298
299    fn insert_control(controls: Option<&IntentControls>, namespace: NamespaceId, control: IntentControl) -> Result<()> {
300        let Some(control_registry) = controls else {
301            return Ok(());
302        };
303        control_registry
304            .lock()
305            .map_err(|error| SyncwebError::operation("daemon intent control mutex is poisoned", error))?
306            .insert(namespace, control);
307        Ok(())
308    }
309
310    fn remove_control(controls: Option<&IntentControls>, namespace: NamespaceId) -> Result<()> {
311        let Some(control_registry) = controls else {
312            return Ok(());
313        };
314        control_registry
315            .lock()
316            .map_err(|error| SyncwebError::operation("daemon intent control mutex is poisoned", error))?
317            .remove(&namespace);
318        Ok(())
319    }
320
321    /// Return the delay for a one-based retry number.
322    #[must_use]
323    pub fn backoff_delay(&self, retry_number: u32) -> Duration {
324        const MIN_BACKOFF: Duration = Duration::from_millis(100);
325        if retry_number == 0 {
326            return Duration::ZERO;
327        }
328        let mut delay = self.backoff_base.max(MIN_BACKOFF);
329        for _ in 1..retry_number {
330            delay = match delay.checked_mul(2) {
331                Some(value) if value <= self.backoff_max => value,
332                _ => self.backoff_max,
333            };
334            if delay == self.backoff_max {
335                break;
336            }
337        }
338        delay.min(self.backoff_max)
339    }
340
341    async fn wait_for_retry(&self, delay: Duration, shutdown: &mut broadcast::Receiver<()>) -> bool {
342        tokio::select! {
343            () = sleep(delay) => true,
344            shutdown_result = shutdown.recv() => {
345                !matches!(shutdown_result, Ok(()) | Err(broadcast::error::RecvError::Closed))
346            }
347        }
348    }
349}
350
351#[cfg(test)]
352mod tests {
353    use super::*;
354
355    #[test]
356    fn backoff_delay_starts_at_base() {
357        let supervisor = IntentSupervisor::new(3, Duration::from_secs(1), Duration::from_mins(1));
358        assert_eq!(supervisor.backoff_delay(1), Duration::from_secs(1));
359    }
360
361    #[test]
362    fn backoff_delay_doubles() {
363        let supervisor = IntentSupervisor::new(3, Duration::from_secs(1), Duration::from_mins(1));
364        assert_eq!(supervisor.backoff_delay(2), Duration::from_secs(2));
365        assert_eq!(supervisor.backoff_delay(3), Duration::from_secs(4));
366    }
367
368    #[test]
369    fn backoff_delay_capped() {
370        let supervisor = IntentSupervisor::new(3, Duration::from_secs(1), Duration::from_secs(3));
371        assert_eq!(supervisor.backoff_delay(2), Duration::from_secs(2));
372        assert_eq!(supervisor.backoff_delay(3), Duration::from_secs(3));
373        assert_eq!(supervisor.backoff_delay(10), Duration::from_secs(3));
374    }
375
376    #[test]
377    fn backoff_delay_zero_retry() {
378        let supervisor = IntentSupervisor::new(3, Duration::from_secs(1), Duration::from_mins(1));
379        assert_eq!(supervisor.backoff_delay(0), Duration::ZERO);
380    }
381
382    #[test]
383    fn backoff_delay_zero_base() {
384        let supervisor = IntentSupervisor::new(3, Duration::ZERO, Duration::from_mins(1));
385        assert_eq!(supervisor.backoff_delay(1), Duration::from_millis(100));
386        assert_eq!(supervisor.backoff_delay(5), Duration::from_millis(1600));
387    }
388
389    #[test]
390    fn auto_backoff_delay_respects_max_retries() {
391        let supervisor = IntentSupervisor::new(5, Duration::from_secs(2), Duration::from_secs(20));
392        let delays: Vec<_> = (1..=8).map(|n| supervisor.backoff_delay(n)).collect();
393        let expected = [
394            Duration::from_secs(2),
395            Duration::from_secs(4),
396            Duration::from_secs(8),
397            Duration::from_secs(16),
398            Duration::from_secs(20),
399            Duration::from_secs(20),
400            Duration::from_secs(20),
401            Duration::from_secs(20),
402        ];
403        for (delay, want) in delays.iter().zip(expected.iter()) {
404            assert_eq!(delay, want);
405        }
406    }
407
408    #[test]
409    fn backoff_delay_starting_value_equals_base() {
410        let supervisor = IntentSupervisor::new(1, Duration::from_millis(500), Duration::from_secs(5));
411        assert_eq!(supervisor.backoff_delay(1), Duration::from_millis(500));
412    }
413}
syncweb_core/daemon/supervisor.rs

syncweb_core/daemon/
supervisor.rs
