syncweb_core/sync/
engine.rs

1use std::{
2    collections::{HashMap, HashSet},
3    path::Path,
4    pin::Pin,
5    time::{Duration, Instant},
6};
7
8use iroh_blobs::Hash;
9use iroh_docs::{NamespaceId, engine::LiveEvent};
10use n0_future::StreamExt;
11use tokio::sync::mpsc;
12
13use crate::{
14    error::Result,
15    filter::{FilterAction, FilterEngine, FilterEntry},
16    folder::FolderManager,
17    node::discovery::TopicTracker,
18    node::iroh_node::IrohNode,
19    node::{blob_store::BlobStore, docs_engine::DocsEngine},
20    storage::node_db::NodeDatabase,
21};
22
23use super::checkpoint::SyncCheckpoint;
24use super::{FetchStrategy, IntentHandle, SessionMode, SubscribeParams, SyncCommand, SyncEvent};
25
26type IntentSetup = (
27    IntentState,
28    Pin<Box<tokio::time::Sleep>>,
29    Pin<Box<tokio::time::Sleep>>,
30    Option<SyncCheckpoint>,
31    HashSet<Vec<u8>>,
32);
33
34/// Current aggregate statistics for a synchronization intent.
35#[derive(Clone, Debug, Default, Eq, PartialEq)]
36#[non_exhaustive]
37pub struct TransferStats {
38    pub bytes_transferred: u64,
39    pub bytes_total: Option<u64>,
40    pub bytes_per_second: u64,
41    pub peer_count: usize,
42    pub eta: Option<Duration>,
43}
44
45impl TransferStats {
46    #[must_use]
47    pub fn from_progress(
48        bytes_transferred: u64,
49        bytes_total: Option<u64>,
50        elapsed: Duration,
51        peer_count: usize,
52    ) -> Self {
53        let elapsed_millis = u64::try_from(elapsed.as_millis()).unwrap_or(u64::MAX);
54        let bytes_per_second = bytes_transferred
55            .saturating_mul(1_000)
56            .checked_div(elapsed_millis)
57            .unwrap_or(0);
58        let eta = bytes_total.and_then(|total| {
59            let remaining = total.saturating_sub(bytes_transferred);
60            remaining.checked_div(bytes_per_second).map(Duration::from_secs)
61        });
62        Self {
63            bytes_transferred,
64            bytes_total,
65            bytes_per_second,
66            peer_count,
67            eta,
68        }
69    }
70}
71
72/// Coordinates document reconciliation and blob availability for a folder.
73#[derive(Clone)]
74pub struct SyncEngine {
75    folder_manager: FolderManager,
76    blob_store: BlobStore,
77    docs_engine: DocsEngine,
78    node_db: Option<NodeDatabase>,
79    topic_tracker: Option<TopicTracker>,
80}
81
82impl SyncEngine {
83    #[must_use]
84    pub const fn new(
85        folder_manager: FolderManager,
86        blob_store: BlobStore,
87        docs_engine: DocsEngine,
88        topic_tracker: Option<TopicTracker>,
89    ) -> Self {
90        Self {
91            folder_manager,
92            blob_store,
93            docs_engine,
94            node_db: None,
95            topic_tracker,
96        }
97    }
98
99    /// Set the node database for checkpoint tracking.
100    #[must_use]
101    pub fn with_node_db(mut self, node_db: NodeDatabase) -> Self {
102        self.node_db = Some(node_db);
103        self
104    }
105
106    #[must_use]
107    pub fn from_node(node: &IrohNode, folder_manager: FolderManager) -> Self {
108        Self::new(
109            folder_manager,
110            node.blob_store().clone(),
111            node.docs_engine().clone(),
112            Some(node.topic_tracker().clone()),
113        )
114    }
115
116    /// Start reconciling a managed folder.
117    ///
118    /// Iroh Docs performs network reconciliation itself. This intent provides
119    /// lifecycle, control, and aggregate transfer events around that process.
120    ///
121    /// # Errors
122    ///
123    /// Returns an error if the folder is not managed locally.
124    pub async fn sync(&self, folder_id: NamespaceId, mode: SessionMode) -> Result<IntentHandle> {
125        self.sync_with_params(folder_id, mode, SubscribeParams::default()).await
126    }
127
128    /// Start a synchronization intent with subscription filtering.
129    ///
130    /// # Errors
131    ///
132    /// Returns an error if the folder is not managed locally or live sync
133    /// cannot be started.
134    pub async fn sync_with_params(
135        &self,
136        folder_id: NamespaceId,
137        mode: SessionMode,
138        params: SubscribeParams,
139    ) -> Result<IntentHandle> {
140        self.start(folder_id, mode, params, None).await
141    }
142
143    /// Subscribe to live changes in a folder.
144    ///
145    /// # Errors
146    ///
147    /// Returns an error if the folder is not managed locally or live sync
148    /// cannot be started.
149    pub async fn subscribe(&self, folder_id: NamespaceId, params: SubscribeParams) -> Result<IntentHandle> {
150        self.sync_with_params(folder_id, SessionMode::Continuous, params).await
151    }
152
153    /// Fetch folder content using a full or filtered reconciliation strategy.
154    ///
155    /// Peer-count filters are applied by callers that have availability
156    /// observations. The document subscription still carries path, size, and
157    /// count limits so the remote reconciliation remains bounded.
158    ///
159    /// # Errors
160    ///
161    /// Returns an error if the folder is not managed locally or synchronization
162    /// cannot be started.
163    pub async fn fetch(&self, folder_id: NamespaceId, strategy: FetchStrategy) -> Result<IntentHandle> {
164        let params = Self::fetch_params(strategy);
165        self.sync_with_params(folder_id, SessionMode::ReconcileOnce, params)
166            .await
167    }
168
169    /// Start a synchronization intent and filter emitted entries.
170    ///
171    /// The document reconciliation remains content-addressed and is handled
172    /// by iroh-docs; the filter controls which entries are reported to the
173    /// automatic synchronization consumer.
174    ///
175    /// # Errors
176    ///
177    /// Returns an error if the folder is not managed locally or live sync
178    /// cannot be started.
179    pub async fn sync_with_filter(
180        &self,
181        folder_id: NamespaceId,
182        mode: SessionMode,
183        params: SubscribeParams,
184        filter: FilterEngine,
185    ) -> Result<IntentHandle> {
186        self.start(folder_id, mode, params, Some(filter)).await
187    }
188
189    fn fetch_params(strategy: FetchStrategy) -> SubscribeParams {
190        let FetchStrategy::Filter(filter) = strategy else {
191            return SubscribeParams::default();
192        };
193        let mut params = SubscribeParams::default();
194        if let Some(paths) = filter.paths
195            && let Some(path) = paths.first()
196        {
197            params = params.with_area(super::AreaFilter::Prefix(path.clone()));
198        }
199        if filter.max_size.is_some() || filter.max_count.is_some() {
200            let area = params.area_filter.clone().unwrap_or(super::AreaFilter::All);
201            let limits = super::AreaOfInterest::with_limits(
202                area,
203                filter.max_size.unwrap_or(0),
204                u64::try_from(filter.max_count.unwrap_or(0)).unwrap_or(u64::MAX),
205            );
206            params = params.with_limits(limits);
207        }
208        params
209    }
210
211    async fn start(
212        &self,
213        folder_id: NamespaceId,
214        mode: SessionMode,
215        params: SubscribeParams,
216        filter: Option<FilterEngine>,
217    ) -> Result<IntentHandle> {
218        let folder = self.folder_manager.get(folder_id).await?;
219        let live_events = self.docs_engine.watch_and_sync(folder.doc()).await?;
220        if let Some(topic_tracker) = &self.topic_tracker {
221            match topic_tracker.find_peers(folder_id).await {
222                Ok(peers) if !peers.is_empty() => {
223                    tracing::debug!(%folder_id, peer_count = peers.len(), "discovered peers via topic tracker");
224                }
225                Ok(_) => {}
226                Err(error) => {
227                    tracing::debug!(%folder_id, %error, "failed to discover peers via topic tracker");
228                }
229            }
230        }
231        let (events, commands, handle) = IntentHandle::channel();
232        let node_db = self.node_db.clone();
233        tokio::spawn(run_intent(
234            folder,
235            IntentConfig { mode, params, filter },
236            self.blob_store.clone(),
237            live_events,
238            events,
239            commands,
240            node_db,
241        ));
242        Ok(handle)
243    }
244}
245
246struct IntentConfig {
247    mode: SessionMode,
248    params: SubscribeParams,
249    filter: Option<FilterEngine>,
250}
251
252async fn run_intent(
253    folder: crate::folder::SyncwebFolder,
254    config: IntentConfig,
255    blob_store: BlobStore,
256    mut live_events: impl n0_future::Stream<Item = Result<LiveEvent>> + Send + Unpin + 'static,
257    events: mpsc::UnboundedSender<SyncEvent>,
258    mut commands: mpsc::UnboundedReceiver<SyncCommand>,
259    node_db: Option<NodeDatabase>,
260) {
261    const SETTLE: Duration = Duration::from_secs(2);
262    const FORCED_QUIESCE: Duration = Duration::from_secs(10);
263
264    let started = Instant::now();
265    if !send_initial_events(&events, started) {
266        return;
267    }
268
269    let (mut state, mut quiesce, mut forced_quiesce, checkpoint, completed_keys) =
270        setup_intent_checkpoint(&folder, node_db.as_ref());
271
272    loop {
273        tokio::select! {
274            received_command = commands.recv() => {
275                let Some(command) = received_command else {
276                    let _result = folder.doc().leave().await;
277                    finalize_checkpoint(checkpoint.as_ref(), &state, &events);
278                    return;
279                };
280                if !handle_command(&folder, command, &mut state.paused, &events).await {
281                    finalize_checkpoint(checkpoint.as_ref(), &state, &events);
282                    return;
283                }
284            }
285            () = &mut quiesce, if !config.mode.is_continuous() && state.sync_finished && state.sizes.is_empty() => {
286                finalize_checkpoint(checkpoint.as_ref(), &state, &events);
287                return;
288            }
289            () = &mut forced_quiesce, if !config.mode.is_continuous() && state.sync_finished => {
290                let pending = state.sizes.len();
291                if pending > 0 {
292                    tracing::warn!(%pending, "forced quiesce: content for some blobs never arrived");
293                }
294                finalize_checkpoint(checkpoint.as_ref(), &state, &events);
295                return;
296            }
297            next_event = live_events.next(), if !state.paused => {
298                let Some(event_result) = next_event else {
299                    let _result = events.send(SyncEvent::Finished);
300                    finalize_checkpoint(checkpoint.as_ref(), &state, &events);
301                    return;
302                };
303                let live_event = match event_result {
304                    Ok(event) => event,
305                    Err(error) => {
306                        let _result = events.send(SyncEvent::Failed(error.to_string()));
307                        finalize_checkpoint(checkpoint.as_ref(), &state, &events);
308                        return;
309                    }
310                };
311                if !include_event(&config.params, &live_event) {
312                    continue;
313                }
314                let entry_key = match &live_event {
315                    LiveEvent::InsertLocal { entry } | LiveEvent::InsertRemote { entry, .. } => {
316                        Some(entry.key().to_vec())
317                    }
318                    LiveEvent::ContentReady { .. }
319                    | LiveEvent::PendingContentReady
320                    | LiveEvent::NeighborUp(_)
321                    | LiveEvent::NeighborDown(_)
322                    | LiveEvent::SyncFinished(_) => None,
323                };
324                if let Some(key) = &entry_key && completed_keys.contains(key) {
325                    continue;
326                }
327                if let LiveEvent::ContentReady { hash } = &live_event {
328                    match blob_store.has(*hash).await {
329                        Ok(true) => {}
330                        Ok(false) => {
331                            let msg = format!("content-ready blob {hash} is missing locally");
332                            let _result = events.send(SyncEvent::Failed(msg));
333                            finalize_checkpoint(checkpoint.as_ref(), &state, &events);
334                            return;
335                        }
336                        Err(error) => {
337                            let _result = events.send(SyncEvent::Failed(error.to_string()));
338                            finalize_checkpoint(checkpoint.as_ref(), &state, &events);
339                            return;
340                        }
341                    }
342                    if let Some(cmd) = throttle_bandwidth(hash, &state, &config, &mut commands).await {
343                        let _ = handle_command(&folder, cmd, &mut state.paused, &events).await;
344                        finalize_checkpoint(checkpoint.as_ref(), &state, &events);
345                        return;
346                    }
347                    if let Some(ref cp) = checkpoint {
348                        let _ = cp.mark_completed(hash.as_bytes(), *hash, state.sizes.get(hash).copied().unwrap_or(0));
349                    }
350                }
351                if let Err(error) = state.apply(live_event, &config.params, config.filter.as_ref()) {
352                    let _result = events.send(SyncEvent::Failed(error));
353                    finalize_checkpoint(checkpoint.as_ref(), &state, &events);
354                    return;
355                }
356                if !config.mode.is_continuous() && state.sync_finished {
357                    let now = tokio::time::Instant::now();
358                    quiesce.as_mut().reset(now.checked_add(SETTLE).unwrap_or(now));
359                    forced_quiesce.as_mut().reset(now.checked_add(FORCED_QUIESCE).unwrap_or(now));
360                }
361                if !send_progress(&events, &state, started.elapsed()) {
362                    finalize_checkpoint(checkpoint.as_ref(), &state, &events);
363                    return;
364                }
365            }
366        }
367    }
368}
369
370async fn throttle_bandwidth(
371    hash: &iroh_blobs::Hash,
372    state: &IntentState,
373    config: &IntentConfig,
374    commands: &mut mpsc::UnboundedReceiver<SyncCommand>,
375) -> Option<SyncCommand> {
376    if let Some(rate) = config.params.bandwidth.as_ref().and_then(|limits| limits.max_download)
377        && rate > 0
378    {
379        let bytes = state.sizes.get(hash).copied().unwrap_or(0);
380        let milliseconds = bytes.saturating_mul(1_000).div_ceil(rate);
381        if milliseconds > 0 {
382            tokio::select! {
383                () = tokio::time::sleep(Duration::from_millis(milliseconds)) => {}
384                received_command = commands.recv() => {
385                    return received_command;
386                }
387            }
388        }
389    }
390    None
391}
392
393fn setup_intent_checkpoint(folder: &crate::folder::SyncwebFolder, node_db: Option<&NodeDatabase>) -> IntentSetup {
394    let namespace_id = folder.namespace_id();
395    let checkpoint = node_db.and_then(|db| match SyncCheckpoint::resume(db, namespace_id) {
396        Ok(Some(cp)) => {
397            tracing::info!(%namespace_id, session = %cp.session_id(), "resuming sync checkpoint");
398            Some(cp)
399        }
400        Ok(None) => SyncCheckpoint::create_session(db, namespace_id)
401            .map_err(|e| tracing::warn!(%namespace_id, %e, "failed to create checkpoint"))
402            .ok(),
403        Err(err) => {
404            tracing::warn!(%namespace_id, %err, "failed to resume checkpoint");
405            SyncCheckpoint::create_session(db, namespace_id)
406                .map_err(|e2| tracing::warn!(%namespace_id, %e2, "failed to create checkpoint"))
407                .ok()
408        }
409    });
410    let completed_keys: HashSet<Vec<u8>> = checkpoint
411        .as_ref()
412        .and_then(|cp| cp.completed_keys().ok())
413        .unwrap_or_default();
414
415    let state = IntentState {
416        completed: u64::try_from(completed_keys.len()).unwrap_or(u64::MAX),
417        ..IntentState::default()
418    };
419    let quiesce = Box::pin(tokio::time::sleep(Duration::MAX));
420    let forced_quiesce = Box::pin(tokio::time::sleep(Duration::MAX));
421    (state, quiesce, forced_quiesce, checkpoint, completed_keys)
422}
423
424fn finalize_checkpoint(
425    checkpoint: Option<&SyncCheckpoint>,
426    state: &IntentState,
427    events: &mpsc::UnboundedSender<SyncEvent>,
428) {
429    if let Some(cp) = checkpoint {
430        if state.completed > 0 {
431            let _ = cp.complete();
432        } else {
433            let _ = cp.leave_unfinished();
434        }
435    }
436    let _ = events.send(SyncEvent::Finished);
437}
438
439fn send_initial_events(events: &mpsc::UnboundedSender<SyncEvent>, started: Instant) -> bool {
440    events.send(SyncEvent::Started).is_ok()
441        && events
442            .send(SyncEvent::Progress {
443                completed: 0,
444                total: Some(0),
445            })
446            .is_ok()
447        && events
448            .send(SyncEvent::Stats(TransferStats::from_progress(
449                0,
450                Some(0),
451                started.elapsed(),
452                0,
453            )))
454            .is_ok()
455}
456
457fn send_progress(events: &mpsc::UnboundedSender<SyncEvent>, state: &IntentState, elapsed: Duration) -> bool {
458    events
459        .send(SyncEvent::Progress {
460            completed: state.completed,
461            total: None,
462        })
463        .is_ok()
464        && events
465            .send(SyncEvent::Stats(TransferStats::from_progress(
466                state.transferred,
467                None,
468                elapsed,
469                state.peer_count,
470            )))
471            .is_ok()
472}
473
474#[derive(Default)]
475struct IntentState {
476    paused: bool,
477    completed: u64,
478    transferred: u64,
479    peer_count: usize,
480    sizes: HashMap<Hash, u64>,
481    area_count: u64,
482    area_size: u64,
483    sync_finished: bool,
484}
485
486impl IntentState {
487    fn apply(
488        &mut self,
489        event: LiveEvent,
490        params: &SubscribeParams,
491        filter: Option<&FilterEngine>,
492    ) -> std::result::Result<(), String> {
493        match event {
494            LiveEvent::NeighborUp(_) => self.peer_count = self.peer_count.saturating_add(1),
495            LiveEvent::NeighborDown(_) => self.peer_count = self.peer_count.saturating_sub(1),
496            LiveEvent::InsertLocal { entry } | LiveEvent::InsertRemote { entry, .. } => {
497                self.apply_insert(&entry, params, filter);
498            }
499            LiveEvent::ContentReady { hash } => {
500                self.transferred = self.transferred.saturating_add(self.sizes.remove(&hash).unwrap_or(0));
501            }
502            LiveEvent::PendingContentReady => {}
503            LiveEvent::SyncFinished(sync_event) => {
504                self.sync_finished = true;
505                match sync_event.result {
506                    Ok(details)
507                        if params.area_filter.is_none() && params.area_of_interest.is_none() && filter.is_none() =>
508                    {
509                        self.completed = self
510                            .completed
511                            .saturating_add(u64::try_from(details.entries_received).unwrap_or(u64::MAX));
512                    }
513                    Ok(_details) => {}
514                    Err(error) => return Err(error),
515                }
516            }
517        }
518        Ok(())
519    }
520
521    fn apply_insert(&mut self, entry: &iroh_docs::Entry, params: &SubscribeParams, filter: Option<&FilterEngine>) {
522        let path = entry_path(entry);
523        let hash = entry.content_hash();
524        let accepted = params.accepts(&path, &hash)
525            && filter.is_none_or(|engine| {
526                engine.evaluate(&FilterEntry::new(path.clone(), entry.content_len())) != FilterAction::Reject
527            });
528        if !accepted {
529            return;
530        }
531        let size = entry.content_len();
532        if params
533            .area_of_interest
534            .as_ref()
535            .is_some_and(|area| !area.permits(self.area_count, self.area_size, size))
536        {
537            return;
538        }
539        if params.area_of_interest.is_some() {
540            self.area_count = self.area_count.saturating_add(1);
541            self.area_size = self.area_size.saturating_add(size);
542        }
543        self.completed = self.completed.saturating_add(1);
544        self.sizes.insert(hash, size);
545    }
546}
547
548async fn handle_command(
549    folder: &crate::folder::SyncwebFolder,
550    command: SyncCommand,
551    paused: &mut bool,
552    events: &mpsc::UnboundedSender<SyncEvent>,
553) -> bool {
554    match command {
555        SyncCommand::Pause => {
556            if !*paused {
557                if let Err(error) = folder.doc().leave().await {
558                    let _result = events.send(SyncEvent::Failed(error.to_string()));
559                    return false;
560                }
561                *paused = true;
562            }
563            events.send(SyncEvent::Paused).is_ok()
564        }
565        SyncCommand::Resume => {
566            if *paused {
567                if let Err(error) = folder.doc().start_sync(Vec::new()).await {
568                    let _result = events.send(SyncEvent::Failed(error.to_string()));
569                    return false;
570                }
571                *paused = false;
572            }
573            events.send(SyncEvent::Resumed).is_ok()
574        }
575        SyncCommand::Cancel => {
576            if let Err(error) = folder.doc().leave().await {
577                let _result = events.send(SyncEvent::Failed(error.to_string()));
578                return false;
579            }
580            let _result = events.send(SyncEvent::Cancelled);
581            false
582        }
583    }
584}
585
586fn include_event(params: &SubscribeParams, event: &LiveEvent) -> bool {
587    if params.ingest_only && matches!(event, LiveEvent::InsertLocal { .. }) {
588        return false;
589    }
590    if params.ignore_session.is_some() && matches!(event, LiveEvent::InsertLocal { .. }) {
591        return false;
592    }
593    match event {
594        LiveEvent::InsertLocal { entry } | LiveEvent::InsertRemote { entry, .. } => {
595            let path = entry_path(entry);
596            params.accepts(&path, &entry.content_hash())
597        }
598        LiveEvent::ContentReady { hash } => params
599            .area_of_interest
600            .as_ref()
601            .is_none_or(|area| area.area.matches_hash(hash)),
602        LiveEvent::PendingContentReady
603        | LiveEvent::NeighborUp(_)
604        | LiveEvent::NeighborDown(_)
605        | LiveEvent::SyncFinished(_) => true,
606    }
607}
608
609fn entry_path(entry: &iroh_docs::Entry) -> std::path::PathBuf {
610    Path::new(String::from_utf8_lossy(entry.key()).as_ref()).to_path_buf()
611}
syncweb_core/sync/engine.rs

syncweb_core/sync/
engine.rs
