syncweb_core/daemon/
transfers.rs

1//! Durable transfer-job processing shared between the daemon and the
2//! embedded CLI path (`transfer enqueue --now`).
3
4use std::path::{Path, PathBuf};
5use std::time::Duration;
6
7use n0_future::StreamExt;
8
9use crate::{
10    allocation::{AllocationCandidate, StorageRoot, materialization_path},
11    error::{Result, SyncwebError},
12    folder::FolderManager,
13    node::iroh_node::IrohNode,
14    storage::node_db::{NodeDatabase, TransferJobRecord},
15    sync::{FetchFilter, FetchStrategy, SyncEngine, SyncEvent},
16};
17
18const TRANSFER_TIMEOUT: Duration = Duration::from_mins(5);
19
20/// Outcome counters from processing a batch of transfer jobs.
21#[derive(Clone, Copy, Debug, Default)]
22#[non_exhaustive]
23pub struct TransferJobSummary {
24    pub completed: u64,
25    pub failed: u64,
26}
27
28impl TransferJobSummary {
29    /// Create a summary from completion counters.
30    #[must_use]
31    pub const fn new(completed: u64, failed: u64) -> Self {
32        Self { completed, failed }
33    }
34}
35
36/// Fetch and materialize all queued transfer jobs, optionally scoped to a
37/// single namespace.
38///
39/// # Errors
40///
41/// Returns an error if the queued jobs cannot be listed.
42pub async fn process_transfer_jobs(
43    node: &IrohNode,
44    node_db: &NodeDatabase,
45    namespace: Option<&str>,
46) -> Result<TransferJobSummary> {
47    let jobs = node_db.list_transfer_jobs(namespace, Some("queued"))?;
48    let mut summary = TransferJobSummary::default();
49    for job in jobs {
50        match process_transfer_job(node, node_db, job).await {
51            TransferJobOutcome::Completed => summary.completed = summary.completed.saturating_add(1),
52            TransferJobOutcome::Failed => summary.failed = summary.failed.saturating_add(1),
53            TransferJobOutcome::Skipped => {}
54        }
55    }
56    Ok(summary)
57}
58
59#[derive(Clone, Copy)]
60enum TransferJobOutcome {
61    Completed,
62    Failed,
63    Skipped,
64}
65
66async fn process_transfer_job(node: &IrohNode, node_db: &NodeDatabase, job: TransferJobRecord) -> TransferJobOutcome {
67    let Some(destination) = job.destination.as_ref() else {
68        mark_transfer_job_failed(
69            node_db,
70            &job.id,
71            &"job has no allocated destination",
72            "missing transfer destination",
73        );
74        return TransferJobOutcome::Failed;
75    };
76    let hash = iroh_blobs::Hash::from(job.hash);
77    if let Err(error) = validate_transfer_job(node_db, &job, destination, hash) {
78        mark_transfer_job_failed(node_db, &job.id, &error, "transfer path validation error");
79        return TransferJobOutcome::Failed;
80    }
81
82    let fetch_claimed = match node_db.transition_transfer_job_state(&job.id, "queued", "fetching", None) {
83        Ok(claimed) => claimed,
84        Err(error) => {
85            tracing::error!(%error, job_id = %job.id, "failed to mark transfer job fetching");
86            return TransferJobOutcome::Failed;
87        }
88    };
89    if !fetch_claimed {
90        return TransferJobOutcome::Skipped;
91    }
92
93    let has_blob = match node.blob_store().has(hash).await {
94        Ok(has_blob) => has_blob,
95        Err(error) => {
96            mark_transfer_job_failed(node_db, &job.id, &error, "blob lookup error");
97            return TransferJobOutcome::Failed;
98        }
99    };
100    if !has_blob {
101        let path = match String::from_utf8(job.entry_key.clone()) {
102            Ok(path) => path,
103            Err(error) => {
104                mark_transfer_job_failed(
105                    node_db,
106                    &job.id,
107                    &format!("job entry path is not UTF-8: {error}"),
108                    "invalid transfer path",
109                );
110                return TransferJobOutcome::Failed;
111            }
112        };
113        let namespace = match job.namespace_id.parse::<iroh_docs::NamespaceId>() {
114            Ok(namespace) => namespace,
115            Err(error) => {
116                mark_transfer_job_failed(
117                    node_db,
118                    &job.id,
119                    &format!("job has an invalid namespace: {error}"),
120                    "invalid transfer namespace",
121                );
122                return TransferJobOutcome::Failed;
123            }
124        };
125        if let Err(error) = fetch_transfer_blob(node, namespace, &path).await {
126            mark_transfer_job_failed(node_db, &job.id, &error, "transfer fetch error");
127            return TransferJobOutcome::Failed;
128        }
129    }
130
131    let materialization_claimed =
132        match node_db.transition_transfer_job_state(&job.id, "fetching", "materializing", None) {
133            Ok(claimed) => claimed,
134            Err(error) => {
135                tracing::error!(%error, job_id = %job.id, "failed to mark transfer job materializing");
136                return TransferJobOutcome::Failed;
137            }
138        };
139    if !materialization_claimed {
140        return TransferJobOutcome::Skipped;
141    }
142
143    match materialize_transfer(node, destination, hash).await {
144        Ok(size) => complete_transfer_job(node_db, &job, size),
145        Err(error) => {
146            mark_transfer_job_failed(node_db, &job.id, &error, "transfer materialization error");
147            TransferJobOutcome::Failed
148        }
149    }
150}
151
152/// Fetch a single blob identified by a folder path via the sync engine.
153async fn fetch_transfer_blob(node: &IrohNode, namespace_id: iroh_docs::NamespaceId, path: &str) -> Result<()> {
154    let sync = SyncEngine::new(
155        FolderManager::new(node),
156        node.blob_store().clone(),
157        node.docs_engine().clone(),
158        Some(node.topic_tracker().clone()),
159    );
160    let filter = FetchFilter::new().with_paths(vec![PathBuf::from(path)]);
161    let mut intent = sync.fetch(namespace_id, FetchStrategy::filter(filter)).await?;
162    let body = async {
163        while let Some(event) = intent.next().await {
164            match event {
165                SyncEvent::Failed(message) => {
166                    return Err(SyncwebError::operation("transfer fetch failed", message));
167                }
168                SyncEvent::Finished => return Ok(()),
169                SyncEvent::Stats(_)
170                | SyncEvent::Started
171                | SyncEvent::Progress { .. }
172                | SyncEvent::Paused
173                | SyncEvent::Resumed
174                | SyncEvent::Cancelled => {}
175            }
176        }
177        Ok(())
178    };
179    match tokio::time::timeout(TRANSFER_TIMEOUT, body).await {
180        Ok(result) => result,
181        Err(_elapsed) => {
182            let _ = intent.cancel();
183            Ok(())
184        }
185    }
186}
187
188/// Validate that a queued job still points at its allocated root and path.
189fn validate_transfer_job(
190    node_db: &NodeDatabase,
191    job: &TransferJobRecord,
192    destination: &Path,
193    hash: iroh_blobs::Hash,
194) -> Result<()> {
195    let root_id = job
196        .root_id
197        .as_deref()
198        .ok_or_else(|| SyncwebError::InvalidConfig("job has no storage root".to_owned()))?;
199    let root = node_db
200        .list_storage_roots()?
201        .into_iter()
202        .find(|root| root.id == root_id)
203        .ok_or_else(|| SyncwebError::InvalidConfig(format!("storage root not found: {root_id}")))?;
204    let job_namespace = job
205        .namespace_id
206        .parse::<iroh_docs::NamespaceId>()
207        .map_err(|error| SyncwebError::operation("job has invalid namespace", error))?;
208    let path = String::from_utf8(job.entry_key.clone())
209        .map_err(|error| SyncwebError::operation("job entry path is not UTF-8", error))?;
210    let root_path = root.path.clone();
211    let candidate = AllocationCandidate::new(
212        job_namespace,
213        PathBuf::from(path),
214        hash,
215        job.size,
216        usize::try_from(job.peer_count).unwrap_or(usize::MAX),
217        false,
218    );
219    let expected = materialization_path(
220        &StorageRoot::new(root.id, &root_path, root.min_free).with_enabled(root.enabled),
221        &candidate,
222    )?;
223    if expected != destination {
224        return Err(SyncwebError::InvalidConfig(format!(
225            "job destination does not match its allocated root: {}",
226            destination.display()
227        )));
228    }
229    reject_symlink_components(
230        expected.strip_prefix(&root_path).unwrap_or_else(|_| Path::new("")),
231        &root_path,
232    )?;
233    Ok(())
234}
235
236async fn materialize_transfer(node: &IrohNode, destination: &Path, hash: iroh_blobs::Hash) -> Result<u64> {
237    if let Ok(metadata) = std::fs::symlink_metadata(destination) {
238        if metadata.file_type().is_symlink() || !metadata.is_file() {
239            return Err(SyncwebError::InvalidConfig(format!(
240                "materialization destination is not a regular file: {}",
241                destination.display()
242            )));
243        }
244        let existing = std::fs::read(destination)?;
245        if blake3::hash(&existing).as_bytes() != hash.as_bytes() {
246            return Err(SyncwebError::InvalidConfig(format!(
247                "materialization destination has a different blob: {}",
248                destination.display()
249            )));
250        }
251    }
252    if let Some(parent) = destination.parent() {
253        std::fs::create_dir_all(parent)?;
254    }
255    node.blob_store().export_to_path(hash, destination).await?;
256    let bytes = std::fs::read(destination)?;
257    if blake3::hash(&bytes).as_bytes() != hash.as_bytes() {
258        return Err(SyncwebError::operation(
259            "materialized transfer hash does not match",
260            destination.display(),
261        ));
262    }
263    u64::try_from(bytes.len()).map_err(|error| SyncwebError::operation("materialized transfer size is invalid", error))
264}
265
266fn complete_transfer_job(node_db: &NodeDatabase, job: &TransferJobRecord, size: u64) -> TransferJobOutcome {
267    if let Err(error) = node_db.update_transfer_job_progress(&job.id, size, job.peer_count, None, job.retries) {
268        tracing::error!(%error, job_id = %job.id, "failed to record transfer completion progress");
269        return TransferJobOutcome::Failed;
270    }
271    match node_db.transition_transfer_job_state(&job.id, "materializing", "completed", None) {
272        Ok(true) => TransferJobOutcome::Completed,
273        Ok(false) => TransferJobOutcome::Skipped,
274        Err(error) => {
275            tracing::error!(%error, job_id = %job.id, "failed to record transfer completion");
276            TransferJobOutcome::Failed
277        }
278    }
279}
280
281fn mark_transfer_job_failed(node_db: &NodeDatabase, job_id: &str, message: &dyn std::fmt::Display, context: &str) {
282    let rendered_message = message.to_string();
283    if let Err(error) = node_db.update_transfer_job_state(job_id, "failed", Some(&rendered_message)) {
284        tracing::error!(%error, %job_id, context, "failed to record transfer job failure");
285    }
286}
287
288fn reject_symlink_components(relative: &Path, root: &Path) -> Result<()> {
289    let mut current = root.to_path_buf();
290    for component in relative.components() {
291        current.push(component.as_os_str());
292        let metadata = match std::fs::symlink_metadata(&current) {
293            Ok(metadata) => metadata,
294            Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue,
295            Err(error) => return Err(error.into()),
296        };
297        if metadata.file_type().is_symlink() {
298            return Err(SyncwebError::InvalidConfig(format!(
299                "materialization path contains a symlink: {}",
300                current.display()
301            )));
302        }
303    }
304    Ok(())
305}
syncweb_core/daemon/transfers.rs

syncweb_core/daemon/
transfers.rs
