Something went wrong. Try again.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713use std::{ collections::{BTreeMap, BTreeSet, VecDeque}, fs, path::Path, str::FromStr,};
use anyhow::Context;use futures_util::{StreamExt, stream};
use crate::{ app::{MAX_CONCURRENT_BLOB_TRANSFERS, SyncContext, manifest::deleted_entry}, domain::{DeviceId, Entry, EntryKind, Manifest, ReconciliationAction}, filesystem::{IgnoreRules, remove_path, safe_destination}, iroh::NodeHost, storage::FolderConfig,};
pub(super) struct RemoteManifestResult { pub(super) manifest: Manifest, pub(super) synchronized_file_count: usize,}
pub(super) async fn apply_remote_manifest( context: &SyncContext<'_>, local: &Manifest, remote: &Manifest, peer: iroh::EndpointAddr,) -> anyhow::Result<RemoteManifestResult> { let mut merged = local.clone(); let mut synchronized_count = 0; let ignore_rules = IgnoreRules::load(&context.folder.path)?; let actions = crate::domain::reconcile_manifests(local, remote); let (mut materialization_plan, downloads) = plan_remote_materialization(&ignore_rules, context.folder, actions, remote); materialize_ready_actions( context, &mut merged, &mut synchronized_count, &mut materialization_plan, ) .await?; receive_and_materialize_files( context, &peer, &mut merged, &mut synchronized_count, &mut materialization_plan, downloads, ) .await?; Ok(RemoteManifestResult { manifest: merged, synchronized_file_count: synchronized_count, })}
pub(super) async fn restore_manifest( context: &SyncContext<'_>, current: &Manifest, historical: &Manifest,) -> anyhow::Result<usize> { let paths = current .entries .keys() .chain(historical.entries.keys()) .collect::<std::collections::BTreeSet<_>>(); let device_id = context.node.endpoint_address().id.to_string(); let mut restored = Manifest::empty(context.folder.id); let mut file_writes = 0; for path in paths { let current_entry = current.entries.get(path); let historical_entry = historical.entries.get(path); let Some(entry) = restored_entry(path, current_entry, historical_entry, &device_id, context)? else { continue; }; if current_entry != Some(&entry) { file_writes += materialize_entries(context, &mut restored, vec![entry]).await?; } else { restored.entries.insert(path.clone(), entry); } } context.state_store.save_manifest(&restored)?; Ok(file_writes)}
fn restored_entry( path: &str, current: Option<&Entry>, historical: Option<&Entry>, device_id: &DeviceId, context: &SyncContext<'_>,) -> anyhow::Result<Option<Entry>> { let Some(historical) = historical else { return current .cloned() .map(|entry| deleted_entry(path.to_owned(), entry, device_id, context)) .transpose(); }; let mut restored = historical.clone(); let mut clock = current .map(|entry| entry.clock.clone()) .unwrap_or_else(|| historical.clock.clone()); clock.insert( device_id.clone(), context .state_store .next_counter(context.folder.id, device_id)?, ); restored.clock = clock; restored.author_device_id = device_id.clone(); Ok(Some(restored))}
fn action_is_ignored( folder: &FolderConfig, ignore_rules: &IgnoreRules, action: &ReconciliationAction,) -> bool { let entry = match action { ReconciliationAction::ApplyRemote { entry } | ReconciliationAction::KeepLocal { entry } => { entry } ReconciliationAction::ResolveConflict { winner, .. } => winner, }; ignore_rules.ignores( &folder.path.join(&entry.path), entry.kind == EntryKind::Directory, )}
pub(super) async fn recover_pending_materialization( context: &SyncContext<'_>,) -> anyhow::Result<()> { let Some(pending) = context .state_store .pending_materialization(context.folder.id)? else { return Ok(()); }; if pending.is_legacy_single_entry { tracing::warn!("Discarding a stale single-entry materialization journal"); context .state_store .clear_pending_materialization(context.folder.id)?; return Ok(()); } if pending.resulting_manifest.folder_id != context.folder.id { anyhow::bail!("pending materialization belongs to another folder"); } tracing::warn!( entry_count = pending.entries.len(), "Recovering interrupted filesystem updates" ); for entry in &pending.entries { apply_entry(context, entry).await?; } context .state_store .save_manifest(&pending.resulting_manifest)?; context .state_store .clear_pending_materialization(context.folder.id)?; Ok(())}
async fn materialize_entries( context: &SyncContext<'_>, manifest: &mut Manifest, entries: Vec<Entry>,) -> anyhow::Result<usize> { let mut resulting_manifest = manifest.clone(); for entry in &entries { resulting_manifest .entries .insert(entry.path.clone(), entry.clone()); } context.state_store.save_pending_materialization( context.folder.id, &entries, &resulting_manifest, )?; let mut writes = 0; for entry in &entries { writes += apply_entry(context, entry).await?; } context.state_store.save_manifest(&resulting_manifest)?; context .state_store .clear_pending_materialization(context.folder.id)?; *manifest = resulting_manifest; Ok(writes)}
async fn apply_entry(context: &SyncContext<'_>, entry: &Entry) -> anyhow::Result<usize> { let destination = safe_destination(&context.folder.path, &entry.path)?; match entry.kind { EntryKind::Deleted => { remove_path(&destination)?; Ok(0) } EntryKind::Directory => { replace_non_directory(&destination)?; fs::create_dir_all(destination)?; Ok(0) } EntryKind::File => { let blob_hash = iroh_blobs::Hash::from_str( entry .blob_hash .as_deref() .ok_or_else(|| anyhow::anyhow!("file entry has no blob hash"))?, )?; remove_existing_destination(&destination)?; write_blob_atomically(context.node, blob_hash, &destination).await?; tracing::info!(path = %entry.path, "Wrote file"); Ok(1) } }}
fn replace_non_directory(destination: &Path) -> anyhow::Result<()> { let metadata = match fs::symlink_metadata(destination) { Ok(metadata) => metadata, Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(()), Err(error) => return Err(error.into()), }; if metadata.is_dir() { return Ok(()); } remove_path(destination)}
fn remove_existing_destination(destination: &Path) -> anyhow::Result<()> { match fs::symlink_metadata(destination) { Ok(_) => remove_path(destination), Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()), Err(error) => Err(error.into()), }}
struct RemoteFile { path: String, blob_hash: String, size_bytes: Option<u64>,}
const SMALL_BLOB_MAX_BYTES: u64 = 256 * 1024;const MAX_SMALL_BLOBS_PER_REQUEST: usize = 128;const MAX_SMALL_BLOB_REQUEST_BYTES: u64 = 4 * 1024 * 1024;
struct PendingAction { action: ReconciliationAction, missing_blob_count: usize,}
struct MaterializationPlan { pending_actions: Vec<Option<PendingAction>>, actions_waiting_by_blob: BTreeMap<String, Vec<usize>>, ready_actions: VecDeque<usize>,}
fn plan_remote_materialization( ignore_rules: &IgnoreRules, folder: &FolderConfig, actions: Vec<ReconciliationAction>, remote: &Manifest,) -> (MaterializationPlan, Vec<RemoteFile>) { let mut files = BTreeMap::new(); let mut pending_actions = Vec::new(); let mut actions_waiting_by_blob = BTreeMap::new(); let mut ready_actions = VecDeque::new(); for action in actions .into_iter() .filter(|action| !action_is_ignored(folder, ignore_rules, action)) { let required_blobs = collect_required_blobs(&mut files, remote, &action); let action_id = pending_actions.len(); if required_blobs.is_empty() { ready_actions.push_back(action_id); } else { for blob_hash in &required_blobs { actions_waiting_by_blob .entry(blob_hash.clone()) .or_insert_with(Vec::new) .push(action_id); } } pending_actions.push(Some(PendingAction { action, missing_blob_count: required_blobs.len(), })); } ( MaterializationPlan { pending_actions, actions_waiting_by_blob, ready_actions, }, files.into_values().collect(), )}
fn collect_required_blobs( files: &mut BTreeMap<String, RemoteFile>, remote: &Manifest, action: &ReconciliationAction,) -> BTreeSet<String> { let mut required_blobs = BTreeSet::new(); for entry in remote_entries_for_action(action) { collect_remote_file(files, &mut required_blobs, remote, entry); } required_blobs}
fn remote_entries_for_action(action: &ReconciliationAction) -> Vec<&Entry> { match action { ReconciliationAction::ApplyRemote { entry } => vec![entry], ReconciliationAction::KeepLocal { .. } => Vec::new(), ReconciliationAction::ResolveConflict { winner, loser, .. } => vec![winner, loser], }}
async fn materialize_ready_actions( context: &SyncContext<'_>, merged: &mut Manifest, synchronized_count: &mut usize, plan: &mut MaterializationPlan,) -> anyhow::Result<()> { while let Some(actions) = plan.take_ready_actions() { *synchronized_count += materialize_action_batch(context, merged, actions).await?; } Ok(())}
impl MaterializationPlan { fn note_completed_blob(&mut self, blob_hash: String) { let Some(waiting_actions) = self.actions_waiting_by_blob.remove(&blob_hash) else { return; }; for action_id in waiting_actions { let Some(pending) = self.pending_actions[action_id].as_mut() else { continue; }; pending.missing_blob_count -= 1; if pending.missing_blob_count == 0 { self.ready_actions.push_back(action_id); } } }
fn take_ready_actions(&mut self) -> Option<Vec<ReconciliationAction>> { const MATERIALIZATION_BATCH_SIZE: usize = 64; let mut actions = Vec::new(); while actions.len() < MATERIALIZATION_BATCH_SIZE { let Some(action_id) = self.ready_actions.pop_front() else { break; }; let Some(pending) = self.pending_actions[action_id].take() else { continue; }; actions.push(pending.action); } (!actions.is_empty()).then_some(actions) }}
async fn materialize_action_batch( context: &SyncContext<'_>, merged: &mut Manifest, actions: Vec<ReconciliationAction>,) -> anyhow::Result<usize> { let entries = actions .into_iter() .flat_map(entries_for_action) .collect::<Vec<_>>(); if entries.is_empty() { return Ok(0); } materialize_entries(context, merged, entries).await}
fn entries_for_action(action: ReconciliationAction) -> Vec<Entry> { match action { ReconciliationAction::ApplyRemote { entry } => vec![entry], ReconciliationAction::KeepLocal { .. } => Vec::new(), ReconciliationAction::ResolveConflict { winner, loser, conflict_path, } => { let mut conflict_entry = loser; conflict_entry.path = conflict_path; vec![conflict_entry, winner] } }}
async fn receive_and_materialize_files( context: &SyncContext<'_>, peer: &iroh::EndpointAddr, merged: &mut Manifest, synchronized_count: &mut usize, plan: &mut MaterializationPlan, downloads: Vec<RemoteFile>,) -> anyhow::Result<()> { // Protect every blob we are about to download from garbage collection until // the manifest references it (which happens inside `materialize_entries`). let _download_guards = protect_download_hashes(context.node, &downloads).await; let mut downloads = stream::iter(plan_download_batches(downloads)) .map(|files| download_files_with_isolation(context, peer, files)) .buffer_unordered(MAX_CONCURRENT_BLOB_TRANSFERS); while let Some(result) = downloads.next().await { match result { Ok(blob_hashes) => { for blob_hash in blob_hashes { plan.note_completed_blob(blob_hash); } materialize_ready_actions(context, merged, synchronized_count, plan).await?; } Err(error) => { tracing::warn!(error = ?error, "Could not receive file; deferring dependent changes") } } } Ok(())}
async fn protect_download_hashes( node: &NodeHost, downloads: &[RemoteFile],) -> Vec<crate::iroh::TempTag> { let hashes = downloads .iter() .filter_map(|file| iroh_blobs::Hash::from_str(&file.blob_hash).ok()) .collect::<Vec<_>>(); match node.protect_blobs(&hashes).await { Ok(guards) => guards, Err(error) => { tracing::warn!(%error, "Could not protect download blobs from GC; relying on GC timing"); Vec::new() } }}
fn plan_download_batches(downloads: Vec<RemoteFile>) -> Vec<Vec<RemoteFile>> { let mut batches = Vec::new(); let mut small_files = Vec::new(); let mut small_file_bytes = 0; for file in downloads { if !is_small_file(&file) { push_small_file_batch(&mut batches, &mut small_files, &mut small_file_bytes); batches.push(vec![file]); continue; } if small_files.len() == MAX_SMALL_BLOBS_PER_REQUEST || small_file_bytes + file.size_bytes.unwrap_or_default() > MAX_SMALL_BLOB_REQUEST_BYTES { push_small_file_batch(&mut batches, &mut small_files, &mut small_file_bytes); } small_file_bytes += file.size_bytes.unwrap_or_default(); small_files.push(file); } push_small_file_batch(&mut batches, &mut small_files, &mut small_file_bytes); batches}
fn is_small_file(file: &RemoteFile) -> bool { file.size_bytes .is_some_and(|size| size <= SMALL_BLOB_MAX_BYTES)}
fn push_small_file_batch( batches: &mut Vec<Vec<RemoteFile>>, small_files: &mut Vec<RemoteFile>, small_file_bytes: &mut u64,) { if small_files.is_empty() { return; } batches.push(std::mem::take(small_files)); *small_file_bytes = 0;}
async fn download_files_with_isolation( context: &SyncContext<'_>, peer: &iroh::EndpointAddr, files: Vec<RemoteFile>,) -> anyhow::Result<Vec<String>> { let mut completed_hashes = Vec::new(); let mut pending_batches = VecDeque::from([files]); while let Some(batch) = pending_batches.pop_front() { match download_file_batch(context, peer, &batch).await { Ok(hashes) => completed_hashes.extend(hashes), Err(error) if batch.len() == 1 => { tracing::warn!(path = %batch[0].path, error = ?error, "Could not receive file; deferring dependent changes"); } Err(error) => { tracing::debug!(file_count = batch.len(), error = ?error, "Splitting failed blob request"); // A failed multi-file request does not identify the bad blob. // Halving preserves transfers that are independent of it. let midpoint = batch.len() / 2; let mut first_half = batch; let second_half = first_half.split_off(midpoint); pending_batches.push_back(first_half); pending_batches.push_back(second_half); } } } Ok(completed_hashes)}
async fn download_file_batch( context: &SyncContext<'_>, peer: &iroh::EndpointAddr, files: &[RemoteFile],) -> anyhow::Result<Vec<String>> { for file in files { tracing::debug!(path = %file.path, size_bytes = ?file.size_bytes, "Receiving file"); } let hashes = files .iter() .map(|file| iroh_blobs::Hash::from_str(&file.blob_hash).map_err(anyhow::Error::from)) .collect::<anyhow::Result<Vec<_>>>()?; if let [blob_hash] = hashes.as_slice() { context .node .download_blob(*blob_hash, peer.clone()) .await .with_context(|| format!("could not receive {}", files[0].path))?; } else { context .node .download_blobs(hashes, peer.clone()) .await .with_context(|| format!("could not receive {} files", files.len()))?; } Ok(files.iter().map(|file| file.blob_hash.clone()).collect())}
fn collect_remote_file( files: &mut BTreeMap<String, RemoteFile>, required_blobs: &mut BTreeSet<String>, remote: &Manifest, entry: &Entry,) { if remote.entries.get(&entry.path) != Some(entry) || entry.kind != EntryKind::File { return; } let Some(blob_hash) = entry.blob_hash.clone() else { return; }; required_blobs.insert(blob_hash.clone()); files.entry(blob_hash.clone()).or_insert(RemoteFile { path: entry.path.clone(), blob_hash, size_bytes: entry.size_bytes, });}
async fn write_blob_atomically( node: &NodeHost, blob_hash: iroh_blobs::Hash, destination: &Path,) -> anyhow::Result<()> { let parent = destination .parent() .ok_or_else(|| anyhow::anyhow!("destination has no parent directory"))?; fs::create_dir_all(parent)?; let temporary_file = tempfile::NamedTempFile::new_in(parent)?; node.export_blob(blob_hash, temporary_file.path()).await?; let (_, temporary_path) = temporary_file.keep()?; fs::rename(temporary_path, destination)?; Ok(())}
#[cfg(test)]mod tests { use std::fs;
use tempfile::TempDir; use time::OffsetDateTime; use uuid::Uuid;
use crate::{ domain::{Entry, EntryKind, FolderMode, Manifest, ReconciliationAction}, filesystem::IgnoreRules, storage::FolderConfig, };
use super::{ MAX_SMALL_BLOBS_PER_REQUEST, RemoteFile, plan_download_batches, plan_remote_materialization, remove_existing_destination, replace_non_directory, };
#[test] fn replaces_a_file_with_a_directory() -> anyhow::Result<()> { let directory = TempDir::new()?; let destination = directory.path().join("item"); fs::write(&destination, "file")?;
replace_non_directory(&destination)?; fs::create_dir(&destination)?;
assert!(destination.is_dir()); Ok(()) }
#[test] fn removes_a_directory_before_writing_a_file() -> anyhow::Result<()> { let directory = TempDir::new()?; let destination = directory.path().join("item"); fs::create_dir(&destination)?; fs::write(destination.join("child"), "contents")?;
remove_existing_destination(&destination)?;
assert!(!destination.exists()); Ok(()) }
#[test] fn plans_independent_file_actions_separately() -> anyhow::Result<()> { let directory = TempDir::new()?; let folder_id = Uuid::new_v4(); let first = file_entry("first.txt", "first-hash"); let second = file_entry("second.txt", "second-hash"); let remote = Manifest { folder_id, entries: [ (first.path.clone(), first.clone()), (second.path.clone(), second.clone()), ] .into(), }; let actions = vec![ ReconciliationAction::ApplyRemote { entry: first }, ReconciliationAction::ApplyRemote { entry: second }, ]; let folder = FolderConfig { id: folder_id, name: "test".to_owned(), path: directory.path().to_owned(), capability: "test-capability".to_owned(), mode: FolderMode::SendReceive, };
let (materialization_plan, downloads) = plan_remote_materialization( &IgnoreRules::load(directory.path())?, &folder, actions, &remote, );
assert_eq!(materialization_plan.pending_actions.len(), 2); assert_eq!( materialization_plan.actions_waiting_by_blob["first-hash"], vec![0] ); assert_eq!( materialization_plan.actions_waiting_by_blob["second-hash"], vec![1] ); assert_eq!(downloads.len(), 2); Ok(()) }
#[test] fn batches_tiny_files_without_mixing_in_large_files() { let tiny_files = (0..=MAX_SMALL_BLOBS_PER_REQUEST) .map(|number| RemoteFile { path: format!("tiny-{number}"), blob_hash: format!("hash-{number}"), size_bytes: Some(1), }) .collect(); let downloads = plan_download_batches(tiny_files);
assert_eq!(downloads.len(), 2); assert_eq!(downloads[0].len(), MAX_SMALL_BLOBS_PER_REQUEST); assert_eq!(downloads[1].len(), 1);
let downloads = plan_download_batches(vec![ RemoteFile { path: "tiny".to_owned(), blob_hash: "tiny-hash".to_owned(), size_bytes: Some(1), }, RemoteFile { path: "large".to_owned(), blob_hash: "large-hash".to_owned(), size_bytes: Some(257 * 1024), }, ]);
assert_eq!(downloads.iter().map(Vec::len).collect::<Vec<_>>(), [1, 1]); }
fn file_entry(path: &str, blob_hash: &str) -> Entry { Entry { path: path.to_owned(), kind: EntryKind::File, blob_hash: Some(blob_hash.to_owned()), size_bytes: Some(1), modified_at: OffsetDateTime::UNIX_EPOCH, clock: Default::default(), author_device_id: "device".to_owned(), } }}