Federated volume-id bound via heartbeat stamps (#11763) (#11819)

* pb: heartbeat fields for the federated volume-id bound (#11763)

Heartbeat carries max_volume_id_seen and last_seen_topology_id up, and
HeartbeatResponse echoes max_volume_id and topology_id down, so the
allocation bound is replicated to every volume server.

* master: echo the volume-id bound and fold reported bounds in (#11763)

Every heartbeat response carries the leader current max volume id and
topology id; every heartbeat reported max_volume_id_seen is folded into
the topology bound via UpAdjustMaxVolumeId.

* volume: persist a per-dir vid stamp of the highest id ever seen (#11763)

The store keeps max_volume_id_seen + last_seen_topology_id in a
.vid_stamp file per volume dir, fed by heartbeat echoes, AllocateVolume,
and local volume/EC creation, seeded at boot from stamps and local
holdings, and reported in every heartbeat. A rebuilt master recovers
the bound from the first surviving veteran report.

* volume-rust: mirror the vid stamp and heartbeat echo handling (#11763)

Same mechanism as the Go volume server: .vid_stamp per dir, bound folded
in from add_location scans and allocate/mount calls, echoed fields
applied on heartbeat responses, reported on every heartbeat.

* volume: serialize vid stamp writes and fsync the directory (#11763)

* volume-rust: serialize vid stamp writes and fsync the directory (#11763)

* master: echo the allocation bound to connected servers on change

The bound was only sent at connect time or on digest mismatch, so a
healthy stream never learned later increases. If the high-id server then
went offline and the master lost state, survivors restored a stale bound
and the id could be reissued. Resend the heartbeat response whenever the
bound or topology id moved.

* volume: retry failed vid stamp saves on later updates

A transient write failure left the in-memory bound raised but the stamp
stale, and repeat updates at the same bound never retried. Track a dirty
flag so any later update re-attempts the save.

* volume: cover stamp bootstrap for pre-stamp dirs in test

* volume-rust: retry failed vid stamp saves, mirroring Go

* volume-rust: stamp every dir with the combined bound on add_location

A dir whose own stamp sat below the combined bound kept its stale value
after fold; losing the higher-stamped disk then restarted the server
below ids it had already learned. Rewrite all stamps after each location
folds in.

* volume: stamp mounted volume and ec shard ids (#11763)

* volume-rust: stamp mounted volume ids, mirroring Go (#11763)

* volume: retry dirty vid stamp save on unchanged topology id (#11763)

* volume-rust: retry dirty vid stamp save on unchanged topology id (#11763)

* master: record echoed bound from the sent heartbeat response (#11763)

seaweed-volume/src/server/heartbeat.rs

@@ -474,6 +474,14 @@ async fn do_heartbeat(
                             )
                             .into());
                         }
+                        let echo_max_volume_id = hb_resp.max_volume_id;
+                        let echo_topology_id = hb_resp.topology_id.clone();
+                        off_runtime(config, state, move |_, state| {
+                            let store = state.store.read().unwrap();
+                            store.update_max_volume_id_seen(echo_max_volume_id);
+                            store.update_last_seen_topology_id(&echo_topology_id);
+                        })
+                        .await?;
                         let options_changed =
                             volume_options_pass(config, state, &hb_resp).await?;
                         if options_changed && maybe_adjust_volume_max(config, state).await? {
@@ -1360,6 +1368,8 @@ fn build_heartbeat_with_ec_status(
         grpc_port: config.grpc_port as u32,
         location_uuids,
         disk_tags,
+        max_volume_id_seen: store.max_volume_id_seen(),
+        last_seen_topology_id: store.last_seen_topology_id(),
         ..Default::default()
     };
     (heartbeat, volumes, actions)

seaweed-volume/src/storage/mod.rs

@@ -11,6 +11,7 @@ pub mod store_ec_mirror;
 pub mod store_ec_reconcile;
 pub mod super_block;
 pub mod types;
+pub mod vid_stamp;
 pub mod volume;
 pub mod volume_idx_rebuild;
 pub mod volume_idx_repair;

seaweed-volume/src/storage/store.rs

@@ -6,7 +6,7 @@
 
 use std::collections::HashSet;
 use std::io;
-use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering};
 use std::sync::{Arc, Mutex};
 
 use crate::config::MinFreeSpace;
@@ -72,6 +72,10 @@ pub struct Store {
     /// reporting the heartbeat that saw it.
     pub reported_collections: Mutex<HashSet<String>>,
     pub reported_ec_collections: Mutex<HashSet<String>>,
+    pub max_volume_id_seen: AtomicU32,
+    pub last_seen_topology_id: Mutex<String>,
+    pub vid_stamp_lock: Mutex<()>,
+    pub vid_stamp_dirty: std::sync::atomic::AtomicBool,
 }
 
 impl Store {
@@ -91,6 +95,10 @@ impl Store {
             rack: String::new(),
             reported_collections: Mutex::new(HashSet::new()),
             reported_ec_collections: Mutex::new(HashSet::new()),
+            max_volume_id_seen: AtomicU32::new(0),
+            last_seen_topology_id: Mutex::new(String::new()),
+            vid_stamp_lock: Mutex::new(()),
+            vid_stamp_dirty: std::sync::atomic::AtomicBool::new(false),
         }
     }
 
@@ -152,6 +160,12 @@ impl Store {
         // (read-only target, partial copy).
         self.reconcile_ec_shards_across_disks();
 
+        self.fold_location_into_vid_stamp(self.locations.last().unwrap());
+        // Write the combined bound back to every dir, including ones whose
+        // own stamp or volumes sit below it — a lost disk must not restart
+        // the server below the bound its remaining dirs once reached.
+        self.save_vid_stamps();
+
         Ok(())
     }
 
@@ -409,6 +423,7 @@ impl Store {
         disk_type: DiskType,
         spec: &VolumeSpec<'_>,
     ) -> Result<(), VolumeError> {
+        self.stamp_allocated_vid(vid);
         if self.find_volume(vid).is_some() {
             return Err(VolumeError::AlreadyExists);
         }
@@ -506,7 +521,7 @@ impl Store {
                         vid
                     ))));
                 }
-                return loc.create_volume(
+                let result = loc.create_volume(
                     vid,
                     self.needle_map_kind,
                     &VolumeSpec {
@@ -514,6 +529,10 @@ impl Store {
                         ..Default::default()
                     },
                 );
+                if result.is_ok() {
+                    self.stamp_allocated_vid(vid);
+                }
+                return result;
             }
         }
         Err(VolumeError::Io(io::Error::new(
@@ -603,7 +622,10 @@ impl Store {
                                 ..Default::default()
                             },
                         ) {
-                            Ok(()) => return Ok(()),
+                            Ok(()) => {
+                                self.stamp_allocated_vid(vid);
+                                return Ok(());
+                            }
                             Err(e) => {
                                 last_err = Some(e);
                                 continue;
@@ -659,7 +681,10 @@ impl Store {
                         ..Default::default()
                     },
                 ) {
-                    Ok(()) => return Ok(()),
+                    Ok(()) => {
+                        self.stamp_allocated_vid(vid);
+                        return Ok(());
+                    }
                     Err(e) => {
                         last_err = Some(e);
                         continue;
@@ -952,6 +977,7 @@ impl Store {
         collection: &str,
         shard_ids: &[ShardId],
     ) -> Result<(), VolumeError> {
+        self.stamp_allocated_vid(vid);
         // Find the location where the EC files live
         let loc_idx = self.find_ec_location(vid, collection).ok_or_else(|| {
             VolumeError::Io(io::Error::new(
@@ -974,6 +1000,7 @@ impl Store {
         shard_id: ShardId,
         source_disk_type: &str,
     ) -> Result<(), VolumeError> {
+        self.stamp_allocated_vid(vid);
         // The .ecx may live on a different disk than the shard being mounted
         // (ec.balance / ec.rebuild can spread shards across sibling disks), so
         // look up its owner once and point EcVolume::new at the directory that

seaweed-volume/src/storage/vid_stamp.rs

@@ -0,0 +1,232 @@
+//! Persisted volume-id high-water mark ("vid stamp"), mirroring Go's
+//! `weed/storage/vid_stamp.go`.
+//!
+//! Each volume dir carries a `.vid_stamp` JSON file with the highest volume
+//! id the server has ever observed plus the topology id it last heartbeated
+//! to. The bound is fed by heartbeat echoes, AllocateVolume, and local
+//! volume/EC creation, is reported back in `Heartbeat.max_volume_id_seen`,
+//! and survives master state loss so rebuilt masters cannot reissue ids an
+//! offline server still holds.
+
+use std::fs;
+use std::io;
+use std::path::Path;
+
+use serde::{Deserialize, Serialize};
+
+use crate::storage::types::VolumeId;
+
+use super::store::Store;
+
+const VID_STAMP_FILE_NAME: &str = ".vid_stamp";
+
+#[derive(Serialize, Deserialize, Default)]
+struct VidStamp {
+    #[serde(rename = "maxVolumeId")]
+    max_volume_id: u32,
+    #[serde(rename = "topologyId", default, skip_serializing_if = "String::is_empty")]
+    topology_id: String,
+}
+
+impl Store {
+    pub fn max_volume_id_seen(&self) -> u32 {
+        self.max_volume_id_seen.load(std::sync::atomic::Ordering::Relaxed)
+    }
+
+    pub fn last_seen_topology_id(&self) -> String {
+        self.last_seen_topology_id
+            .lock()
+            .unwrap()
+            .clone()
+    }
+
+    pub fn update_max_volume_id_seen(&self, vid: u32) {
+        use std::sync::atomic::Ordering::Relaxed;
+        let prev = self.max_volume_id_seen.fetch_max(vid, Relaxed);
+        if vid > prev || self.vid_stamp_dirty.load(Relaxed) {
+            self.save_vid_stamps();
+        }
+    }
+
+    pub fn update_last_seen_topology_id(&self, id: &str) {
+        if id.is_empty() {
+            return;
+        }
+        {
+            let mut guard = self.last_seen_topology_id.lock().unwrap();
+            if *guard == id && !self.vid_stamp_dirty.load(std::sync::atomic::Ordering::Relaxed) {
+                return;
+            }
+            *guard = id.to_string();
+        }
+        self.save_vid_stamps();
+    }
+
+    /// Seed the bound from a freshly added location: its persisted stamp,
+    /// its volumes, and its EC shards. Called per location from
+    /// `add_location` after the EC reconcile passes have run.
+    pub(crate) fn fold_location_into_vid_stamp(&self, loc: &super::disk_location::DiskLocation) {
+        if let Ok(stamp) = read_vid_stamp(Path::new(&loc.directory)) {
+            self.update_max_volume_id_seen(stamp.max_volume_id);
+            if !stamp.topology_id.is_empty() {
+                self.update_last_seen_topology_id(&stamp.topology_id);
+            }
+        }
+        let mut max_vid: u32 = 0;
+        for vid in loc.volume_ids() {
+            max_vid = max_vid.max(vid.0);
+        }
+        for (vid, _) in loc.ec_volumes() {
+            max_vid = max_vid.max(vid.0);
+        }
+        if max_vid > 0 {
+            self.update_max_volume_id_seen(max_vid);
+        }
+    }
+
+    pub(crate) fn save_vid_stamps(&self) {
+        let _guard = self.vid_stamp_lock.lock().unwrap();
+        let stamp = match serde_json::to_vec(&VidStamp {
+            max_volume_id: self.max_volume_id_seen(),
+            topology_id: self.last_seen_topology_id(),
+        }) {
+            Ok(s) => s,
+            Err(_) => return,
+        };
+        let mut failed = false;
+        for loc in &self.locations {
+            if let Err(e) = write_vid_stamp(Path::new(&loc.directory), &stamp) {
+                tracing::warn!("write vid stamp in {}: {}", loc.directory, e);
+                failed = true;
+            }
+        }
+        self.vid_stamp_dirty
+            .store(failed, std::sync::atomic::Ordering::Relaxed);
+    }
+}
+
+fn read_vid_stamp(dir: &Path) -> io::Result<VidStamp> {
+    let data = fs::read(dir.join(VID_STAMP_FILE_NAME))?;
+    serde_json::from_slice(&data).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))
+}
+
+fn write_vid_stamp(dir: &Path, stamp: &[u8]) -> io::Result<()> {
+    let tmp = dir.join(format!("{}.tmp", VID_STAMP_FILE_NAME));
+    {
+        use std::io::Write;
+        let mut f = fs::File::create(&tmp)?;
+        f.write_all(stamp)?;
+        f.sync_all()?;
+    }
+    fs::rename(&tmp, dir.join(VID_STAMP_FILE_NAME))?;
+    if cfg!(unix) {
+        fs::File::open(dir)?.sync_all()?;
+    }
+    Ok(())
+}
+
+/// Stamps the volume id of a newly created volume or EC shard set. Called
+/// from `Store::add_volume` / `Store::mount_ec_shards*` so the bound covers
+/// ids before their files exist.
+impl Store {
+    pub(crate) fn stamp_allocated_vid(&self, vid: VolumeId) {
+        self.update_max_volume_id_seen(vid.0);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::config::MinFreeSpace;
+    use crate::storage::needle_map::NeedleMapKind;
+    use crate::storage::types::DiskType;
+    use tempfile::TempDir;
+
+    fn make_store(dir: &str) -> Store {
+        let mut store = Store::new(NeedleMapKind::InMemory);
+        store
+            .add_location(dir, dir, 10, DiskType::HardDrive, MinFreeSpace::Percent(1.0), Vec::new())
+            .unwrap();
+        store
+    }
+
+    #[test]
+    fn stamp_written_on_bound_increase() {
+        let tmp = TempDir::new().unwrap();
+        let dir = tmp.path().to_str().unwrap();
+        let store = make_store(dir);
+
+        store.update_max_volume_id_seen(42);
+
+        let stamp = read_vid_stamp(tmp.path()).unwrap();
+        assert_eq!(stamp.max_volume_id, 42);
+    }
+
+    #[test]
+    fn stamp_survives_restart() {
+        let tmp = TempDir::new().unwrap();
+        let dir = tmp.path().to_str().unwrap();
+
+        let store = make_store(dir);
+        store.update_max_volume_id_seen(42);
+        store.update_last_seen_topology_id("cluster-a");
+        drop(store);
+
+        let restarted = make_store(dir);
+        assert_eq!(restarted.max_volume_id_seen(), 42);
+        assert_eq!(restarted.last_seen_topology_id(), "cluster-a");
+    }
+
+    #[test]
+    fn later_disk_gets_combined_stamp() {
+        let tmp1 = TempDir::new().unwrap();
+        let tmp2 = TempDir::new().unwrap();
+        // dir2 carries a stale stamp from an older run
+        write_vid_stamp(
+            tmp2.path(),
+            &serde_json::to_vec(&VidStamp {
+                max_volume_id: 10,
+                topology_id: String::new(),
+            })
+            .unwrap(),
+        )
+        .unwrap();
+
+        let mut store = Store::new(NeedleMapKind::InMemory);
+        store
+            .add_location(
+                tmp1.path().to_str().unwrap(),
+                tmp1.path().to_str().unwrap(),
+                10,
+                DiskType::HardDrive,
+                MinFreeSpace::Percent(1.0),
+                Vec::new(),
+            )
+            .unwrap();
+        store.update_max_volume_id_seen(100);
+        store
+            .add_location(
+                tmp2.path().to_str().unwrap(),
+                tmp2.path().to_str().unwrap(),
+                10,
+                DiskType::HardDrive,
+                MinFreeSpace::Percent(1.0),
+                Vec::new(),
+            )
+            .unwrap();
+
+        let stamp2 = read_vid_stamp(tmp2.path()).unwrap();
+        assert_eq!(stamp2.max_volume_id, 100);
+    }
+
+    #[test]
+    fn bound_is_monotonic() {
+        let tmp = TempDir::new().unwrap();
+        let dir = tmp.path().to_str().unwrap();
+        let store = make_store(dir);
+
+        store.update_max_volume_id_seen(10);
+        store.update_max_volume_id_seen(5);
+        assert_eq!(store.max_volume_id_seen(), 10);
+    }
+}