From 867a5378026e6d84233f786e867ecdb6750af856 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Wed, 21 Oct 2026 20:18:56 -0700 Subject: [PATCH] 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 | 90 ++++++++++++++++++++- weed/storage/store.go | 38 ++++++++- weed/storage/store_configure_volume_test.go | 47 +++++++++++ weed/storage/volume.go | 29 +++++++ 4 files changed, 199 insertions(+), 5 deletions(-) create mode 100644 weed/storage/store_configure_volume_test.go diff --git a/seaweed-volume/src/storage/store.rs b/seaweed-volume/src/storage/store.rs index e1e12fb608..4cef145590 100644 --- a/seaweed-volume/src/storage/store.rs +++ b/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 { 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 diff --git a/weed/storage/store.go b/weed/storage/store.go index 0186b0d269..0d2b51b50d 100644 --- a/weed/storage/store.go +++ b/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) } diff --git a/weed/storage/store_configure_volume_test.go b/weed/storage/store_configure_volume_test.go new file mode 100644 index 0000000000..8cb4c85629 --- /dev/null +++ b/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")) +} diff --git a/weed/storage/volume.go b/weed/storage/volume.go index 553435e5b8..d92a930309 100644 --- a/weed/storage/volume.go +++ b/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()