syncweb_core/sync/
partial_fetch.rs

1use std::collections::HashMap;
2use std::path::PathBuf;
3
4use iroh_blobs::Hash;
5use serde::{Deserialize, Serialize};
6
7use crate::error::Result;
8use crate::folder::SyncwebFolder;
9use crate::node::iroh_node::IrohNode;
10
11/// Fetch the blobs selected by `strategy` directly from the folder's namespace
12/// peers.
13///
14/// A metadata-only folder syncs entries while deliberately leaving blobs out of
15/// the local store; a later doc re-sync does not re-download that content, so
16/// `download` on such a folder must fetch each selected blob from its peers
17/// directly. Returns the number of bytes fetched.
18///
19/// # Errors
20///
21/// Returns an error if the folder's entries cannot be listed.
22pub async fn fetch_selected_content(node: &IrohNode, folder: &SyncwebFolder, strategy: &FetchStrategy) -> Result<u64> {
23    let mut candidates = Vec::new();
24    for entry in folder.content_entries().await? {
25        let hash = entry.content_hash();
26        let local = folder.has_local(hash).await?;
27        let path = PathBuf::from(String::from_utf8_lossy(entry.key()).into_owned());
28        candidates.push(FetchCandidate::new(path, hash, entry.content_len(), 0, local));
29    }
30    let selected = strategy.select(&candidates);
31    let peers = node
32        .topic_tracker()
33        .find_peers(folder.namespace_id())
34        .await
35        .unwrap_or_default();
36    let mut bytes = 0_u64;
37    for candidate in selected {
38        if candidate.local {
39            continue;
40        }
41        for peer in &peers {
42            if node
43                .blob_store()
44                .force_fetch_from_peer(node.endpoint(), peer, candidate.hash)
45                .await
46                .is_ok()
47            {
48                bytes = bytes.saturating_add(candidate.size);
49                break;
50            }
51        }
52    }
53    Ok(bytes)
54}
55
56/// A document blob considered for a partial fetch.
57#[derive(Clone, Debug, Eq, PartialEq)]
58#[non_exhaustive]
59pub struct FetchCandidate {
60    pub path: PathBuf,
61    pub hash: Hash,
62    pub size: u64,
63    pub peer_count: usize,
64    pub local: bool,
65}
66
67impl FetchCandidate {
68    #[must_use]
69    pub fn new(path: impl Into<PathBuf>, hash: Hash, size: u64, peer_count: usize, local: bool) -> Self {
70        Self {
71            path: path.into(),
72            hash,
73            size,
74            peer_count,
75            local,
76        }
77    }
78}
79
80/// Constraints for selecting a subset of folder content.
81#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
82#[non_exhaustive]
83pub struct FetchFilter {
84    pub paths: Option<Vec<PathBuf>>,
85    pub min_size: Option<u64>,
86    pub max_size: Option<u64>,
87    pub min_peers: Option<usize>,
88    pub max_peers: Option<usize>,
89    pub min_count: Option<usize>,
90    pub max_count: Option<usize>,
91}
92
93impl FetchFilter {
94    #[must_use]
95    pub const fn new() -> Self {
96        Self {
97            paths: None,
98            min_size: None,
99            max_size: None,
100            min_peers: None,
101            max_peers: None,
102            min_count: None,
103            max_count: None,
104        }
105    }
106
107    #[must_use]
108    pub fn with_paths(mut self, paths: Vec<PathBuf>) -> Self {
109        self.paths = Some(paths);
110        self
111    }
112
113    #[must_use]
114    pub const fn with_min_size(mut self, size: u64) -> Self {
115        self.min_size = Some(size);
116        self
117    }
118
119    #[must_use]
120    pub const fn with_max_size(mut self, size: u64) -> Self {
121        self.max_size = Some(size);
122        self
123    }
124
125    #[must_use]
126    pub const fn with_min_peers(mut self, peers: usize) -> Self {
127        self.min_peers = Some(peers);
128        self
129    }
130
131    #[must_use]
132    pub const fn with_max_peers(mut self, peers: usize) -> Self {
133        self.max_peers = Some(peers);
134        self
135    }
136
137    #[must_use]
138    pub const fn with_min_count(mut self, count: usize) -> Self {
139        self.min_count = Some(count);
140        self
141    }
142
143    #[must_use]
144    pub const fn with_max_count(mut self, count: usize) -> Self {
145        self.max_count = Some(count);
146        self
147    }
148
149    /// Return candidates satisfying this filter.
150    ///
151    /// Candidates with fewer peers are selected first so a capped fetch
152    /// improves folder seeding rather than repeatedly selecting common blobs.
153    #[must_use]
154    pub fn select(&self, candidates: &[FetchCandidate]) -> Vec<FetchCandidate> {
155        let mut selected = candidates
156            .iter()
157            .filter(|candidate| self.matches(candidate))
158            .cloned()
159            .collect::<Vec<_>>();
160        selected.sort_by(|left, right| {
161            left.peer_count
162                .cmp(&right.peer_count)
163                .then(left.path.cmp(&right.path))
164                .then(left.hash.cmp(&right.hash))
165        });
166        if let Some(max_count) = self.max_count {
167            selected.truncate(max_count);
168        }
169        selected
170    }
171
172    #[must_use]
173    pub fn matches(&self, candidate: &FetchCandidate) -> bool {
174        self.paths.as_ref().is_none_or(|paths| {
175            paths
176                .iter()
177                .any(|path| candidate.path == *path || candidate.path.starts_with(path))
178        }) && self.min_size.is_none_or(|size| candidate.size >= size)
179            && self.max_size.is_none_or(|size| candidate.size <= size)
180            && self.min_peers.is_none_or(|peers| candidate.peer_count >= peers)
181            && self.max_peers.is_none_or(|peers| candidate.peer_count <= peers)
182    }
183}
184
185/// Controls whether a fetch reconciles all content or only matching content.
186#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
187#[non_exhaustive]
188pub enum FetchStrategy {
189    #[default]
190    All,
191    Filter(FetchFilter),
192}
193
194impl FetchStrategy {
195    #[must_use]
196    pub const fn filter(filter: FetchFilter) -> Self {
197        Self::Filter(filter)
198    }
199
200    #[must_use]
201    pub fn select(&self, candidates: &[FetchCandidate]) -> Vec<FetchCandidate> {
202        match self {
203            Self::All => candidates.to_vec(),
204            Self::Filter(filter) => filter.select(candidates),
205        }
206    }
207}
208
209/// Per-blob availability information used by the health command.
210#[derive(Clone, Debug, Eq, PartialEq)]
211#[non_exhaustive]
212pub struct BlobHealth {
213    pub path: PathBuf,
214    pub hash: Hash,
215    pub size: u64,
216    pub peer_count: usize,
217    pub local: bool,
218}
219
220/// Aggregate availability counters for a folder.
221#[derive(Clone, Debug, Eq, PartialEq)]
222#[non_exhaustive]
223pub struct HealthReport {
224    pub total: usize,
225    pub well_seeded: usize,
226    pub under_seeded: usize,
227    pub unseeded: usize,
228    pub least_seeded: Vec<BlobHealth>,
229}
230
231impl HealthReport {
232    /// Build a report using `well_seeded_threshold` as the minimum healthy peer count.
233    #[must_use]
234    pub fn from_candidates(candidates: &[FetchCandidate], well_seeded_threshold: usize) -> Self {
235        let mut least_seeded = candidates
236            .iter()
237            .map(|candidate| BlobHealth {
238                path: candidate.path.clone(),
239                hash: candidate.hash,
240                size: candidate.size,
241                peer_count: candidate.peer_count,
242                local: candidate.local,
243            })
244            .collect::<Vec<_>>();
245        least_seeded.sort_by(|left, right| left.peer_count.cmp(&right.peer_count).then(left.path.cmp(&right.path)));
246        let total = candidates.len();
247        let well_seeded = candidates
248            .iter()
249            .filter(|candidate| candidate.peer_count >= well_seeded_threshold)
250            .count();
251        let unseeded = candidates.iter().filter(|candidate| candidate.peer_count == 0).count();
252        Self {
253            total,
254            well_seeded,
255            under_seeded: total.saturating_sub(well_seeded).saturating_sub(unseeded),
256            unseeded,
257            least_seeded,
258        }
259    }
260
261    /// Build a report with verified peer counts from a `hash -> verified_count` map.
262    ///
263    /// Each candidate's `peer_count` is overridden with the value from
264    /// `peers_per_hash` when present. This lets the health display reflect live
265    /// provider-lease data instead of always showing zero.
266    #[must_use]
267    pub fn from_candidates_with_peers_per_hash(
268        candidates: &[FetchCandidate],
269        peers_per_hash: &HashMap<Hash, usize>,
270        well_seeded_threshold: usize,
271    ) -> Self {
272        let enriched: Vec<FetchCandidate> = candidates
273            .iter()
274            .map(|c| {
275                let peers = peers_per_hash.get(&c.hash).copied().unwrap_or(c.peer_count);
276                FetchCandidate::new(&c.path, c.hash, c.size, peers, c.local)
277            })
278            .collect();
279        Self::from_candidates(&enriched, well_seeded_threshold)
280    }
281}
syncweb_core/sync/partial_fetch.rs

syncweb_core/sync/
partial_fetch.rs
