syncweb_core/sync/
engine.rs

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

syncweb_core/sync/
engine.rs
