Something went wrong. Try again.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455use std::collections::BTreeSet;
use crate::{ app::{AppResult, AppaService, MAX_CONCURRENT_PEER_REQUESTS, SyncContext}, app::{ manifest::{ManifestBuild, build_manifest_with_stages, report_local_changes}, materialize::{ RemoteManifestResult, apply_remote_manifest, recover_pending_materialization, }, }, domain::{AuditEvent, AuditEventKind, FolderRoster, manifest_root_hash}, iroh::{FolderSession, MAX_AUDIT_EVENTS_PER_RESPONSE, NodeHost, PeerSummaryResponse}, storage::FolderConfig,};use futures_util::{StreamExt, stream};
impl AppaService { #[cfg(test)] pub(super) async fn sync_with_node( &self, folder: &FolderConfig, node: &NodeHost, ) -> AppResult<usize> { self.sync_with_node_using_local_scan(folder, node, true) .await }
pub(super) async fn sync_with_node_using_local_scan( &self, folder: &FolderConfig, node: &NodeHost, should_scan_local_files: bool, ) -> AppResult<usize> { let context = SyncContext { folder, node, state_store: &self.state_store, now: time::OffsetDateTime::now_utc(), }; recover_pending_materialization(&context).await?; self.save_lan_discovered_peers(folder, node)?; self.save_discovered_peers(folder, node)?; let local_roster = self.load_roster(folder.id)?; let can_publish_changes = folder.mode.can_send() && local_roster.can_publish(&node.endpoint_address().id.to_string()); self.register_sync_folder( folder, node, &local_roster, self.state_store.load_manifest(folder.id)?, ) .await?; let manifest_build = if should_scan_local_files && can_publish_changes { build_manifest_with_stages(&context, |manifest| async move { self.publish_staged_manifest(folder, node, manifest).await }) .await? } else { ManifestBuild { manifest: self.state_store.load_manifest(folder.id)?, changed_paths: Vec::new(), } }; report_local_changes(&manifest_build.changed_paths); let has_local_changes = !manifest_build.changed_paths.is_empty(); let mut local_manifest = manifest_build.manifest; self.register_sync_folder(folder, node, &local_roster, local_manifest.clone()) .await?; self.persist_published_manifest(node, &local_manifest) .await?; if has_local_changes && can_publish_changes { let report = node .announce_manifest( self.state_store.peers(folder.id)?, folder.id, folder.capability.clone(), manifest_root_hash(&local_manifest)?, ) .await; tracing::debug!(folder = %folder.name, announced_count = report.announced_count, unavailable_count = report.unavailable_count, "Announced local manifest changes"); } let mut synchronized_count = 0; let mut attempted_peer_ids = BTreeSet::new(); loop { let peers: Vec<_> = self .state_store .peers(folder.id)? .into_iter() .filter(|peer| attempted_peer_ids.insert(peer.id)) .collect(); if peers.is_empty() { break; } let peer_summaries = request_peer_summaries( node, peers, folder.id, folder.capability.clone(), self.state_store.join_invite(folder.id)?, ) .await; for (peer, response) in peer_summaries { let Some(remote_result) = self .synchronize_peer(&context, &local_manifest, peer, response) .await? else { continue; }; local_manifest = remote_result.manifest; synchronized_count += remote_result.synchronized_file_count; self.state_store.save_manifest(&local_manifest)?; node.publish_manifest(local_manifest.clone()).await?; } } self.save_discovered_peers(folder, node)?; self.save_lan_discovered_peers(folder, node)?; self.state_store.record_successful_sync(folder.id)?; Ok(synchronized_count) }
async fn publish_staged_manifest( &self, folder: &FolderConfig, node: &NodeHost, manifest: crate::domain::Manifest, ) -> AppResult<()> { self.persist_published_manifest(node, &manifest).await?; let report = node .announce_manifest( self.state_store.peers(folder.id)?, folder.id, folder.capability.clone(), manifest_root_hash(&manifest)?, ) .await; tracing::debug!(folder = %folder.name, announced_count = report.announced_count, unavailable_count = report.unavailable_count, "Announced staged local manifest changes"); Ok(()) }
async fn register_sync_folder( &self, folder: &FolderConfig, node: &NodeHost, roster: &FolderRoster, manifest: crate::domain::Manifest, ) -> AppResult<()> { node.register_folder( FolderSession { folder_id: folder.id, capability: folder.capability.clone(), active_member_ids: roster.active_device_ids(), roster: roster.clone(), peer_endpoints: self.state_store.peers(folder.id)?, audit_events: self.state_store.audit_events(folder.id)?, }, manifest, ) .await }
async fn persist_published_manifest( &self, node: &NodeHost, manifest: &crate::domain::Manifest, ) -> AppResult<()> { let audit_event = self.manifest_audit_event(manifest, time::OffsetDateTime::now_utc())?; self.state_store .save_manifest_with_audit(manifest, &audit_event)?; node.publish_audit_events( manifest.folder_id, self.state_store.audit_events(manifest.folder_id)?, ) .await?; node.publish_manifest(manifest.clone()).await }
fn manifest_audit_event( &self, manifest: &crate::domain::Manifest, now: time::OffsetDateTime, ) -> AppResult<AuditEvent> { let identity = self.paths.load_identity()?; let roster = self.load_roster(manifest.folder_id)?; let author_device_id = identity.public().to_string(); if !roster.can_publish(&author_device_id) { anyhow::bail!("a non-writing member cannot sign a manifest audit event"); } let (sequence, parent_hash) = self .state_store .next_audit_sequence_and_parent(manifest.folder_id, &author_device_id)?; let event = AuditEvent::create( manifest.folder_id, author_device_id, sequence, parent_hash, AuditEventKind::ManifestCommitted, Some(manifest_root_hash(manifest)?), roster.hash()?, roster.clone(), now, ); event.sign(&identity) }
pub(super) fn save_discovered_peers( &self, folder: &FolderConfig, node: &NodeHost, ) -> AppResult<()> { for peer in node.take_discovered_peers() { if peer.folder_id == folder.id { self.state_store.save_peer(folder.id, &peer.endpoint)?; self.enroll_discovered_member(folder.id, &peer.endpoint.id.to_string())?; } } Ok(()) }
fn save_lan_discovered_peers(&self, folder: &FolderConfig, node: &NodeHost) -> AppResult<()> { let active_member_ids = self.active_member_ids(folder.id)?; for peer in node.lan_discovered_peers() { let device_id = peer.id.to_string(); if active_member_ids.contains(&device_id) { tracing::debug!(folder = %folder.name, peer = %device_id, "Refreshed trusted local peer route"); self.state_store.save_peer(folder.id, &peer)?; } else { tracing::debug!(folder = %folder.name, peer = %device_id, "Ignoring untrusted local peer"); } } Ok(()) }
async fn save_peer_response( &self, folder: &FolderConfig, node: &NodeHost, peer: &iroh::EndpointAddr, roster: &FolderRoster, peer_endpoints: Vec<iroh::EndpointAddr>, capability: &str, ) -> AppResult<()> { if roster.folder_id != folder.id { anyhow::bail!("peer returned a roster for another folder"); } if roster.capability != capability { anyhow::bail!("peer capability is not bound to its signed roster"); } let current_roster = self.load_roster(folder.id)?; if !current_roster.contains_member(&peer.id.to_string()) { anyhow::bail!("peer is not an active member of the local roster"); } if roster.owner_device_id != current_roster.owner_device_id { anyhow::bail!("peer returned a roster with another owner"); } if roster.epoch < current_roster.epoch { self.state_store.save_peer(folder.id, peer)?; self.save_roster_peers(folder.id, ¤t_roster, peer_endpoints)?; return self.publish_peer_endpoints(folder.id, node).await; } if roster.epoch == current_roster.epoch && roster != ¤t_roster { anyhow::bail!("peer returned a conflicting folder roster epoch"); } self.state_store.save_peer(folder.id, peer)?; let accepted_new_roster = self.state_store.save_roster_if_newer(roster)?; self.save_roster_peers(folder.id, roster, peer_endpoints)?; self.publish_peer_endpoints(folder.id, node).await?; if accepted_new_roster && capability != folder.capability { self.state_store.update_capability(folder.id, capability)?; tracing::info!(folder = %folder.name, "Updated folder capability after member revocation"); } Ok(()) }
async fn publish_peer_endpoints( &self, folder_id: crate::domain::FolderId, node: &NodeHost, ) -> AppResult<()> { node.publish_peer_endpoints(folder_id, self.state_store.peers(folder_id)?) .await }
fn save_roster_peers( &self, folder_id: crate::domain::FolderId, roster: &FolderRoster, peer_endpoints: Vec<iroh::EndpointAddr>, ) -> AppResult<()> { for peer in peer_endpoints { if roster.contains_member(&peer.id.to_string()) { self.state_store.save_peer(folder_id, &peer)?; } } Ok(()) }
async fn synchronize_peer( &self, context: &SyncContext<'_>, local_manifest: &crate::domain::Manifest, peer: iroh::EndpointAddr, response: anyhow::Result<PeerSummaryResponse>, ) -> AppResult<Option<RemoteManifestResult>> { let folder = context.folder; let PeerSummaryResponse { summary, roster, peer_endpoints, capability, } = match response { Ok(response) => response, Err(error) => { tracing::debug!(folder = %folder.name, peer = %peer.id, %error, "peer is unavailable; continuing with other peers"); return Ok(None); } }; if summary.folder_id != folder.id { tracing::warn!(folder = %folder.name, peer = %peer.id, "peer returned a summary for another folder"); return Ok(None); } self.save_peer_response( folder, context.node, &peer, &roster, peer_endpoints, &capability, ) .await?; self.receive_peer_audit_events(context, &peer).await?; if summary.root_hash == manifest_root_hash(local_manifest)? { tracing::debug!(folder = %folder.name, peer = %peer.id, "Folder roots match; skipping manifest transfer"); return Ok(None); } let remote_manifest = match context .node .request_manifest_delta( peer.clone(), folder.id, folder.capability.clone(), self.state_store.join_invite(folder.id)?, local_manifest, ) .await { Ok(manifest) => manifest, Err(error) => { tracing::debug!(folder = %folder.name, peer = %peer.id, %error, "peer did not provide its changed entries"); return Ok(None); } }; if remote_manifest.folder_id != folder.id { tracing::warn!(folder = %folder.name, peer = %peer.id, "peer returned a manifest for another folder"); return Ok(None); } if !folder.mode.can_receive() { tracing::debug!(folder = %folder.name, "Ignoring remote changes for send-only folder"); return Ok(None); } match apply_remote_manifest(context, local_manifest, &remote_manifest, peer.clone()).await { Ok(result) => Ok(Some(result)), Err(error) => { tracing::warn!(folder = %folder.name, peer = %peer.id, %error, "could not apply peer manifest"); Ok(None) } } }
async fn receive_peer_audit_events( &self, context: &SyncContext<'_>, peer: &iroh::EndpointAddr, ) -> AppResult<()> { loop { let events = match context .node .request_audit_events( peer.clone(), context.folder.id, context.folder.capability.clone(), self.state_store.join_invite(context.folder.id)?, self.state_store .audit_heads(context.folder.id)? .into_iter() .map(|head| (head.author_device_id, head.sequence)) .collect(), ) .await { Ok(events) => events, Err(error) => { tracing::debug!(folder = %context.folder.name, peer = %peer.id, %error, "peer did not provide audit events"); return Ok(()); } }; let is_last_page = events.len() < MAX_AUDIT_EVENTS_PER_RESPONSE; for event in events { if event.folder_id != context.folder.id { anyhow::bail!("peer returned an audit event for another folder"); } if let Err(error) = self.state_store.append_audit_event(&event) { let detail = format!( "rejected audit event from {}: {error}", event.author_device_id ); self.state_store.record_audit_fault( context.folder.id, &event.author_device_id, &detail, )?; anyhow::bail!(detail); } } if is_last_page { break; } } context .node .publish_audit_events( context.folder.id, self.state_store.audit_events(context.folder.id)?, ) .await }}
async fn request_peer_summaries( node: &NodeHost, peers: Vec<iroh::EndpointAddr>, folder_id: crate::domain::FolderId, capability: String, invite: Option<crate::protocol::Invite>,) -> Vec<(iroh::EndpointAddr, anyhow::Result<PeerSummaryResponse>)> { stream::iter(peers) .map(|peer| { let request_capability = capability.clone(); let request_invite = invite.clone(); async move { let response = node .request_manifest_summary( peer.clone(), folder_id, request_capability, request_invite, ) .await; (peer, response) } }) .buffer_unordered(MAX_CONCURRENT_PEER_REQUESTS) .collect() .await}