sync: delete only after a complete scan proves absence

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

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

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,

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,

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<Finding> {
         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<Finding> {
             .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<Finding> {
         ));
     }
 
-    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(),
             }
         })

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::<Vec<_>>()
+        .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,