From 876298a2bfdffbfed16c3d40d8b7c737172259e9 Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Sun, 30 Aug 2026 15:04:43 +0200 Subject: [PATCH] sync: delete only after a complete scan proves absence --- CHANGELOG.md | 18 +- README.md | 8 + src/control.rs | 3 + src/daemon.rs | 1 + src/doctor.rs | 123 +++++++- src/main.rs | 32 +- src/sync/engine.rs | 699 +++++++++++++++++++++++++++++++++++++++---- src/sync/wire.rs | 74 ++++- tests/folder_sync.rs | 31 ++ 9 files changed, 912 insertions(+), 77 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c7fd63c..5a6eb37 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -269,6 +269,21 @@ EXPERIMENTAL, so on-disk formats and the CLI may change without notice. ### Fixed +- **A delete now requires affirmative absence.** A scan distinguishes a present + file, a path absent from a completely read parent directory, and a path whose + state is unknown. Only the second state becomes a tombstone. An unreadable + file or directory no longer stops the whole entry, and skipping it cannot + turn it into a delete. + + A file over 512 MiB is present but not syncable. Fabric does not read or hash + it, does not overwrite it during materialization, and reports its path under + `scan_issues`. If the file is later deleted, a complete parent scan still + proves that delete and propagates it normally. + + `fabric doctor` also distinguishes a missing remote sync entry and residual + size-limit errors from an unreachable peer. These states need a configuration + or file change; waiting for the network cannot fix them. + - **Include globs are now a receive-side boundary, not only a scan-side one.** A node adopted every winning entry a peer sent, whatever its own include said, so a host with a broad include (or a mistaken `["**"]`) had its machine-local @@ -379,9 +394,6 @@ EXPERIMENTAL, so on-disk formats and the CLI may change without notice. file. `fabric sync ls` now prints `content_bytes`, which is the number that would have said so on 19 August. - Not changed here: a file over 512 MiB is still read into memory and refused by - every peer, and every present file's bytes are still held once. Both are - separate changes. - **A dial to a peer that cannot be reached no longer holds its permit after the consumer leaves.** Every local connection to a dial socket, and every `shell` and `exec`, holds one of 32 dial permits for the life of its session. A session diff --git a/README.md b/README.md index a464991..2aab2a6 100644 --- a/README.md +++ b/README.md @@ -1009,6 +1009,14 @@ last durable local-disk receipt). `drift=clean` means the logical Present paths and observed bytes agree. A `drift=WARNING` names `missing` Present paths, `unexpected` observed paths whose manifest is tombstoned or absent, and `mismatched` paths whose observed content hash differs from the logical Present. +`scan_issues` names existing paths that the last scan could not read as syncable +regular files. + +A delete propagates only when a complete parent directory listing proves the +path is absent. An unreadable path remains present with an unknown state. A file +over 512 MiB is also present but not syncable: fabric does not read, hash, +overwrite, or send it. Reduce it to 512 MiB or less, or exclude it from the sync +entry. If it is later deleted, the next complete scan propagates that delete. The per-entry `full_scans`, `inbound_noop_transactions`, and `inbound_guarded_transactions` counters are monotonic while that name remains continuously configured in the same daemon process. They let operators measure diff --git a/src/control.rs b/src/control.rs index ed1d94f..634d5a8 100644 --- a/src/control.rs +++ b/src/control.rs @@ -194,6 +194,9 @@ pub struct SyncEntryStatus { pub unexpected: usize, #[serde(default)] pub mismatched: usize, + /// Existing paths the last scan could not read as syncable files. + #[serde(default)] + pub scan_issues: Vec<(String, String)>, /// Monotonic full-folder scan attempts for this entry instance. #[serde(default)] pub full_scans: u64, diff --git a/src/daemon.rs b/src/daemon.rs index bc7dcb8..bbe3f8e 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -2978,6 +2978,7 @@ async fn process_control_request( missing: status.missing, unexpected: status.unexpected, mismatched: status.mismatched, + scan_issues: status.scan_issues, full_scans: status.full_scans, inbound_noop_transactions: status.inbound_noop_transactions, inbound_guarded_transactions: status.inbound_guarded_transactions, diff --git a/src/doctor.rs b/src/doctor.rs index 525aad6..d72f057 100644 --- a/src/doctor.rs +++ b/src/doctor.rs @@ -109,6 +109,8 @@ pub struct SyncFact { pub folder: PathBuf, pub folder_exists: bool, pub drift_clean: bool, + /// Paths the last scan saw but could not read as syncable files. + pub scan_issues: Vec<(String, String)>, /// Peers this entry has stopped syncing with, and why. pub stopped: Vec<(String, String)>, } @@ -390,6 +392,31 @@ fn sync_findings(sync: &SyncFact) -> Vec { return out; } + for (path, reason) in &sync.scan_issues { + let finding = if reason == "too-large" { + Finding::new( + "sync", + Verdict::Problem, + format!( + "{name} cannot sync {path}: the file exceeds 512 MiB. fabric still treats it as present" + ), + ) + .with_action(format!( + "reduce {path} to 512 MiB or less, or exclude it from {name}" + )) + } else { + Finding::new( + "sync", + Verdict::Unknown, + format!( + "{name} cannot read {path} as a syncable file. fabric will not treat it as deleted" + ), + ) + .with_action(format!("make {path} a readable regular file, or exclude it from {name}")) + }; + out.push(finding); + } + for (peer, reason) in &sync.stopped { // Denied waits for a person; unreachable waits for the network. Telling // them apart is the difference between a chore and weather. @@ -427,6 +454,26 @@ fn sync_findings(sync: &SyncFact) -> Vec { .with_action(format!( "on {peer}, add `sync` to this machine's allow list in peers.toml" )) + } else if reason == "missing-entry" { + Finding::new( + "sync", + Verdict::Problem, + format!( + "{name} has stopped syncing with {peer}: {peer} has no local sync entry named {name}" + ), + ) + .with_action(format!( + "on {peer}, add {name} to syncs.toml or fix the shared entry name" + )) + } else if reason == "too-large" { + Finding::new( + "sync", + Verdict::Problem, + format!("{name} has stopped syncing with {peer}: sync content exceeds 512 MiB"), + ) + .with_action(format!( + "reduce the large file to 512 MiB or less, or exclude it from {name}" + )) } else { Finding::new( "sync", @@ -446,7 +493,7 @@ fn sync_findings(sync: &SyncFact) -> Vec { )); } - if sync.stopped.is_empty() && sync.drift_clean { + if sync.stopped.is_empty() && sync.scan_issues.is_empty() && sync.drift_clean { out.push(Finding::new( "sync", Verdict::Ok, @@ -580,6 +627,7 @@ mod tests { folder: PathBuf::from("/tmp/bus"), folder_exists: true, drift_clean: true, + scan_issues: Vec::new(), stopped: Vec::new(), }], ca: CaFact { @@ -859,6 +907,74 @@ mod tests { ); } + #[test] + fn local_sync_faults_do_not_read_as_unreachable() { + let mut facts = configured(); + facts.syncs[0].stopped = vec![ + ("droppy".to_string(), "missing-entry".to_string()), + ("hetz".to_string(), "too-large".to_string()), + ]; + let findings = diagnose(&facts); + let syncs = find(&findings, "sync"); + + for peer in ["droppy", "hetz"] { + let finding = syncs + .iter() + .find(|finding| finding.detail.contains(peer)) + .expect("no finding for the local sync fault"); + assert!(!finding.detail.contains("unreachable")); + assert!( + !finding + .action + .as_deref() + .unwrap_or_default() + .contains("comes back") + ); + } + assert!( + syncs + .iter() + .find(|finding| finding.detail.contains("droppy")) + .unwrap() + .action + .as_deref() + .is_some_and(|action| action.contains("syncs.toml")) + ); + assert!( + syncs + .iter() + .find(|finding| finding.detail.contains("hetz")) + .unwrap() + .action + .as_deref() + .is_some_and(|action| action.contains("512 MiB")) + ); + } + + #[test] + fn an_oversized_local_path_is_named_and_is_not_called_clean() { + let mut facts = configured(); + facts.syncs[0].scan_issues = vec![("archive.bin".into(), "too-large".into())]; + let findings = diagnose(&facts); + let syncs = find(&findings, "sync"); + let issue = syncs + .iter() + .find(|finding| finding.detail.contains("archive.bin")) + .expect("the scan issue did not name its path"); + assert!(issue.detail.contains("still treats it as present")); + assert!( + issue + .action + .as_deref() + .is_some_and(|action| action.contains("512 MiB")) + ); + assert!( + !syncs + .iter() + .any(|finding| finding.detail.contains("clean and syncing")) + ); + } + /// The engine treats a missing folder as "wait", deliberately, so nothing /// else will ever mention it. #[test] @@ -1115,7 +1231,10 @@ where name: entry.name.clone(), folder_exists: folder.exists(), folder, - drift_clean: entry.missing == 0 && entry.mismatched == 0, + drift_clean: entry.missing == 0 + && entry.mismatched == 0 + && entry.scan_issues.is_empty(), + scan_issues: entry.scan_issues.clone(), stopped: entry.stopped_peers.clone(), } }) diff --git a/src/main.rs b/src/main.rs index 6b66dce..4b47faf 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1133,9 +1133,13 @@ async fn run_sync(home: &FabricHome, command: SyncCommands) -> Result<()> { } for entry in entries { let present = logical_present(&entry); - if entry.missing == 0 && entry.unexpected == 0 && entry.mismatched == 0 { + if entry.missing == 0 + && entry.unexpected == 0 + && entry.mismatched == 0 + && entry.scan_issues.is_empty() + { println!( - "{}\t{}\t{}\tpeers={}\tpresent={present}\ttombstones={}\tobserved={}\tdrift=clean\tstopped={}\tsync_passes={}\tfull_scans={}\tinbound_noop_transactions={}\tinbound_guarded_transactions={}\tscan_ms={}\tmaterialize_ms={}\tpersist_ms={}\treconcile_ms={}\treconcile_wire_bytes={}\treconcile_failures={}\tsweep={}\tdelta_fallbacks={}\tfull_payload_sends={}\tcontent_bytes={}\tdigest={}", + "{}\t{}\t{}\tpeers={}\tpresent={present}\ttombstones={}\tobserved={}\tdrift=clean\tscan_issues=none\tstopped={}\tsync_passes={}\tfull_scans={}\tinbound_noop_transactions={}\tinbound_guarded_transactions={}\tscan_ms={}\tmaterialize_ms={}\tpersist_ms={}\treconcile_ms={}\treconcile_wire_bytes={}\treconcile_failures={}\tsweep={}\tdelta_fallbacks={}\tfull_payload_sends={}\tcontent_bytes={}\tdigest={}", entry.name, entry.folder, entry.policy, @@ -1161,7 +1165,7 @@ async fn run_sync(home: &FabricHome, command: SyncCommands) -> Result<()> { ); } else { println!( - "{}\t{}\t{}\tpeers={}\tpresent={present}\ttombstones={}\tobserved={}\tdrift=WARNING missing={} unexpected={} mismatched={}\tstopped={}\tsync_passes={}\tfull_scans={}\tinbound_noop_transactions={}\tinbound_guarded_transactions={}\tscan_ms={}\tmaterialize_ms={}\tpersist_ms={}\treconcile_ms={}\treconcile_wire_bytes={}\treconcile_failures={}\tsweep={}\tdelta_fallbacks={}\tfull_payload_sends={}\tcontent_bytes={}\tdigest={}", + "{}\t{}\t{}\tpeers={}\tpresent={present}\ttombstones={}\tobserved={}\tdrift=WARNING missing={} unexpected={} mismatched={}\tscan_issues={}\tstopped={}\tsync_passes={}\tfull_scans={}\tinbound_noop_transactions={}\tinbound_guarded_transactions={}\tscan_ms={}\tmaterialize_ms={}\tpersist_ms={}\treconcile_ms={}\treconcile_wire_bytes={}\treconcile_failures={}\tsweep={}\tdelta_fallbacks={}\tfull_payload_sends={}\tcontent_bytes={}\tdigest={}", entry.name, entry.folder, entry.policy, @@ -1171,6 +1175,7 @@ async fn run_sync(home: &FabricHome, command: SyncCommands) -> Result<()> { entry.missing, entry.unexpected, entry.mismatched, + scan_issues_token(&entry), stopped_token(&entry), entry.sync_passes, entry.full_scans, @@ -1223,6 +1228,7 @@ struct SyncLsJsonEntry<'a> { missing: usize, unexpected: usize, mismatched: usize, + scan_issues: &'a [(String, String)], /// Calls to `sync_once`. NOT `full_scans`, which is two per call. sync_passes: u64, full_scans: u64, @@ -1267,10 +1273,14 @@ impl<'a> From<&'a fabric::control::SyncEntryStatus> for SyncLsJsonEntry<'a> { present: logical_present(entry), tombstones: entry.tombstones, observed: entry.observed, - drift: entry.missing != 0 || entry.unexpected != 0 || entry.mismatched != 0, + drift: entry.missing != 0 + || entry.unexpected != 0 + || entry.mismatched != 0 + || !entry.scan_issues.is_empty(), missing: entry.missing, unexpected: entry.unexpected, mismatched: entry.mismatched, + scan_issues: &entry.scan_issues, stopped_peers: entry .stopped_peers .iter() @@ -1462,6 +1472,18 @@ fn stopped_token(entry: &fabric::control::SyncEntryStatus) -> String { .join(",") } +fn scan_issues_token(entry: &fabric::control::SyncEntryStatus) -> String { + if entry.scan_issues.is_empty() { + return "none".to_string(); + } + entry + .scan_issues + .iter() + .map(|(path, reason)| format!("{path}:{reason}")) + .collect::>() + .join(",") +} + fn sweep_token(entry: &fabric::control::SyncEntryStatus) -> &str { if entry.sweep.is_empty() { "unknown" @@ -1697,6 +1719,7 @@ mod sync_ls_tests { missing: 0, unexpected: 2, mismatched: 0, + scan_issues: vec![("large.bin".into(), "too-large".into())], sync_passes: 9, full_scans: 17, inbound_noop_transactions: 11, @@ -1729,6 +1752,7 @@ mod sync_ls_tests { "missing": 0, "unexpected": 2, "mismatched": 0, + "scan_issues": [["large.bin", "too-large"]], "full_scans": 17, "inbound_noop_transactions": 11, "inbound_guarded_transactions": 3, diff --git a/src/sync/engine.rs b/src/sync/engine.rs index bb4a6a9..606f5e3 100644 --- a/src/sync/engine.rs +++ b/src/sync/engine.rs @@ -16,9 +16,9 @@ //! versions stay monotonic across daemon restarts. use std::{ - collections::BTreeMap, - collections::{HashMap, HashSet, VecDeque}, + collections::{BTreeMap, HashMap, HashSet, VecDeque}, future::Future, + io::Read, path::{Path, PathBuf}, sync::{ Arc, Mutex as StdMutex, @@ -40,6 +40,7 @@ use crate::config::FabricHome; use super::config::{PolicyRules, SyncBook, SyncEntry, SyncPeers}; use super::manifest::{Author, ContentHash, Entry, FileMeta, Manifest}; use super::node::{Reconciled, SweepEvidence, SyncNode, content_hash}; +use super::wire::MAX_BLOB; /// How long to wait after a filesystem event settles before syncing, so a burst /// of writes coalesces into one reconcile. @@ -509,6 +510,9 @@ struct EntryState { observed: Arc>>, /// Local hash cache, keyed on this machine's own disk facts. Never sent. scan_cache: Arc>>, + /// Paths the last scan saw but could not read as syncable regular files. + /// Never sent and never used as deletion evidence. + scan_issues: Arc>>, /// Local wall-clock time of the last reconcile this node completed with /// each peer, keyed by peer id. Never sent; it is this node's own evidence /// of what a peer has been told, and the tombstone sweep will not forget a @@ -611,6 +615,10 @@ pub enum PeerSyncState { /// The selector names no peer in `peers.toml`. Not a refusal and not the /// network: a name nobody answers to, which a person has to fix in a file. Unknown, + /// The remote daemon has no configured entry with this shared name. + MissingEntry, + /// Local or remote sync data exceeds a wire limit. + TooLarge, } impl PeerSyncState { @@ -620,10 +628,26 @@ impl PeerSyncState { PeerSyncState::Refused => "denied", PeerSyncState::Unreachable => "unreachable", PeerSyncState::Unknown => "unknown", + PeerSyncState::MissingEntry => "missing-entry", + PeerSyncState::TooLarge => "too-large", } } } +fn classify_reconcile_error(message: &str) -> PeerSyncState { + if crate::config::Denied::is_refusal(message) { + PeerSyncState::Refused + } else if message.contains("no local sync entry named") { + PeerSyncState::MissingEntry + } else if message.contains("sync") + && (message.contains("exceeds limit") || message.contains("-byte limit")) + { + PeerSyncState::TooLarge + } else { + PeerSyncState::Unreachable + } +} + /// One path's durable state, as appended to the log. /// /// The manifest entry and the observed receipt travel TOGETHER because they must @@ -731,13 +755,14 @@ impl SyncEngine { .unwrap_or_else(EntryWork::new); // Reuse an existing node for an unchanged entry so in-memory content // survives a reload; otherwise start one from the persisted manifest. - let (node, operation, observed, scan_cache, peer_acks, expired_since) = + let (node, operation, observed, scan_cache, scan_issues, peer_acks, expired_since) = match entries.get(&cfg.name) { Some(existing) if existing.config == *cfg => ( existing.node.clone(), existing.operation.clone(), existing.observed.clone(), existing.scan_cache.clone(), + existing.scan_issues.clone(), existing.peer_acks.clone(), existing.expired_since.clone(), ), @@ -751,6 +776,7 @@ impl SyncEngine { Arc::new(Mutex::new(())), Arc::new(StdMutex::new(observed)), Arc::new(StdMutex::new(scan_cache)), + Arc::new(StdMutex::new(BTreeMap::new())), Arc::new(StdMutex::new(peer_acks)), Arc::new(StdMutex::new(HashMap::new())), ) @@ -771,6 +797,7 @@ impl SyncEngine { persisted_observed, observed, scan_cache, + scan_issues, peer_acks, expired_since, last_sweep: Arc::new(StdMutex::new(None)), @@ -1009,6 +1036,13 @@ impl SyncEngine { .is_some_and(|hash| hash != &meta.hash) }) .count(); + let scan_issues = entry + .scan_issues + .lock() + .unwrap() + .iter() + .map(|(path, issue)| (path.clone(), issue.token().to_string())) + .collect(); out.push(SyncStatus { digest: manifest.digest(), name: name.clone(), @@ -1021,6 +1055,7 @@ impl SyncEngine { missing, unexpected, mismatched, + scan_issues, full_scans: entry.work.full_scans.load(Ordering::Relaxed), inbound_noop_transactions: entry .work @@ -1168,11 +1203,7 @@ impl SyncEngine { Err(error) => { // Refused and unreachable are different answers. One waits // for a person, the other waits for the network. - let state = if crate::config::Denied::is_refusal(&format!("{error:#}")) { - PeerSyncState::Refused - } else { - PeerSyncState::Unreachable - }; + let state = classify_reconcile_error(&format!("{error:#}")); entry .peer_state .lock() @@ -1251,7 +1282,17 @@ impl SyncEngine { let mut node = entry.node.lock().await; let mut observed = entry.observed.lock().unwrap(); let mut cache = entry.scan_cache.lock().unwrap(); - scan_into_node_observed(&mut node, &root, &cfg, policy, &mut observed, &mut cache) + let mut issues = entry.scan_issues.lock().unwrap(); + scan_into_node_observed_with_limit( + &mut node, + &root, + &cfg, + policy, + &mut observed, + &mut cache, + &mut issues, + MAX_BLOB as u64, + ) } async fn materialize_entry_state( @@ -1264,9 +1305,10 @@ impl SyncEngine { let generation = entry.work.mutation_generation.load(Ordering::Acquire); let mut node = entry.node.lock().await; let mut observed = entry.observed.lock().unwrap(); - // Same lock order as `scan_entry`: node, then observed, then cache. + // Same lock order as `scan_entry`: node, observed, cache, then issues. let cache = entry.scan_cache.lock().unwrap(); - materialize_tracked( + let issues = entry.scan_issues.lock().unwrap(); + materialize_tracked_with_issues( &mut node, &root, &entry.config, @@ -1274,6 +1316,7 @@ impl SyncEngine { protected, &mut observed, &cache, + &issues, Some((&entry.work, generation)), ) } @@ -1994,6 +2037,9 @@ pub struct SyncStatus { pub missing: usize, pub unexpected: usize, pub mismatched: usize, + /// Paths that exist but the last scan could not read as syncable files. + /// Each pair is `path:reason`, where reason is a stable short token. + pub scan_issues: Vec<(String, String)>, /// Completed or attempted full folder scans since this entry instance was /// loaded. Monotonic while the name remains continuously configured in the /// same daemon. @@ -2041,6 +2087,167 @@ pub struct SyncStatus { // ---- filesystem scan / materialize (sync helpers, unit-testable) ---- +#[derive(Debug, Clone, PartialEq, Eq)] +enum ScanIssue { + TooLarge, + Unreadable, + Unsupported, +} + +impl ScanIssue { + fn token(&self) -> &'static str { + match self { + Self::TooLarge => "too-large", + Self::Unreadable => "unreadable", + Self::Unsupported => "unsupported", + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum ScanPathState { + Present, + AffirmativelyGone, + Unknown, +} + +/// One folder walk and the evidence it can give about every tracked path. +/// +/// A path is gone only when a complete parent listing proves that its next +/// component is absent. An unreadable directory, file, or unsupported object +/// is present but unknown. This distinction lets the walk continue without +/// turning a skipped path into a deletion. +#[derive(Default)] +struct FolderScan { + files: Vec, + present_paths: HashSet, + seen_paths: HashSet, + complete_dirs: HashSet, + opaque_paths: HashSet, + issues: BTreeMap, +} + +impl FolderScan { + fn iter(&self) -> std::slice::Iter<'_, ScannedFile> { + self.files.iter() + } + + #[cfg(test)] + fn len(&self) -> usize { + self.files.len() + } + + fn record_issue(&mut self, path: String, issue: ScanIssue) { + self.opaque_paths.insert(path.clone()); + let display = if path.is_empty() { + ".".to_string() + } else { + path + }; + self.issues.insert(display, issue); + } + + fn state(&self, path: &str) -> ScanPathState { + if self.present_paths.contains(path) { + return ScanPathState::Present; + } + if path_ancestor_in_set(path, &self.opaque_paths).is_some() { + return ScanPathState::Unknown; + } + + let mut parent = String::new(); + for component in path.split('/') { + if !self.complete_dirs.contains(&parent) { + return ScanPathState::Unknown; + } + let child = if parent.is_empty() { + component.to_string() + } else { + format!("{parent}/{component}") + }; + if !self.seen_paths.contains(&child) { + return ScanPathState::AffirmativelyGone; + } + parent = child; + } + // The name exists, but it is not a readable regular file. + ScanPathState::Unknown + } + + fn has_presence_evidence(&self, path: &str) -> bool { + self.seen_paths.contains(path) + || path_ancestor_in_set(path, &self.opaque_paths).is_some_and(|path| !path.is_empty()) + } +} + +impl IntoIterator for FolderScan { + type Item = ScannedFile; + type IntoIter = std::vec::IntoIter; + + fn into_iter(self) -> Self::IntoIter { + self.files.into_iter() + } +} + +impl<'a> IntoIterator for &'a FolderScan { + type Item = &'a ScannedFile; + type IntoIter = std::slice::Iter<'a, ScannedFile>; + + fn into_iter(self) -> Self::IntoIter { + self.files.iter() + } +} + +fn path_ancestor_in_set<'a>(path: &str, paths: &'a HashSet) -> Option<&'a str> { + let mut candidate = path; + loop { + if let Some(found) = paths.get(candidate) { + return Some(found); + } + let Some(index) = candidate.rfind('/') else { + break; + }; + candidate = &candidate[..index]; + } + paths.get("").map(String::as_str) +} + +fn path_has_issue(path: &str, issues: &BTreeMap) -> bool { + if issues.contains_key(".") { + return true; + } + let mut candidate = path; + loop { + if issues.contains_key(candidate) { + return true; + } + let Some(index) = candidate.rfind('/') else { + return false; + }; + candidate = &candidate[..index]; + } +} + +enum BoundedRead { + Bytes(Vec), + TooLarge(u64), +} + +fn read_file_bounded(path: &Path, limit: u64) -> std::io::Result { + let file = std::fs::File::open(path)?; + let size = file.metadata()?.len(); + if size > limit { + return Ok(BoundedRead::TooLarge(size)); + } + let mut bytes = Vec::new(); + file.take(limit.saturating_add(1)).read_to_end(&mut bytes)?; + if bytes.len() as u64 > limit { + Ok(BoundedRead::TooLarge(bytes.len() as u64)) + } else { + Ok(BoundedRead::Bytes(bytes)) + } +} + struct ScannedFile { rel: String, path: PathBuf, @@ -2063,8 +2270,35 @@ impl ScannedFile { fn read_bytes(&self) -> Result> { match &self.bytes { Some(bytes) => Ok(bytes.clone()), - None => std::fs::read(&self.path) - .with_context(|| format!("failed to read {}", self.path.display())), + None => match read_file_bounded(&self.path, MAX_BLOB as u64) + .with_context(|| format!("failed to read {}", self.path.display()))? + { + BoundedRead::Bytes(bytes) => Ok(bytes), + BoundedRead::TooLarge(size) => anyhow::bail!( + "sync file {} has {size} bytes, above the {MAX_BLOB}-byte limit", + self.path.display() + ), + }, + } + } +} + +fn scanned_bytes_or_unknown( + file: &ScannedFile, + previous: &HashMap, + current: &mut HashMap, + issues: &mut BTreeMap, +) -> Option> { + match file.read_bytes() { + Ok(bytes) => Some(bytes), + Err(_) => { + issues.insert(file.rel.clone(), ScanIssue::Unreadable); + if let Some(hash) = previous.get(&file.rel) { + current.insert(file.rel.clone(), *hash); + } else { + current.remove(&file.rel); + } + None } } } @@ -2090,23 +2324,61 @@ impl ScannedFile { /// keying it on the manifest made a local caching decision from a value another /// machine chose, so two contending entries of equal size could collide on size /// plus mtime and the cache then reported content the file did not hold. +#[cfg(test)] fn scan_folder( root: &Path, entry: &SyncEntry, cache: &HashMap, -) -> Result> { - let mut out = Vec::new(); +) -> Result { + scan_folder_with_limit(root, entry, cache, MAX_BLOB as u64) +} + +fn scan_folder_with_limit( + root: &Path, + entry: &SyncEntry, + cache: &HashMap, + max_blob: u64, +) -> Result { + let mut scan = FolderScan::default(); if !root.exists() { - return Ok(out); - } - let mut stack = vec![root.to_path_buf()]; - while let Some(dir) = stack.pop() { - for child in - std::fs::read_dir(&dir).with_context(|| format!("failed to read {}", dir.display()))? - { - let child = child?; - let file_type = child.file_type()?; + return Ok(scan); + } + let mut stack = vec![(root.to_path_buf(), String::new())]; + while let Some((dir, dir_rel)) = stack.pop() { + let children = match std::fs::read_dir(&dir) { + Ok(children) => children, + Err(_) => { + scan.record_issue(dir_rel, ScanIssue::Unreadable); + continue; + } + }; + let mut complete = true; + for child in children { + let child = match child { + Ok(child) => child, + Err(_) => { + complete = false; + continue; + } + }; let path = child.path(); + let Ok(rel) = path.strip_prefix(root) else { + complete = false; + continue; + }; + let rel = rel.to_string_lossy(); + let Some(norm) = Manifest::normalize_path(&rel) else { + complete = false; + continue; + }; + scan.seen_paths.insert(norm.clone()); + let file_type = match child.file_type() { + Ok(file_type) => file_type, + Err(_) => { + scan.record_issue(norm, ScanIssue::Unreadable); + continue; + } + }; if file_type.is_symlink() { // Git tracks a symlink as a first-class object; fabric does not // yet, because a symlink is a different KIND of manifest entry @@ -2116,22 +2388,23 @@ fn scan_folder( "fabric: skipping symlink {} — fabric does not sync symlinks, git does", path.display() ); + scan.opaque_paths.insert(norm.clone()); + if entry.includes(&norm) { + scan.record_issue(norm, ScanIssue::Unsupported); + } continue; // never follow symlinks out of the folder } if file_type.is_dir() { - stack.push(path); + stack.push((path, norm)); continue; } if !file_type.is_file() { + scan.opaque_paths.insert(norm.clone()); + if entry.includes(&norm) { + scan.record_issue(norm, ScanIssue::Unsupported); + } continue; } - let Ok(rel) = path.strip_prefix(root) else { - continue; - }; - let rel = rel.to_string_lossy(); - let Some(norm) = Manifest::normalize_path(&rel) else { - continue; - }; if !entry.includes(&norm) { continue; } @@ -2140,13 +2413,20 @@ fn scan_folder( // inside `mtime_of` and again on the next line. `mtime_of_metadata` // already takes the result, so the single-call shape was in the // file the whole time. - let disk = child.metadata().ok(); - let (mtime_secs, mtime_nanos) = disk - .as_ref() - .map(mtime_of_metadata) - .unwrap_or((0, 0)); - let size = disk.as_ref().map(|meta| meta.len()).unwrap_or(u64::MAX); - let executable = disk.as_ref().is_some_and(is_executable); + let disk = match child.metadata() { + Ok(disk) => disk, + Err(_) => { + scan.record_issue(norm, ScanIssue::Unreadable); + continue; + } + }; + let (mtime_secs, mtime_nanos) = mtime_of_metadata(&disk); + let size = disk.len(); + let executable = is_executable(&disk); + if size > max_blob { + scan.record_issue(norm, ScanIssue::TooLarge); + continue; + } // Reuse the recorded hash when size and both mtime components are // byte-identical to what THIS MACHINE last observed. Anything that // differs, or is unknown, is read and hashed as before. @@ -2157,14 +2437,23 @@ fn scan_folder( }); let (bytes, hash) = match known { Some(seen) => (None, seen.hash), - None => { - let bytes = std::fs::read(&path) - .with_context(|| format!("failed to read {}", path.display()))?; - let hash = content_hash(&bytes); - (Some(bytes), hash) - } + None => match read_file_bounded(&path, max_blob) { + Ok(BoundedRead::Bytes(bytes)) => { + let hash = content_hash(&bytes); + (Some(bytes), hash) + } + Ok(BoundedRead::TooLarge(_)) => { + scan.record_issue(norm, ScanIssue::TooLarge); + continue; + } + Err(_) => { + scan.record_issue(norm, ScanIssue::Unreadable); + continue; + } + }, }; - out.push(ScannedFile { + scan.present_paths.insert(norm.clone()); + scan.files.push(ScannedFile { rel: norm, path, bytes, @@ -2175,8 +2464,13 @@ fn scan_folder( executable, }); } + if complete { + scan.complete_dirs.insert(dir_rel); + } else { + scan.record_issue(dir_rel, ScanIssue::Unreadable); + } } - Ok(out) + Ok(scan) } /// Scan `root` into `node`: record every file, and treat files that vanished @@ -2211,9 +2505,19 @@ fn observed_from_disk( manifest: &Manifest, entry: &SyncEntry, cache: &mut HashMap, +) -> Result> { + observed_from_disk_with_limit(manifest, entry, cache, MAX_BLOB as u64) +} + +fn observed_from_disk_with_limit( + manifest: &Manifest, + entry: &SyncEntry, + cache: &mut HashMap, + max_blob: u64, ) -> Result> { let mut observed = HashMap::new(); - for file in scan_folder(&entry.folder, entry, cache)? { + let scan = scan_folder_with_limit(&entry.folder, entry, cache, max_blob)?; + for file in &scan { let hash = file.hash; cache.insert(file.rel.clone(), file.cache_entry()); if manifest @@ -2221,7 +2525,16 @@ fn observed_from_disk( .and_then(|entry| entry.meta()) .is_some_and(|meta| meta.hash == hash) { - observed.insert(file.rel, hash); + observed.insert(file.rel.clone(), hash); + } + } + // An oversized or unreadable path still gives presence evidence. Retain a + // manifest hash as the delete receipt, but only when the walk saw the path + // or a non-root opaque ancestor. A missing or unreadable root proves no + // local presence and must not arm a later mass delete. + for (path, meta) in manifest.present_paths() { + if scan.state(path) == ScanPathState::Unknown && scan.has_presence_evidence(path) { + observed.entry(path.clone()).or_insert(meta.hash); } } Ok(observed) @@ -2230,6 +2543,7 @@ fn observed_from_disk( /// Scan against the last state actually observed on disk. Manifest-only Present /// entries may have arrived from a concurrent reconcile and must not become /// tombstones merely because they have not been materialized yet. +#[cfg(test)] fn scan_into_node_observed( node: &mut SyncNode, root: &Path, @@ -2237,6 +2551,29 @@ fn scan_into_node_observed( policy: PolicyRules, observed: &mut HashMap, cache: &mut HashMap, +) -> Result { + let mut issues = BTreeMap::new(); + scan_into_node_observed_with_limit( + node, + root, + entry, + policy, + observed, + cache, + &mut issues, + MAX_BLOB as u64, + ) +} + +fn scan_into_node_observed_with_limit( + node: &mut SyncNode, + root: &Path, + entry: &SyncEntry, + policy: PolicyRules, + observed: &mut HashMap, + cache: &mut HashMap, + issues: &mut BTreeMap, + max_blob: u64, ) -> Result { // A ROOT THAT IS NOT THERE IS NOT A FOLDER SOMEBODY EMPTIED. // @@ -2253,9 +2590,11 @@ fn scan_into_node_observed( // Deleting the CONTENTS of a folder still propagates normally, because the // root survives that and the scan runs. if !root.exists() { + issues.clear(); return Ok(false); } - let scanned = scan_folder(root, entry, cache)?; + let scanned = scan_folder_with_limit(root, entry, cache, max_blob)?; + *issues = scanned.issues.clone(); // Refresh the cache from what this scan actually saw, so the next scan of an // untouched file is free. Rebuilt rather than merged, so a vanished path // does not leak an entry forever. @@ -2282,9 +2621,13 @@ fn scan_into_node_observed( .get(&file.rel) .is_some_and(|entry| !entry.is_present()) { + let Some(bytes) = scanned_bytes_or_unknown(file, &previous, &mut current, issues) + else { + continue; + }; if node.local_write_with_mode( &file.rel, - &file.read_bytes()?, + &bytes, file.mtime_secs, file.mtime_nanos, file.executable, @@ -2309,21 +2652,31 @@ fn scan_into_node_observed( .and_then(|entry| entry.meta()) .is_some_and(|meta| meta.hash == hash) { - node.put_content(file.read_bytes()?); + let Some(bytes) = scanned_bytes_or_unknown(file, &previous, &mut current, issues) + else { + continue; + }; + node.put_content(bytes); + } + } else { + let Some(bytes) = scanned_bytes_or_unknown(file, &previous, &mut current, issues) + else { + continue; + }; + if node.local_write_with_mode( + &file.rel, + &bytes, + file.mtime_secs, + file.mtime_nanos, + file.executable, + ) { + changed = true; } - } else if node.local_write_with_mode( - &file.rel, - &file.read_bytes()?, - file.mtime_secs, - file.mtime_nanos, - file.executable, - ) { - changed = true; } } let now = now_secs(); - for path in previous.keys() { + for (path, previous_hash) in &previous { if current.contains_key(path) { continue; } @@ -2341,6 +2694,29 @@ fn scan_into_node_observed( if !entry.includes(path) { continue; } + match scanned.state(path) { + ScanPathState::Present => continue, + ScanPathState::Unknown => { + // Keep the last presence receipt. If this path later becomes + // affirmatively absent, that receipt remains the evidence that + // a real local delete occurred. The issue map prevents status + // from calling the unreadable or oversized path clean. + if let Some(opaque) = path_ancestor_in_set(path, &scanned.opaque_paths) + .filter(|opaque| !opaque.is_empty()) + { + issues + .entry(opaque.to_string()) + .or_insert(ScanIssue::Unsupported); + } else if scanned.seen_paths.contains(path) { + issues + .entry(path.clone()) + .or_insert(ScanIssue::Unsupported); + } + current.insert(path.clone(), *previous_hash); + continue; + } + ScanPathState::AffirmativelyGone => {} + } if node.local_remove(path, policy, now) { changed = true; } @@ -2427,8 +2803,8 @@ fn materialize(node: &SyncNode, root: &Path, policy: PolicyRules) -> Result<()> /// `materialization_does_not_manufacture_an_mtime_collision`, which pins that /// only this machine can create one, so no peer can manufacture a hit. /// -/// Anything unknown, differing, or unreadable returns false and takes the full -/// read-and-hash path exactly as before. +/// Anything unknown or differing returns false. The caller performs a bounded +/// read for a readable path and skips a path named by the scan issue map. fn already_materialized( path: &Path, rel: &str, @@ -2449,6 +2825,7 @@ fn already_materialized( seen.size == disk.len() && seen.mtime_secs == mtime_secs && seen.mtime_nanos == mtime_nanos } +#[cfg(test)] fn materialize_tracked( node: &mut SyncNode, root: &Path, @@ -2458,6 +2835,30 @@ fn materialize_tracked( observed: &mut HashMap, cache: &HashMap, daemon_writes: Option<(&EntryWork, u64)>, +) -> Result<()> { + materialize_tracked_with_issues( + node, + root, + entry, + policy, + protected, + observed, + cache, + &BTreeMap::new(), + daemon_writes, + ) +} + +fn materialize_tracked_with_issues( + node: &mut SyncNode, + root: &Path, + entry: &SyncEntry, + policy: PolicyRules, + protected: &HashMap, + observed: &mut HashMap, + cache: &HashMap, + scan_issues: &BTreeMap, + daemon_writes: Option<(&EntryWork, u64)>, ) -> Result<()> { std::fs::create_dir_all(root) .with_context(|| format!("failed to create {}", root.display()))?; @@ -2490,11 +2891,18 @@ fn materialize_tracked( continue; } let path = root.join(&rel); + if path_has_issue(&rel, scan_issues) { + continue; + } if already_materialized(&path, &rel, &meta, cache) { observed.insert(rel.clone(), meta.hash); continue; } - let existing = std::fs::read(&path); + let existing = match read_file_bounded(&path, MAX_BLOB as u64) { + Ok(BoundedRead::Bytes(bytes)) => Ok(bytes), + Ok(BoundedRead::TooLarge(_)) => continue, + Err(error) => Err(error), + }; let protected_local_path = protected.contains_key(&rel); if policy.propagate_deletes && protected_local_path @@ -2528,7 +2936,8 @@ fn materialize_tracked( true } } - Err(_) => true, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => true, + Err(_) => continue, }; if needs_write { let Some(bytes) = node.get_content(&meta.hash) else { @@ -4204,6 +4613,166 @@ mod tests { assert_eq!(paths, vec!["agent.toml".to_string()]); } + #[test] + fn an_incomplete_parent_makes_absence_unknown() { + let scan = FolderScan { + complete_dirs: HashSet::from([String::new()]), + seen_paths: HashSet::from(["locked".to_string()]), + opaque_paths: HashSet::from(["locked".to_string()]), + ..FolderScan::default() + }; + + assert!(matches!( + scan.state("gone.txt"), + ScanPathState::AffirmativelyGone + )); + assert!(matches!( + scan.state("locked/kept.txt"), + ScanPathState::Unknown + )); + } + + #[test] + fn a_file_over_the_limit_is_unknown_until_a_real_delete() { + let dir = tempfile::tempdir().unwrap(); + let root = dir.path(); + let entry = entry_with_policy("bus", root, SyncPolicy::Bus); + let rules = entry.policy.rules(); + let old = b"old"; + std::fs::write(root.join("large.bin"), old).unwrap(); + std::fs::write(root.join("small.txt"), b"ok").unwrap(); + + let mut node = SyncNode::new(Author([1; 32])); + assert!(node.local_write("large.bin", old, 0, 0)); + assert!(node.local_write("small.txt", b"ok", 0, 0)); + let old_hash = content_hash(old); + let mut observed = HashMap::from([ + ("large.bin".to_string(), old_hash), + ("small.txt".to_string(), content_hash(b"ok")), + ]); + let mut cache = HashMap::new(); + let mut issues = BTreeMap::new(); + + std::fs::write(root.join("large.bin"), b"five!").unwrap(); + let changed = scan_into_node_observed_with_limit( + &mut node, + root, + &entry, + rules, + &mut observed, + &mut cache, + &mut issues, + 4, + ) + .unwrap(); + assert!(!changed, "an unsyncable file became a manifest change"); + assert!(node.manifest().get("large.bin").unwrap().is_present()); + assert_eq!(observed.get("large.bin"), Some(&old_hash)); + assert_eq!( + issues.get("large.bin").map(ScanIssue::token), + Some("too-large") + ); + assert!(node.manifest().get("small.txt").unwrap().is_present()); + + let mut restart_cache = HashMap::new(); + let restart_observed = observed_from_disk_with_limit( + node.manifest(), + &entry, + &mut restart_cache, + 4, + ) + .unwrap(); + assert_eq!( + restart_observed.get("large.bin"), + Some(&old_hash), + "a restart lost the presence receipt for the unsyncable file" + ); + + let protected = observed.clone(); + materialize_tracked_with_issues( + &mut node, + root, + &entry, + rules, + &protected, + &mut observed, + &cache, + &issues, + None, + ) + .unwrap(); + assert_eq!( + std::fs::read(root.join("large.bin")).unwrap(), + b"five!", + "materialization overwrote a present but unsyncable file" + ); + + std::fs::remove_file(root.join("large.bin")).unwrap(); + assert!( + scan_into_node_observed_with_limit( + &mut node, + root, + &entry, + rules, + &mut observed, + &mut cache, + &mut issues, + 4, + ) + .unwrap(), + "a complete parent scan must still prove the later delete" + ); + assert!(!node.manifest().get("large.bin").unwrap().is_present()); + } + + #[test] + fn a_directory_replacing_a_tracked_file_is_named_as_unsupported() { + let dir = tempfile::tempdir().unwrap(); + let root = dir.path(); + let entry = entry_with_policy("bus", root, SyncPolicy::Bus); + let rules = entry.policy.rules(); + let old = b"old"; + let hash = content_hash(old); + let mut node = SyncNode::new(Author([1; 32])); + assert!(node.local_write("path", old, 0, 0)); + let mut observed = HashMap::from([("path".to_string(), hash)]); + let mut cache = HashMap::new(); + let mut issues = BTreeMap::new(); + std::fs::create_dir(root.join("path")).unwrap(); + + let changed = scan_into_node_observed_with_limit( + &mut node, + root, + &entry, + rules, + &mut observed, + &mut cache, + &mut issues, + 4, + ) + .unwrap(); + + assert!(!changed, "an unsupported replacement became a delete"); + assert!(node.manifest().get("path").unwrap().is_present()); + assert_eq!(observed.get("path"), Some(&hash)); + assert_eq!( + issues.get("path").map(ScanIssue::token), + Some("unsupported") + ); + } + + #[test] + fn local_data_errors_are_not_network_weather() { + assert_eq!( + classify_reconcile_error("no local sync entry named \"catalog\""), + PeerSyncState::MissingEntry + ); + assert_eq!( + classify_reconcile_error("sync frame of 536870913 bytes exceeds limit 536870912"), + PeerSyncState::TooLarge + ); + } + /// A WATCHED FOLDER THAT IS NOT THERE IS NOT A FOLDER SOMEBODY EMPTIED. /// /// `scan_folder` returns an EMPTY result when the root does not exist. Every diff --git a/src/sync/wire.rs b/src/sync/wire.rs index 698341f..c9040c2 100644 --- a/src/sync/wire.rs +++ b/src/sync/wire.rs @@ -35,7 +35,7 @@ use super::node::{Reconciled, SyncNode, content_hash}; /// generous headroom, not a content limit). const MAX_JSON_FRAME: usize = 64 * 1024 * 1024; /// Largest single content blob accepted (per file). -const MAX_BLOB: usize = 512 * 1024 * 1024; +pub(crate) const MAX_BLOB: usize = 512 * 1024 * 1024; /// Largest blob count in one bundle. const MAX_BLOB_COUNT: u32 = 1_000_000; @@ -67,6 +67,9 @@ struct HelloHeader { struct ReplyHeader { manifest: Manifest, wanted: Vec, + /// A configuration or data error the initiator must act on. + #[serde(default, skip_serializing_if = "Option::is_none")] + error: Option, /// The responder's digest AFTER adopting what the initiator sent. /// /// This is the acknowledgement. The join is commutative, so once both sides @@ -112,6 +115,7 @@ async fn write_blobs( w: &mut W, blobs: &[(ContentHash, Vec)], ) -> Result<()> { + validate_blobs(blobs)?; write_u32(w, blobs.len() as u32).await?; for (hash, bytes) in blobs { w.write_all(&hash.0).await?; @@ -120,6 +124,37 @@ async fn write_blobs( Ok(()) } +fn validate_blobs(blobs: &[(ContentHash, Vec)]) -> Result<()> { + if blobs.len() > MAX_BLOB_COUNT as usize { + bail!( + "sync bundle of {} blobs exceeds limit {MAX_BLOB_COUNT}", + blobs.len() + ); + } + if let Some((_, bytes)) = blobs.iter().find(|(_, bytes)| bytes.len() > MAX_BLOB) { + bail!( + "sync content blob of {} bytes exceeds limit {MAX_BLOB}", + bytes.len() + ); + } + Ok(()) +} + +async fn write_error_reply(w: &mut W, message: &str) -> Result<()> { + let reply = ReplyHeader { + manifest: Manifest::new(), + wanted: Vec::new(), + error: Some(message.to_string()), + digest: String::new(), + is_delta: false, + }; + write_len_bytes(w, &serde_json::to_vec(&reply)?).await?; + // An older client ignores `error` and still expects the bundle count. + write_u32(w, 0).await?; + w.flush().await?; + Ok(()) +} + /// A content bundle read from the wire: how many blobs stored and their bytes. #[derive(Debug, Clone, Copy, Default)] struct Received { @@ -216,6 +251,9 @@ where // The peer's whole manifest comes back the same way. wire_bytes += reply_frame.len(); let reply: ReplyHeader = serde_json::from_slice(&reply_frame)?; + if let Some(error) = &reply.error { + bail!("{error}"); + } let received = read_blobs_into(&mut stream, &node).await?; wire_bytes += received.bytes; @@ -405,7 +443,9 @@ where }) .await? else { - bail!("no local sync entry named {name:?}"); + let message = format!("no local sync entry named {name:?}"); + write_error_reply(&mut stream, &message).await?; + bail!(message); }; // 2. Snapshot BEFORE adopting so the client still pushes the content we need, @@ -473,6 +513,7 @@ where let reply = ReplyHeader { manifest: server_payload, wanted: node.missing_content_hashes(), + error: None, digest: node.manifest().digest(), is_delta: reply_is_delta, }; @@ -485,6 +526,11 @@ where }; // 3. Send reply header + our content bundle, then read the client's push. + if let Err(error) = validate_blobs(&blobs_for_client) { + let message = format!("{error:#}"); + write_error_reply(&mut stream, &message).await?; + return Err(error); + } write_len_bytes(&mut stream, &serde_json::to_vec(&reply)?).await?; write_blobs(&mut stream, &blobs_for_client).await?; stream.flush().await?; @@ -555,6 +601,27 @@ mod tests { super::super::manifest::Author([n; 32]) } + #[tokio::test] + async fn a_missing_remote_entry_reaches_the_initiator_as_a_configuration_error() { + let node = Arc::new(Mutex::new(SyncNode::new(author(1)))); + let (client_end, server_end) = tokio::io::duplex(1 << 20); + let server = tokio::spawn(async move { + run_server(server_end, "peer-a", move |_| async move { + Ok::<_, anyhow::Error>(None::<(Arc>, ())>) + }) + .await + }); + + let error = run_client(client_end, node, "catalog", "peer-b") + .await + .expect_err("a missing remote entry was accepted"); + assert!( + format!("{error:#}").contains("no local sync entry named \"catalog\""), + "the initiator lost the remote configuration error: {error:#}" + ); + assert!(server.await.unwrap().is_err()); + } + /// WHAT `delta_fallbacks` DOES NOT COUNT. /// /// An initiator that keeps being changed mid-exchange never reaches a @@ -610,6 +677,7 @@ mod tests { let reply = ReplyHeader { manifest: Manifest::new(), wanted: Vec::new(), + error: None, digest: hello.digest.clone(), is_delta: true, }; @@ -847,6 +915,7 @@ mod tests { let reply = ReplyHeader { manifest: Manifest::new(), wanted: Vec::new(), + error: None, digest: converged.digest(), is_delta: true, }; @@ -1014,7 +1083,6 @@ mod tests { #[cfg(test)] mod wire_cost_tests { - use super::tests::*; use super::*; fn author(n: u8) -> super::super::manifest::Author { diff --git a/tests/folder_sync.rs b/tests/folder_sync.rs index f671b59..6b9a3d4 100644 --- a/tests/folder_sync.rs +++ b/tests/folder_sync.rs @@ -1430,6 +1430,37 @@ async fn stopped_peers_of(home: &FabricHome) -> Vec<(String, String)> { stopped } +/// A reachable peer without the named entry has a configuration mismatch. +/// Network retry cannot repair it, so status must not call it unreachable. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_missing_remote_entry_reports_a_configuration_error() -> Result<()> { + let _guard = FOLDER_SYNC_LOCK.lock().await; + let a_dir = TempDir::new()?; + let b_dir = TempDir::new()?; + let a_home = FabricHome::new(a_dir.path()); + let b_home = FabricHome::new(b_dir.path()); + let a_folder = a_dir.path().join("shared"); + std::fs::create_dir_all(&a_folder)?; + write_sync(a_dir.path(), &a_folder, "bus"); + + let node_a = FabricNode::start(a_home.clone()).await?; + let node_b = FabricNode::start(b_home.clone()).await?; + trust_peer(&a_home, &node_a, node_b.id(), "node-b", node_b.addr()).await?; + trust_peer(&b_home, &node_b, node_a.id(), "node-a", node_a.addr()).await?; + + reload_sync(&a_home).await?; + let stopped = stopped_peers_of(&a_home).await; + assert_eq!( + stopped, + vec![("node-b".to_string(), "missing-entry".to_string())], + "a reachable peer without the entry reported {stopped:?}" + ); + + node_b.shutdown().await?; + node_a.shutdown().await?; + Ok(()) +} + /// Finding 3 of the 2026-08-29 review. A selector that matches no peer in /// `peers.toml` was dropped without a record. The engine looped over the peers /// that did resolve, recorded nothing for the one that did not, and every