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

syncweb_core/daemon/
supervisor.rs
