From 93998dbeee76bac073703751b0be728b48ad960a Mon Sep 17 00:00:00 2001 From: Aly Raffauf Date: Mon, 03 Aug 2026 04:53:41 +0000 Subject: [PATCH] Materialize files as blobs arrive --- src/app/materialize.rs | 269 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++----------------------------------------------------- src/app/tests.rs | 31 +++++++++++++++++++++++++++++-- 2 file(s) changed, 245 insertion(s)(+), 55 deletion(s)(-) diff --git a/src/app/materialize.rs b/src/app/materialize.rs --- a/src/app/materialize.rs +++ b/src/app/materialize.rs @@ -1,6 +1,12 @@ -use std::{collections::BTreeMap, fs, path::Path, str::FromStr}; +use std::{ + collections::{BTreeMap, BTreeSet}, + fs, + path::Path, + str::FromStr, +}; -use futures_util::{StreamExt, TryStreamExt, stream}; +use anyhow::Context; +use futures_util::{StreamExt, stream}; use time::OffsetDateTime; use crate::{ @@ -21,27 +27,27 @@ let mut synchronized_count = 0; let ignore_rules = IgnoreRules::load(&context.folder.path)?; let actions = crate::domain::reconcile_manifests(local, remote); - download_remote_files(context, &ignore_rules, &actions, remote, &peer).await?; - for action in actions { - if action_is_ignored(context.folder, &ignore_rules, &action) { - continue; - } - match action { - ReconciliationAction::ApplyRemote { entry } => { - synchronized_count += materialize_entry(context, &mut merged, entry).await?; - } - ReconciliationAction::KeepLocal { .. } => {} - ReconciliationAction::ResolveConflict { - winner, - loser, - conflict_path, - } => { - synchronized_count += - materialize_conflict(context, &mut merged, &winner, &loser, &conflict_path) - .await?; - } - } - } + let (mut pending_actions, downloads) = + plan_remote_materialization(&ignore_rules, context.folder, actions, remote); + let mut available_blobs = BTreeSet::new(); + materialize_ready_actions( + context, + &mut merged, + &mut synchronized_count, + &mut pending_actions, + &available_blobs, + ) + .await?; + receive_and_materialize_files( + context, + &peer, + &mut merged, + &mut synchronized_count, + &mut pending_actions, + &mut available_blobs, + downloads, + ) + .await?; Ok((merged, synchronized_count)) } @@ -236,48 +242,139 @@ size_bytes: Option, } -async fn download_remote_files( - context: &SyncContext<'_>, +struct PendingAction { + action: ReconciliationAction, + required_blobs: BTreeSet, +} + +fn plan_remote_materialization( ignore_rules: &IgnoreRules, - actions: &[ReconciliationAction], + folder: &FolderConfig, + actions: Vec, remote: &Manifest, - peer: &iroh::EndpointAddr, -) -> anyhow::Result<()> { +) -> (Vec, Vec) { let mut files = BTreeMap::new(); - for action in actions { - if action_is_ignored(context.folder, ignore_rules, action) { - continue; - } - match action { - ReconciliationAction::ApplyRemote { entry } => { - collect_remote_file(&mut files, remote, entry); - } - ReconciliationAction::KeepLocal { .. } => {} - ReconciliationAction::ResolveConflict { winner, loser, .. } => { - collect_remote_file(&mut files, remote, winner); - collect_remote_file(&mut files, remote, loser); - } - } - } - stream::iter(files.into_values()) - .map(|file| async move { - let blob_hash = iroh_blobs::Hash::from_str(&file.blob_hash)?; - tracing::info!(path = %file.path, size_bytes = ?file.size_bytes, "Receiving file"); - context.node.download_blob(blob_hash, peer.clone()).await + let pending_actions = actions + .into_iter() + .filter(|action| !action_is_ignored(folder, ignore_rules, action)) + .map(|action| PendingAction { + required_blobs: collect_required_blobs(&mut files, remote, &action), + action, }) - .buffer_unordered(MAX_CONCURRENT_BLOB_TRANSFERS) - .try_collect::>() - .await?; + .collect(); + (pending_actions, files.into_values().collect()) +} + +fn collect_required_blobs( + files: &mut BTreeMap, + remote: &Manifest, + action: &ReconciliationAction, +) -> BTreeSet { + 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, + pending_actions: &mut Vec, + available_blobs: &BTreeSet, +) -> anyhow::Result<()> { + while let Some(position) = pending_actions + .iter() + .position(|pending| pending.required_blobs.is_subset(available_blobs)) + { + let pending = pending_actions.remove(position); + *synchronized_count += materialize_action(context, merged, pending.action).await?; + } Ok(()) } -fn collect_remote_file(files: &mut BTreeMap, remote: &Manifest, entry: &Entry) { +async fn materialize_action( + context: &SyncContext<'_>, + merged: &mut Manifest, + action: ReconciliationAction, +) -> anyhow::Result { + match action { + ReconciliationAction::ApplyRemote { entry } => { + materialize_entry(context, merged, entry).await + } + ReconciliationAction::KeepLocal { .. } => Ok(0), + ReconciliationAction::ResolveConflict { + winner, + loser, + conflict_path, + } => materialize_conflict(context, merged, &winner, &loser, &conflict_path).await, + } +} + +async fn receive_and_materialize_files( + context: &SyncContext<'_>, + peer: &iroh::EndpointAddr, + merged: &mut Manifest, + synchronized_count: &mut usize, + pending_actions: &mut Vec, + available_blobs: &mut BTreeSet, + downloads: Vec, +) -> anyhow::Result<()> { + let mut downloads = stream::iter(downloads) + .map(|file| async move { + let blob_hash = iroh_blobs::Hash::from_str(&file.blob_hash)?; + tracing::info!(path = %file.path, size_bytes = ?file.size_bytes, "Receiving file"); + context + .node + .download_blob(blob_hash, peer.clone()) + .await + .with_context(|| format!("could not receive {}", file.path))?; + Ok::<_, anyhow::Error>(file.blob_hash) + }) + .buffer_unordered(MAX_CONCURRENT_BLOB_TRANSFERS); + while let Some(result) = downloads.next().await { + match result { + Ok(blob_hash) => { + available_blobs.insert(blob_hash); + materialize_ready_actions( + context, + merged, + synchronized_count, + pending_actions, + available_blobs, + ) + .await?; + } + Err(error) => { + tracing::warn!(error = ?error, "Could not receive file; deferring dependent changes") + } + } + } + Ok(()) +} + +fn collect_remote_file( + files: &mut BTreeMap, + required_blobs: &mut BTreeSet, + 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, @@ -306,8 +403,16 @@ use std::fs; use tempfile::TempDir; + use time::OffsetDateTime; + use uuid::Uuid; - use super::{remove_existing_destination, replace_non_directory}; + use crate::{ + domain::{Entry, EntryKind, FolderMode, Manifest, ReconciliationAction}, + filesystem::IgnoreRules, + storage::FolderConfig, + }; + + use super::{plan_remote_materialization, remove_existing_destination, replace_non_directory}; #[test] fn replaces_a_file_with_a_directory() -> anyhow::Result<()> { @@ -333,5 +438,63 @@ 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 (pending_actions, downloads) = plan_remote_materialization( + &IgnoreRules::load(directory.path())?, + &folder, + actions, + &remote, + ); + + assert_eq!(pending_actions.len(), 2); + assert_eq!( + pending_actions[0].required_blobs, + ["first-hash".to_owned()].into() + ); + assert_eq!( + pending_actions[1].required_blobs, + ["second-hash".to_owned()].into() + ); + assert_eq!(downloads.len(), 2); + Ok(()) + } + + 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(), + } } } diff --git a/src/app/tests.rs b/src/app/tests.rs --- a/src/app/tests.rs +++ b/src/app/tests.rs @@ -1,7 +1,8 @@ -use std::fs; +use std::{fs, path::Path}; use tempfile::TempDir; use time::{Duration, OffsetDateTime}; +use tokio::time::sleep; use crate::{ app::manifest::build_manifest, @@ -10,6 +11,9 @@ }; use super::{AppPaths, AppaService, ConfigAuditAction, Invite, PROTOCOL_VERSION}; + +const INITIAL_SYNC_ATTEMPTS: usize = 3; +const INITIAL_SYNC_RETRY_DELAY: std::time::Duration = std::time::Duration::from_millis(100); #[test] fn bounds_exponential_retry_delays() { @@ -156,7 +160,10 @@ .state_store .import_invite(target_folder.path(), &invite)?; - assert_eq!(target.sync_once(target_folder.path()).await?, 1); + assert_eq!( + sync_until_file_materializes(&target, target_folder.path(), &target_file).await?, + 1 + ); assert_eq!(fs::read_to_string(&target_file)?, "shared from Appa"); assert!(target_empty_directory.is_dir()); source.save_discovered_peers(&source_config, &source_node)?; @@ -178,6 +185,26 @@ assert!(!target_empty_directory.exists()); source_node.shutdown().await?; Ok(()) +} + +async fn sync_until_file_materializes( + appa: &AppaService, + folder: &Path, + expected_file: &Path, +) -> anyhow::Result { + for attempt in 0..INITIAL_SYNC_ATTEMPTS { + let synchronized_files = appa.sync_once(folder).await?; + if expected_file.exists() { + return Ok(synchronized_files); + } + if attempt + 1 < INITIAL_SYNC_ATTEMPTS { + sleep(INITIAL_SYNC_RETRY_DELAY).await; + } + } + anyhow::bail!( + "{} did not materialize after {INITIAL_SYNC_ATTEMPTS} sync attempts", + expected_file.display() + ) } #[tokio::test] -- tangled.sh