volume.configure: keep the superblock in sync with the .vif (#11930)

* volume.configure: keep the superblock in sync with the .vif

ConfigureVolume only rewrote the .vif's replication field, leaving the
.dat superblock stale. A mounted volume also kept reporting the old
placement in heartbeats, and its next compaction would rewrite the stale
placement into a fresh .dat.

Parse the placement once, then update the live volume when mounted: the
in-memory .vif copy and superblock, and the .dat header through the
volume's own backend under the data-file locks. For unmounted volumes
rewrite the header directly; remote-only volumes skip it since the .vif
carries the placement.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* volume.configure: update the .dat superblock in the Rust volume server

Mirror the Go ConfigureVolume change: after rewriting the .vif, rewrite
the replica placement in the .dat header too, so a remounted volume's
superblock, heartbeats, and later compactions agree with the .vif. A
missing .dat (remote-only volume) is skipped since the .vif carries the
placement.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

---------

Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

seaweed-volume/src/storage/store.rs

@@ -18,7 +18,7 @@ use crate::storage::erasure_coding::ec_volume::{
 };
 use crate::storage::needle::needle::{Needle, get_actual_size};
 use crate::storage::needle_map::NeedleMapKind;
-use crate::storage::super_block::{ReplicaPlacement, SUPER_BLOCK_SIZE};
+use crate::storage::super_block::{ReplicaPlacement, SUPER_BLOCK_SIZE, SuperBlock};
 use crate::storage::types::*;
 use crate::storage::volume::{CompactionJob, VifVolumeInfo, Volume, VolumeError, VolumeSpec};
 
@@ -762,8 +762,8 @@ impl Store {
     }
 
     /// Configure a volume's replica placement on disk.
-    /// The volume must already be unmounted. This opens the .dat file directly,
-    /// modifies the replica_placement byte (offset 1), and writes it back.
+    /// The volume must already be unmounted. Rewrites the .vif and the .dat
+    /// superblock so both carry the new placement.
     pub fn configure_volume(&self, vid: VolumeId, rp: ReplicaPlacement) -> Result<(), VolumeError> {
         let (_, base_path, _) = self.find_volume_file_base(vid).ok_or_else(|| {
             VolumeError::Io(io::Error::new(
@@ -775,7 +775,7 @@ impl Store {
         let mut vif = load_vif_volume_info(&vif_path)?;
         vif.replication = rp.to_string();
         save_vif_volume_info(&vif_path, &vif)?;
-        Ok(())
+        update_dat_replica_placement(&format!("{}.dat", base_path), vid, rp)
     }
 
     // ---- Read / Write / Delete ----
@@ -1740,6 +1740,61 @@ fn strip_volume_suffix(filename: &str) -> Option<&str> {
         .or_else(|| filename.strip_suffix(".idx"))
 }
 
+/// Rewrites the replica placement in a .dat header for a volume that is not
+/// mounted; a missing .dat means a remote-only volume, where the .vif carries
+/// the placement.
+fn update_dat_replica_placement(
+    dat_path: &str,
+    vid: VolumeId,
+    rp: ReplicaPlacement,
+) -> Result<(), VolumeError> {
+    use std::io::{Read, Seek, SeekFrom, Write};
+
+    let mut file = match std::fs::OpenOptions::new()
+        .read(true)
+        .write(true)
+        .open(dat_path)
+    {
+        Ok(f) => f,
+        Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(()),
+        Err(e) => {
+            return Err(VolumeError::Io(io::Error::new(
+                e.kind(),
+                format!("volume {} failed to open {}: {}", vid, dat_path, e),
+            )))
+        }
+    };
+    let mut header = [0u8; SUPER_BLOCK_SIZE];
+    file.read_exact(&mut header).map_err(|e| {
+        VolumeError::Io(io::Error::new(
+            e.kind(),
+            format!("volume {} failed to parse superblock: {}", vid, e),
+        ))
+    })?;
+    let extra_size = u16::from_be_bytes([header[6], header[7]]) as usize;
+    let mut block = header.to_vec();
+    block.resize(SUPER_BLOCK_SIZE + extra_size, 0);
+    if extra_size > 0 {
+        file.read_exact(&mut block[SUPER_BLOCK_SIZE..]).map_err(|e| {
+            VolumeError::Io(io::Error::new(
+                e.kind(),
+                format!("volume {} failed to parse superblock: {}", vid, e),
+            ))
+        })?;
+    }
+    let mut super_block = SuperBlock::from_bytes(&block).map_err(|e| {
+        VolumeError::Io(io::Error::new(
+            io::ErrorKind::InvalidData,
+            format!("volume {} failed to parse superblock: {}", vid, e),
+        ))
+    })?;
+    super_block.replica_placement = rp;
+    file.seek(SeekFrom::Start(0))?;
+    file.write_all(&super_block.to_bytes())?;
+    file.sync_all()?;
+    Ok(())
+}
+
 fn load_vif_volume_info(path: &str) -> Result<VifVolumeInfo, VolumeError> {
     let content = match std::fs::read_to_string(path) {
         Ok(c) => c,
@@ -1977,6 +2032,33 @@ mod tests {
         assert_eq!(store.total_volume_count(), 1);
     }
 
+    // Mirrors Go's TestConfigureVolumeUpdatesSuperblock: configure must leave
+    // the .dat header and the .vif both on the new placement.
+    #[test]
+    fn test_configure_volume_updates_superblock() {
+        let tmp = TempDir::new().unwrap();
+        let dir = tmp.path().to_str().unwrap();
+        let mut store = make_test_store(&[dir]);
+
+        store
+            .add_volume(VolumeId(1), DiskType::HardDrive, &VolumeSpec::default())
+            .unwrap();
+        assert!(store.unmount_volume(VolumeId(1)).unwrap());
+
+        let rp = ReplicaPlacement::from_string("010").unwrap();
+        store.configure_volume(VolumeId(1), rp).unwrap();
+
+        let base = volume_file_name(dir, "", VolumeId(1));
+        let vif = load_vif_volume_info(&format!("{}.vif", base)).unwrap();
+        assert_eq!(vif.replication, "010");
+
+        let dat = std::fs::read(format!("{}.dat", base)).unwrap();
+        let super_block = SuperBlock::from_bytes(&dat).unwrap();
+        assert_eq!(super_block.replica_placement, rp);
+
+        assert!(store.configure_volume(VolumeId(2), rp).is_err());
+    }
+
     #[test]
     fn test_mount_volume_by_id_hint_skips_lone_sidecar() {
         // A lone .vif/.idx sidecar (e.g. an EC sidecar whose .ecx is on a

weed/storage/store.go

@@ -21,6 +21,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
 	"github.com/seaweedfs/seaweedfs/weed/stats"
+	"github.com/seaweedfs/seaweedfs/weed/storage/backend"
 	"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
 	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
@@ -1292,6 +1293,11 @@ func (s *Store) deleteVolumeGuarded(i needle.VolumeId, onlyEmpty bool, onlyGarba
 
 func (s *Store) ConfigureVolume(i needle.VolumeId, replication string) error {
 
+	rp, err := super_block.NewReplicaPlacementFromString(replication)
+	if err != nil {
+		return fmt.Errorf("volume %d replication %q: %v", i, replication, err)
+	}
+
 	for _, location := range s.Locations {
 		fileInfo, found := location.LocateVolume(i)
 		if !found {
@@ -1309,12 +1315,42 @@ func (s *Store) ConfigureVolume(i needle.VolumeId, replication string) error {
 		if err != nil {
 			return fmt.Errorf("volume %d failed to save vif: %v", i, err)
 		}
-		return nil
+		if v := s.GetVolume(i); v != nil {
+			return v.UpdateReplicaPlacement(rp)
+		}
+		return updateDatReplicaPlacement(filepath.Join(location.Directory, baseFileName+".dat"), i, rp)
 	}
 
 	return fmt.Errorf("volume %d not found on disk", i)
 }
 
+// updateDatReplicaPlacement rewrites the replica placement in a .dat header
+// for a volume that is not mounted; a mounted volume goes through
+// UpdateReplicaPlacement so the change lands under its data-file locks.
+func updateDatReplicaPlacement(datFileName string, vid needle.VolumeId, rp *super_block.ReplicaPlacement) error {
+	f, err := os.OpenFile(datFileName, os.O_RDWR, 0644)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return nil // remote-only volume: the .vif carries the placement
+		}
+		return fmt.Errorf("volume %d failed to open %s: %v", vid, datFileName, err)
+	}
+	defer f.Close()
+	datBackend := backend.NewDiskFile(f)
+	superBlock, err := super_block.ReadSuperBlock(datBackend)
+	if err != nil {
+		return fmt.Errorf("volume %d failed to parse superblock: %v", vid, err)
+	}
+	superBlock.ReplicaPlacement = rp
+	if _, err := datBackend.WriteAt(superBlock.Bytes(), 0); err != nil {
+		return fmt.Errorf("volume %d failed to write superblock: %v", vid, err)
+	}
+	if err := datBackend.Sync(); err != nil {
+		return fmt.Errorf("volume %d failed to sync superblock: %v", vid, err)
+	}
+	return nil
+}
+
 func (s *Store) SetVolumeSizeLimit(x uint64) {
 	s.volumeSizeLimit.Store(x)
 }

weed/storage/store_configure_volume_test.go

@@ -0,0 +1,47 @@
+package storage
+
+import (
+	"os"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/seaweedfs/seaweedfs/weed/stats"
+	"github.com/seaweedfs/seaweedfs/weed/storage/backend"
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
+	"github.com/seaweedfs/seaweedfs/weed/storage/types"
+	"github.com/seaweedfs/seaweedfs/weed/storage/volume_info"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+// volume.configure must leave the live superblock, the .dat header, the
+// in-memory .vif copy and the .vif on disk all on the new placement.
+func TestConfigureVolumeUpdatesSuperblock(t *testing.T) {
+	dir := t.TempDir()
+	store := NewStore(nil, "localhost", 8080, 18080, "http://localhost:8080", "store-id",
+		[]string{dir}, []int32{10}, []util.MinFreeSpace{{}}, "", NeedleMapInMemory,
+		[]types.DiskType{types.HardDriveType}, nil, 0, stats.DefaultDiskIOProbeConfig())
+	t.Cleanup(store.Close)
+	require.NoError(t, store.AddVolume(1, "", NeedleMapInMemory, "000", "", 0, needle.GetCurrentVersion(), 0, types.HardDriveType, 0))
+
+	require.NoError(t, store.ConfigureVolume(1, "010"))
+
+	v := store.GetVolume(1)
+	require.NotNil(t, v)
+	require.Equal(t, "010", v.ReplicaPlacement.String())
+	require.Equal(t, "010", v.GetVolumeInfo().GetReplication())
+
+	datFile, err := os.Open(v.FileName(".dat"))
+	require.NoError(t, err)
+	sb, err := super_block.ReadSuperBlock(backend.NewDiskFile(datFile))
+	require.NoError(t, err)
+	datFile.Close()
+	require.Equal(t, "010", sb.ReplicaPlacement.String())
+
+	volumeInfo, _, _, err := volume_info.MaybeLoadVolumeInfo(v.FileName(".vif"))
+	require.NoError(t, err)
+	require.Equal(t, "010", volumeInfo.GetReplication())
+
+	require.Error(t, store.ConfigureVolume(1, "bogus"))
+}

weed/storage/volume.go

@@ -188,6 +188,35 @@ func (v *Volume) Version() needle.Version {
 	return v.SuperBlock.Version
 }
 
+// UpdateReplicaPlacement records rp in the live volume — the in-memory .vif
+// copy and the superblock — and rewrites the .dat header when the data file
+// is local. Keeps a mounted volume consistent with the .vif the caller just
+// configured on disk, so heartbeats and later compactions report and rewrite
+// the same placement.
+func (v *Volume) UpdateReplicaPlacement(rp *super_block.ReplicaPlacement) error {
+	v.dataFileSyncLock.Lock()
+	defer v.dataFileSyncLock.Unlock()
+	v.dataFileAccessLock.Lock()
+	defer v.dataFileAccessLock.Unlock()
+	v.superBlockAccessLock.Lock()
+	defer v.superBlockAccessLock.Unlock()
+	v.volumeInfoRWLock.Lock()
+	defer v.volumeInfoRWLock.Unlock()
+
+	v.SuperBlock.ReplicaPlacement = rp
+	if v.volumeInfo != nil {
+		v.volumeInfo.Replication = rp.String()
+	}
+	diskFile, ok := v.DataBackend.(*backend.DiskFile)
+	if !ok {
+		return nil // remote-tiered: the .vif carries the placement
+	}
+	if _, err := diskFile.WriteAt(v.SuperBlock.Bytes(), 0); err != nil {
+		return fmt.Errorf("volume %d failed to write superblock: %v", v.Id, err)
+	}
+	return diskFile.Sync()
+}
+
 func (v *Volume) FileStat() (datSize uint64, idxSize uint64, modTime time.Time) {
 	v.dataFileAccessLock.RLock()
 	defer v.dataFileAccessLock.RUnlock()