wolfdisk v2.11.6: followers self-heal divergent state instead of getting stuck (wabil)

A bulk delete of 250k files followed by a restart left the cluster wedged:
the leader had 226 files, followers had 262k, yet all reported "v0 — in sync"
and deletions never propagated. Three compounding causes, all fixed:

1. The replication index_version + changelog were in-memory only, so a restart
   reset the leader to v0 while the on-disk file index persisted. Every node
   then falsely agreed on v0 and never reconciled. Now persisted to
   replication_state.json (atomic temp+rename) and reloaded on startup.

2. Follower sync was ADDITIVE — it inserted the leader's entries and removed
   only changelog-listed deletions, so a follower holding stale extras never
   dropped them. New is_full flag on SyncResponse marks a full index; the
   follower then reconciles AUTHORITATIVELY, deleting any local path the leader
   no longer has (batched, lock-friendly; guarded so an empty/failed-load leader
   index can never wipe a follower). This is the self-heal: a stuck follower now
   converges on its own within a resync cycle — no manual re-seed.

3. changes_since() returned Some((empty,empty)) for an empty changelog while
   behind ("nothing changed"); now returns None to force a full sync.

WIRE CHANGE (is_full appended to SyncResponseMsg, bincode positional): upgrade
the LEADER first, then followers — or all nodes together. Tests cover the
restart trap, persistence round-trip, and the authoritative delete selection.
Independent code review: no critical findings; data-loss/deadlock paths clear.

Author: Paul C

wolfdisk/Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "wolfdisk"
-version = "2.11.5"
+version = "2.11.6"
 edition = "2021"
 authors = ["Wolf Software Systems Ltd"]
 description = "Distributed file system with replicated and shared storage modes"

wolfdisk/Cargo.toml

@@ -35,6 +35,18 @@ pub struct PeerInfo {
     pub last_seen: Instant,
 }
 
+/// On-disk filename for the persisted replication version + changelog,
+/// stored alongside the file index.
+const REPL_STATE_FILENAME: &str = "replication_state.json";
+
+/// Serializable snapshot of the replication state that must survive a restart:
+/// the monotonic index version and the recent changelog used for delta sync.
+#[derive(serde::Serialize, serde::Deserialize)]
+struct PersistedReplicationState {
+    version: u64,
+    changelog: Vec<(u64, std::path::PathBuf, bool)>,
+}
+
 /// Cluster manager - handles leader election and state
 pub struct ClusterManager {
     config: Config,
@@ -184,13 +196,30 @@ impl ClusterManager {
         &self,
         from_version: u64,
     ) -> Option<(Vec<std::path::PathBuf>, Vec<std::path::PathBuf>)> {
+        let current = self.index_version();
+
+        // Already at (or somehow beyond) our version → genuinely nothing to send.
+        if from_version >= current {
+            return Some((Vec::new(), Vec::new()));
+        }
+
         let log = self.changelog.read().unwrap();
 
-        // Check if changelog goes back far enough
-        if let Some((oldest_version, _, _)) = log.front() {
-            if from_version < *oldest_version {
-                // Changelog truncated, need full sync
-                return None;
+        // We're behind (from_version < current), so the changelog MUST contain
+        // every change after from_version to answer with a delta. If it's empty,
+        // or its oldest entry is newer than from_version+1, there's a gap we
+        // can't fill → force a full sync. The empty case is the critical one:
+        // after a restart the in-memory changelog is empty while `current` was
+        // restored from disk, and the old code returned Some((empty, empty))
+        // here — telling the follower "nothing changed" when in fact it must
+        // re-reconcile against the leader's full index (wabil 2026-06-29).
+        match log.front() {
+            None => return None,
+            Some((oldest_version, _, _)) => {
+                if from_version.saturating_add(1) < *oldest_version {
+                    // Changelog truncated, need full sync
+                    return None;
+                }
             }
         }
 
@@ -222,6 +251,60 @@ impl ClusterManager {
         *self.index_version.write().unwrap() = version;
     }
 
+    /// Persist the replication index version + changelog next to the file index,
+    /// written atomically (temp + rename). Without this the version is in-memory
+    /// only, so a restart reset the leader to v0 — every node then falsely agreed
+    /// "v0, in sync" while their file sets differed by a quarter-million entries,
+    /// and deletions never propagated (wabil 2026-06-29).
+    pub fn save_replication_state(&self, index_dir: &std::path::Path) -> std::io::Result<()> {
+        // Snapshot version + changelog CONSISTENTLY: hold the version lock across
+        // the changelog read, mirroring record_change's lock order (version then
+        // changelog). Otherwise a concurrent record_change could bump the version
+        // and append its entry between the two reads, persisting a version one
+        // behind its own changelog.
+        let (version, changelog) = {
+            let v = self.index_version.read().unwrap();
+            let log: Vec<(u64, std::path::PathBuf, bool)> =
+                self.changelog.read().unwrap().iter().cloned().collect();
+            (*v, log)
+        };
+        let state = PersistedReplicationState { version, changelog };
+        let bytes = serde_json::to_vec(&state).map_err(std::io::Error::other)?;
+        std::fs::create_dir_all(index_dir)?;
+        let path = index_dir.join(REPL_STATE_FILENAME);
+        let tmp = index_dir.join(format!("{}.tmp", REPL_STATE_FILENAME));
+        std::fs::write(&tmp, &bytes)?;
+        std::fs::rename(&tmp, &path)?;
+        Ok(())
+    }
+
+    /// Restore the replication version + changelog on startup. A missing or
+    /// corrupt file simply leaves us at v0 (the prior behaviour) — which now
+    /// self-heals via the authoritative full sync regardless.
+    pub fn load_replication_state(&self, index_dir: &std::path::Path) {
+        let path = index_dir.join(REPL_STATE_FILENAME);
+        let bytes = match std::fs::read(&path) {
+            Ok(b) => b,
+            Err(_) => return,
+        };
+        let state: PersistedReplicationState = match serde_json::from_slice(&bytes) {
+            Ok(s) => s,
+            Err(e) => {
+                warn!("Ignoring corrupt replication state ({}); starting at v0", e);
+                return;
+            }
+        };
+        *self.index_version.write().unwrap() = state.version;
+        let mut log = self.changelog.write().unwrap();
+        log.clear();
+        log.extend(state.changelog);
+        info!(
+            "Restored replication state: version {}, {} changelog entries",
+            state.version,
+            log.len()
+        );
+    }
+
     /// Mark initial sync as complete (allows election monitor to promote to leader)
     pub fn set_sync_complete(&self) {
         *self.initial_sync_complete.write().unwrap() = true;
@@ -571,3 +654,67 @@ impl ClusterManager {
         }
     }
 }
+
+#[cfg(test)]
+mod replication_state_tests {
+    use super::*;
+    use std::path::PathBuf;
+
+    fn mgr() -> ClusterManager {
+        ClusterManager::new(Config::default())
+    }
+
+    #[test]
+    fn changes_since_at_version_is_empty_not_none() {
+        let m = mgr();
+        // Genuinely caught up (from == current == 0) → "no changes", NOT a full sync.
+        assert_eq!(m.changes_since(0), Some((Vec::new(), Vec::new())));
+        m.increment_index_version(PathBuf::from("/a"));
+        assert_eq!(m.changes_since(1), Some((Vec::new(), Vec::new())));
+    }
+
+    #[test]
+    fn changes_since_reports_changes_and_deletions() {
+        let m = mgr();
+        m.increment_index_version(PathBuf::from("/a")); // v1 change
+        m.record_deletion(PathBuf::from("/b")); // v2 delete
+        let (changed, deleted) = m.changes_since(0).expect("covered by changelog");
+        assert_eq!(changed, vec![PathBuf::from("/a")]);
+        assert_eq!(deleted, vec![PathBuf::from("/b")]);
+    }
+
+    #[test]
+    fn behind_with_empty_changelog_forces_full_sync() {
+        // The restart trap: version was restored from disk (so current > 0) but
+        // the in-memory changelog is empty. The old code returned Some((empty,
+        // empty)) here — telling the follower "nothing changed" when it must in
+        // fact reconcile against the full index. It MUST be None (full sync).
+        let m = mgr();
+        m.set_index_version(262_700);
+        assert_eq!(m.changes_since(0), None);
+        assert_eq!(m.changes_since(100), None);
+    }
+
+    #[test]
+    fn replication_state_survives_save_load() {
+        let dir = std::env::temp_dir().join(format!("wolfdisk_rs_test_{}", std::process::id()));
+        std::fs::create_dir_all(&dir).unwrap();
+
+        let m1 = mgr();
+        m1.increment_index_version(PathBuf::from("/x")); // v1
+        m1.increment_index_version(PathBuf::from("/y")); // v2
+        m1.record_deletion(PathBuf::from("/x")); // v3
+        m1.save_replication_state(&dir).unwrap();
+
+        let m2 = mgr();
+        assert_eq!(m2.index_version(), 0);
+        m2.load_replication_state(&dir);
+        assert_eq!(m2.index_version(), 3);
+        // Changelog restored → delta sync still works across the restart.
+        let (changed, deleted) = m2.changes_since(1).expect("changelog restored");
+        assert!(changed.contains(&PathBuf::from("/y")));
+        assert!(deleted.contains(&PathBuf::from("/x")));
+
+        std::fs::remove_dir_all(&dir).ok();
+    }
+}

wolfdisk/src/cluster/state.rs

@@ -69,6 +69,20 @@ enum Commands {
     },
 }
 
+/// Paths the follower must delete on a full (authoritative) sync: every local
+/// path the leader's complete index no longer contains. This is the
+/// data-deleting decision of the self-heal, so it's a pure function with a
+/// direct regression test (see tests at the bottom of this file).
+fn full_sync_stale_paths<'a>(
+    leader_paths: &std::collections::HashSet<std::path::PathBuf>,
+    local_paths: impl Iterator<Item = &'a std::path::PathBuf>,
+) -> Vec<std::path::PathBuf> {
+    local_paths
+        .filter(|p| !leader_paths.contains(*p))
+        .cloned()
+        .collect()
+}
+
 fn main() {
     let cli = Cli::parse();
 
@@ -124,6 +138,12 @@ fn main() {
             ));
             let file_index_for_handler = file_index.clone();
 
+            // Restore the replication version + changelog from disk so a restart
+            // doesn't reset us to v0 — which made every node falsely agree "v0, in
+            // sync" while their file sets diverged, and forced a heavy full
+            // reconcile on every restart (wabil 2026-06-29).
+            cluster.load_replication_state(&config.index_dir());
+
             // Create chunk store for replication (shared with WolfDiskFS). When an SSD
             // cache tier is configured it sits in front of the bulk data_dir.
             let chunk_store = std::sync::Arc::new(
@@ -1307,6 +1327,7 @@ fn main() {
                                     current_version,
                                     entries: Vec::new(),
                                     deleted_paths: Vec::new(),
+                                    is_full: false,
                                 }))
                             } else if sync_req.from_version > 0 {
                                 // Try delta sync — only send files that changed since their version
@@ -1364,6 +1385,7 @@ fn main() {
                                             current_version,
                                             entries,
                                             deleted_paths: deleted_strs,
+                                            is_full: false,
                                         }))
                                     }
                                     None => {
@@ -1411,6 +1433,8 @@ fn main() {
                                             current_version,
                                             entries,
                                             deleted_paths: Vec::new(),
+                                            // Full index → follower reconciles authoritatively.
+                                            is_full: true,
                                         }))
                                     }
                                 }
@@ -1465,6 +1489,8 @@ fn main() {
                                     current_version,
                                     entries,
                                     deleted_paths: Vec::new(),
+                                    // Full index (initial / from_version 0) → authoritative.
+                                    is_full: true,
                                 }))
                             }
                         }
@@ -2292,6 +2318,41 @@ fn main() {
                                     }
                                 }
 
+                                // Authoritative full sync: the leader sent its COMPLETE index, so
+                                // any local entry it did NOT send is stale and must go. This is what
+                                // makes a follower actually CONVERGE after it falls out of changelog
+                                // range or misses a bulk delete — the old code only ever added/updated
+                                // and applied changelog deletions, so a follower holding 250k stale
+                                // files re-received the leader's small index every cycle and kept the
+                                // extras forever (wabil 2026-06-29). Snapshot local paths under a
+                                // brief read lock, diff against the leader's set off-lock, then delete
+                                // in small batches so FUSE ops aren't starved on a huge index.
+                                // Guard: never authoritatively wipe to match an EMPTY leader index.
+                                // A leader whose index failed to load reads as 0 entries; honouring
+                                // that would nuke every follower. A genuine "delete everything" still
+                                // propagates as normal changelog deletions, not a full-sync wipe.
+                                if response.is_full && !response.entries.is_empty() {
+                                    let keep: std::collections::HashSet<std::path::PathBuf> = response
+                                        .entries
+                                        .iter()
+                                        .map(|e| std::path::PathBuf::from(&e.path))
+                                        .collect();
+                                    let stale: Vec<std::path::PathBuf> = {
+                                        let index = resync_file_index.read().unwrap();
+                                        full_sync_stale_paths(&keep, index.paths())
+                                    };
+                                    for del_batch in stale.chunks(50) {
+                                        let mut index = resync_file_index.write().unwrap();
+                                        let mut inode_tbl = resync_inode_table.write().unwrap();
+                                        for del_path in del_batch {
+                                            if index.remove(del_path).is_some() {
+                                                inode_tbl.remove_path(del_path);
+                                                removed += 1;
+                                            }
+                                        }
+                                    }
+                                }
+
                                 // Update our version to the leader's version, and publish it
                                 // to the cluster so the health UI reflects real progress.
                                 // Previously this stayed in this thread's local variable, so a
@@ -2383,11 +2444,17 @@ fn main() {
             // the index directly without setting the FUSE dirty flag.
             let persist_file_index = file_index.clone();
             let persist_index_dir = config.index_dir().to_path_buf();
+            let persist_cluster = cluster.clone();
             std::thread::spawn(move || loop {
                 std::thread::sleep(std::time::Duration::from_secs(5));
                 if std::sync::Arc::strong_count(&persist_file_index) <= 1 {
                     break;
                 }
+                // Persist the replication version + changelog so a restart resumes
+                // at the right version instead of resetting to v0 (wabil 2026-06-29).
+                if let Err(e) = persist_cluster.save_replication_state(&persist_index_dir) {
+                    tracing::warn!("Failed to persist replication state: {}", e);
+                }
                 // Serialize under the read lock (CPU only), then write to disk
                 // AFTER releasing it. The old code held the read lock across the
                 // whole disk write of a 250k-entry index every 5s, blocking every
@@ -2689,3 +2756,30 @@ fn main() {
         }
     }
 }
+
+#[cfg(test)]
+mod full_sync_tests {
+    use super::full_sync_stale_paths;
+    use std::collections::HashSet;
+    use std::path::PathBuf;
+
+    fn p(s: &str) -> PathBuf { PathBuf::from(s) }
+
+    #[test]
+    fn deletes_only_paths_absent_from_leader() {
+        // Leader's authoritative set (post bulk-delete): two files.
+        let leader: HashSet<PathBuf> = [p("/keep/a"), p("/keep/b")].into_iter().collect();
+        // Follower still holds the leader's two PLUS three stale extras.
+        let local = vec![p("/keep/a"), p("/keep/b"), p("/old/1"), p("/old/2"), p("/old/3")];
+        let mut stale = full_sync_stale_paths(&leader, local.iter());
+        stale.sort();
+        assert_eq!(stale, vec![p("/old/1"), p("/old/2"), p("/old/3")]);
+    }
+
+    #[test]
+    fn keeps_everything_when_follower_matches_leader() {
+        let leader: HashSet<PathBuf> = [p("/a"), p("/b")].into_iter().collect();
+        let local = vec![p("/a"), p("/b")];
+        assert!(full_sync_stale_paths(&leader, local.iter()).is_empty());
+    }
+}

wolfdisk/src/main.rs

@@ -197,6 +197,21 @@ pub struct SyncResponseMsg {
     pub entries: Vec<IndexEntryMsg>,
     /// Paths deleted since the requested version (for delta sync)
     pub deleted_paths: Vec<String>,
+    /// True when `entries` is the leader's COMPLETE index (a full sync), not a
+    /// delta. The follower then reconciles authoritatively — it removes any
+    /// local path absent from `entries`, instead of merely merging. Without
+    /// this, a follower that fell out of changelog range (e.g. a bulk delete of
+    /// 250k files, or a leader that restarted to v0) re-received the leader's
+    /// small index every cycle but never dropped its stale extras, so it never
+    /// converged (wabil 2026-06-29). Appended LAST (bincode is positional — see
+    /// encode_message's WIRE-COMPAT note). Rolling-upgrade order for THIS field:
+    /// **leader first, then followers**. A new follower cannot decode an old
+    /// leader's (shorter) response and would loop on sync errors; an old follower
+    /// harmlessly ignores this trailing field on a new leader's response (it just
+    /// never runs the authoritative reconcile). Upgrading all nodes together is
+    /// also safe.
+    #[serde(default)] // no-op for the bincode transport (positional); kept for JSON/debug only
+    pub is_full: bool,
 }
 
 /// Index entry in sync response
@@ -400,7 +415,11 @@ pub struct CreateSymlinkMsg {
 /// JSON). Adding a field to a message struct is a breaking wire change: mixed
 /// versions during a rolling upgrade WILL mis-decode, and for a `Vec<…>` payload
 /// (e.g. SyncResponse.entries) a new→old decode corrupts every entry after the
-/// first. Upgrade order: all nodes together, or followers before the leader.
+/// first. Upgrade order is DIRECTION-DEPENDENT: upgrade the side that DECODES the
+/// changed message last. For a field added to a response (leader→follower, e.g.
+/// SyncResponseMsg.is_full) upgrade the LEADER first; for a field added to a
+/// request (follower→leader) upgrade FOLLOWERS first. Upgrading all nodes
+/// together is always safe.
 pub fn encode_message(msg: &Message) -> Result<Vec<u8>, bincode::Error> {
     let serialized = bincode::serialize(msg)?;
     Ok(lz4_flex::compress_prepend_size(&serialized))