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(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        self.docs_engine.start_sync(folder.doc(), Vec::new()).await?;
232        let (events, commands, handle) = IntentHandle::channel();
233        let node_db = self.node_db.clone();
234        tokio::spawn(run_intent(
235            folder,
236            IntentConfig { mode, params, filter },
237            self.blob_store.clone(),
238            live_events,
239            events,
240            commands,
241            node_db,
242        ));
243        Ok(handle)
244    }
245}
246
247struct IntentConfig {
248    mode: SessionMode,
249    params: SubscribeParams,
250    filter: Option<FilterEngine>,
251}
252
253async fn run_intent(
254    folder: crate::folder::SyncwebFolder,
255    config: IntentConfig,
256    blob_store: BlobStore,
257    mut live_events: impl n0_future::Stream<Item = Result<LiveEvent>> + Send + Unpin + 'static,
258    events: mpsc::UnboundedSender<SyncEvent>,
259    mut commands: mpsc::UnboundedReceiver<SyncCommand>,
260    node_db: Option<NodeDatabase>,
261) {
262    const SETTLE: Duration = Duration::from_secs(2);
263    const FORCED_QUIESCE: Duration = Duration::from_secs(10);
264
265    let started = Instant::now();
266    if !send_initial_events(&events, started) {
267        return;
268    }
269
270    let (mut state, mut quiesce, mut forced_quiesce, checkpoint, completed_keys) =
271        setup_intent_checkpoint(&folder, node_db.as_ref());
272
273    loop {
274        tokio::select! {
275            received_command = commands.recv() => {
276                let Some(command) = received_command else {
277                    let _result = folder.doc().leave().await;
278                    finalize_checkpoint(checkpoint.as_ref(), &state, &events);
279                    return;
280                };
281                if !handle_command(&folder, command, &mut state.paused, &events).await {
282                    finalize_checkpoint(checkpoint.as_ref(), &state, &events);
283                    return;
284                }
285            }
286            () = &mut quiesce, if !config.mode.is_continuous() && state.sync_finished && state.sizes.is_empty() => {
287                finalize_checkpoint(checkpoint.as_ref(), &state, &events);
288                return;
289            }
290            () = &mut forced_quiesce, if !config.mode.is_continuous() && state.sync_finished => {
291                let pending = state.sizes.len();
292                if pending > 0 {
293                    tracing::warn!(%pending, "forced quiesce: content for some blobs never arrived");
294                }
295                finalize_checkpoint(checkpoint.as_ref(), &state, &events);
296                return;
297            }
298            next_event = live_events.next(), if !state.paused => {
299                let Some(event_result) = next_event else {
300                    let _result = events.send(SyncEvent::Finished);
301                    finalize_checkpoint(checkpoint.as_ref(), &state, &events);
302                    return;
303                };
304                let live_event = match event_result {
305                    Ok(event) => event,
306                    Err(error) => {
307                        let _result = events.send(SyncEvent::Failed(error.to_string()));
308                        finalize_checkpoint(checkpoint.as_ref(), &state, &events);
309                        return;
310                    }
311                };
312                if !include_event(&config.params, &live_event) {
313                    continue;
314                }
315                let entry_key = match &live_event {
316                    LiveEvent::InsertLocal { entry } | LiveEvent::InsertRemote { entry, .. } => {
317                        Some(entry.key().to_vec())
318                    }
319                    LiveEvent::ContentReady { .. }
320                    | LiveEvent::PendingContentReady
321                    | LiveEvent::NeighborUp(_)
322                    | LiveEvent::NeighborDown(_)
323                    | LiveEvent::SyncFinished(_) => None,
324                };
325                if let Some(key) = &entry_key && completed_keys.contains(key) {
326                    continue;
327                }
328                if let LiveEvent::ContentReady { hash } = &live_event {
329                    match blob_store.has(*hash).await {
330                        Ok(true) => {}
331                        Ok(false) => {
332                            let msg = format!("content-ready blob {hash} is missing locally");
333                            let _result = events.send(SyncEvent::Failed(msg));
334                            finalize_checkpoint(checkpoint.as_ref(), &state, &events);
335                            return;
336                        }
337                        Err(error) => {
338                            let _result = events.send(SyncEvent::Failed(error.to_string()));
339                            finalize_checkpoint(checkpoint.as_ref(), &state, &events);
340                            return;
341                        }
342                    }
343                    if let Some(cmd) = throttle_bandwidth(hash, &state, &config, &mut commands).await {
344                        let _ = handle_command(&folder, cmd, &mut state.paused, &events).await;
345                        finalize_checkpoint(checkpoint.as_ref(), &state, &events);
346                        return;
347                    }
348                    if let Some(ref cp) = checkpoint {
349                        let _ = cp.mark_completed(hash.as_bytes(), *hash, state.sizes.get(hash).copied().unwrap_or(0));
350                    }
351                }
352                if let Err(error) = state.apply(live_event, &config.params, config.filter.as_ref()) {
353                    let _result = events.send(SyncEvent::Failed(error));
354                    finalize_checkpoint(checkpoint.as_ref(), &state, &events);
355                    return;
356                }
357                if !config.mode.is_continuous() && state.sync_finished {
358                    let now = tokio::time::Instant::now();
359                    quiesce.as_mut().reset(now.checked_add(SETTLE).unwrap_or(now));
360                    forced_quiesce.as_mut().reset(now.checked_add(FORCED_QUIESCE).unwrap_or(now));
361                }
362                if !send_progress(&events, &state, started.elapsed()) {
363                    finalize_checkpoint(checkpoint.as_ref(), &state, &events);
364                    return;
365                }
366            }
367        }
368    }
369}
370
371async fn throttle_bandwidth(
372    hash: &iroh_blobs::Hash,
373    state: &IntentState,
374    config: &IntentConfig,
375    commands: &mut mpsc::UnboundedReceiver<SyncCommand>,
376) -> Option<SyncCommand> {
377    if let Some(rate) = config.params.bandwidth.as_ref().and_then(|limits| limits.max_download)
378        && rate > 0
379    {
380        let bytes = state.sizes.get(hash).copied().unwrap_or(0);
381        let milliseconds = bytes.saturating_mul(1_000).div_ceil(rate);
382        if milliseconds > 0 {
383            tokio::select! {
384                () = tokio::time::sleep(Duration::from_millis(milliseconds)) => {}
385                received_command = commands.recv() => {
386                    return received_command;
387                }
388            }
389        }
390    }
391    None
392}
393
394fn setup_intent_checkpoint(folder: &crate::folder::SyncwebFolder, node_db: Option<&NodeDatabase>) -> IntentSetup {
395    let namespace_id = folder.namespace_id();
396    let checkpoint = node_db.and_then(|db| match SyncCheckpoint::resume(db, namespace_id) {
397        Ok(Some(cp)) => {
398            tracing::info!(%namespace_id, session = %cp.session_id(), "resuming sync checkpoint");
399            Some(cp)
400        }
401        Ok(None) => SyncCheckpoint::create_session(db, namespace_id)
402            .map_err(|e| tracing::warn!(%namespace_id, %e, "failed to create checkpoint"))
403            .ok(),
404        Err(err) => {
405            tracing::warn!(%namespace_id, %err, "failed to resume checkpoint");
406            SyncCheckpoint::create_session(db, namespace_id)
407                .map_err(|e2| tracing::warn!(%namespace_id, %e2, "failed to create checkpoint"))
408                .ok()
409        }
410    });
411    let completed_keys: HashSet<Vec<u8>> = checkpoint
412        .as_ref()
413        .and_then(|cp| cp.completed_keys().ok())
414        .unwrap_or_default();
415
416    let state = IntentState {
417        completed: u64::try_from(completed_keys.len()).unwrap_or(u64::MAX),
418        ..IntentState::default()
419    };
420    let quiesce = Box::pin(tokio::time::sleep(Duration::MAX));
421    let forced_quiesce = Box::pin(tokio::time::sleep(Duration::MAX));
422    (state, quiesce, forced_quiesce, checkpoint, completed_keys)
423}
424
425fn finalize_checkpoint(
426    checkpoint: Option<&SyncCheckpoint>,
427    state: &IntentState,
428    events: &mpsc::UnboundedSender<SyncEvent>,
429) {
430    if let Some(cp) = checkpoint {
431        if state.completed > 0 {
432            let _ = cp.complete();
433        } else {
434            let _ = cp.leave_unfinished();
435        }
436    }
437    let _ = events.send(SyncEvent::Finished);
438}
439
440fn send_initial_events(events: &mpsc::UnboundedSender<SyncEvent>, started: Instant) -> bool {
441    events.send(SyncEvent::Started).is_ok()
442        && events
443            .send(SyncEvent::Progress {
444                completed: 0,
445                total: Some(0),
446            })
447            .is_ok()
448        && events
449            .send(SyncEvent::Stats(TransferStats::from_progress(
450                0,
451                Some(0),
452                started.elapsed(),
453                0,
454            )))
455            .is_ok()
456}
457
458fn send_progress(events: &mpsc::UnboundedSender<SyncEvent>, state: &IntentState, elapsed: Duration) -> bool {
459    events
460        .send(SyncEvent::Progress {
461            completed: state.completed,
462            total: None,
463        })
464        .is_ok()
465        && events
466            .send(SyncEvent::Stats(TransferStats::from_progress(
467                state.transferred,
468                None,
469                elapsed,
470                state.peer_count,
471            )))
472            .is_ok()
473}
474
475#[derive(Default)]
476struct IntentState {
477    paused: bool,
478    completed: u64,
479    transferred: u64,
480    peer_count: usize,
481    sizes: HashMap<Hash, u64>,
482    area_count: u64,
483    area_size: u64,
484    sync_finished: bool,
485}
486
487impl IntentState {
488    fn apply(
489        &mut self,
490        event: LiveEvent,
491        params: &SubscribeParams,
492        filter: Option<&FilterEngine>,
493    ) -> std::result::Result<(), String> {
494        match event {
495            LiveEvent::NeighborUp(_) => self.peer_count = self.peer_count.saturating_add(1),
496            LiveEvent::NeighborDown(_) => self.peer_count = self.peer_count.saturating_sub(1),
497            LiveEvent::InsertLocal { entry } | LiveEvent::InsertRemote { entry, .. } => {
498                self.apply_insert(&entry, params, filter);
499            }
500            LiveEvent::ContentReady { hash } => {
501                self.transferred = self.transferred.saturating_add(self.sizes.remove(&hash).unwrap_or(0));
502            }
503            LiveEvent::PendingContentReady => {}
504            LiveEvent::SyncFinished(sync_event) => {
505                self.sync_finished = true;
506                match sync_event.result {
507                    Ok(details)
508                        if params.area_filter.is_none() && params.area_of_interest.is_none() && filter.is_none() =>
509                    {
510                        self.completed = self
511                            .completed
512                            .saturating_add(u64::try_from(details.entries_received).unwrap_or(u64::MAX));
513                    }
514                    Ok(_details) => {}
515                    Err(error) => return Err(error),
516                }
517            }
518        }
519        Ok(())
520    }
521
522    fn apply_insert(&mut self, entry: &iroh_docs::Entry, params: &SubscribeParams, filter: Option<&FilterEngine>) {
523        let path = entry_path(entry);
524        let hash = entry.content_hash();
525        let accepted = params.accepts(&path, &hash)
526            && filter.is_none_or(|engine| {
527                engine.evaluate(&FilterEntry::new(path.clone(), entry.content_len())) != FilterAction::Reject
528            });
529        if !accepted {
530            return;
531        }
532        let size = entry.content_len();
533        if params
534            .area_of_interest
535            .as_ref()
536            .is_some_and(|area| !area.permits(self.area_count, self.area_size, size))
537        {
538            return;
539        }
540        if params.area_of_interest.is_some() {
541            self.area_count = self.area_count.saturating_add(1);
542            self.area_size = self.area_size.saturating_add(size);
543        }
544        self.completed = self.completed.saturating_add(1);
545        self.sizes.insert(hash, size);
546    }
547}
548
549async fn handle_command(
550    folder: &crate::folder::SyncwebFolder,
551    command: SyncCommand,
552    paused: &mut bool,
553    events: &mpsc::UnboundedSender<SyncEvent>,
554) -> bool {
555    match command {
556        SyncCommand::Pause => {
557            if !*paused {
558                if let Err(error) = folder.doc().leave().await {
559                    let _result = events.send(SyncEvent::Failed(error.to_string()));
560                    return false;
561                }
562                *paused = true;
563            }
564            events.send(SyncEvent::Paused).is_ok()
565        }
566        SyncCommand::Resume => {
567            if *paused {
568                if let Err(error) = folder.doc().start_sync(Vec::new()).await {
569                    let _result = events.send(SyncEvent::Failed(error.to_string()));
570                    return false;
571                }
572                *paused = false;
573            }
574            events.send(SyncEvent::Resumed).is_ok()
575        }
576        SyncCommand::Cancel => {
577            if let Err(error) = folder.doc().leave().await {
578                let _result = events.send(SyncEvent::Failed(error.to_string()));
579                return false;
580            }
581            let _result = events.send(SyncEvent::Cancelled);
582            false
583        }
584    }
585}
586
587fn include_event(params: &SubscribeParams, event: &LiveEvent) -> bool {
588    if params.ingest_only && matches!(event, LiveEvent::InsertLocal { .. }) {
589        return false;
590    }
591    if params.ignore_session.is_some() && matches!(event, LiveEvent::InsertLocal { .. }) {
592        return false;
593    }
594    match event {
595        LiveEvent::InsertLocal { entry } | LiveEvent::InsertRemote { entry, .. } => {
596            let path = entry_path(entry);
597            params.accepts(&path, &entry.content_hash())
598        }
599        LiveEvent::ContentReady { hash } => params
600            .area_of_interest
601            .as_ref()
602            .is_none_or(|area| area.area.matches_hash(hash)),
603        LiveEvent::PendingContentReady
604        | LiveEvent::NeighborUp(_)
605        | LiveEvent::NeighborDown(_)
606        | LiveEvent::SyncFinished(_) => true,
607    }
608}
609
610fn entry_path(entry: &iroh_docs::Entry) -> std::path::PathBuf {
611    Path::new(String::from_utf8_lossy(entry.key()).as_ref()).to_path_buf()
612}
syncweb_core/sync/engine.rs

syncweb_core/sync/
engine.rs
