syncweb_core/sync/
engine.rs

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

syncweb_core/sync/
engine.rs
