diff --git a/seaweed-volume/src/storage/disk_location.rs b/seaweed-volume/src/storage/disk_location.rs index 5e39b361c..3929945f6 100644 --- a/seaweed-volume/src/storage/disk_location.rs +++ b/seaweed-volume/src/storage/disk_location.rs @@ -843,7 +843,8 @@ impl DiskLocation { // .ecx open error, .ecj create error, malformed .vif) would // have to panic via unwrap(). Build the EcVolume up front and // propagate the error to the caller. - if !self.ec_volumes.contains_key(&vid) { + let created = !self.ec_volumes.contains_key(&vid); + if created { let ec_vol = EcVolume::new(&dir, idx_dir, collection, vid) .map_err(VolumeError::Io)?; self.ec_volumes.insert(vid, ec_vol); @@ -866,9 +867,27 @@ impl DiskLocation { } for &shard_id in shard_ids { + // A mount retry re-listing a shard this volume already holds: + // keep the existing registration (mirrors Go's AddEcVolumeShard + // added=false) — re-adding would replace a serving fd and bump + // the ec_shards gauge without growing the mounted count. + if ec_vol.has_shard(shard_id as u8) { + continue; + } let mut shard = EcVolumeShard::new(&dir, collection, vid, shard_id as u8); shard.disk_type = ec_vol.disk_type.clone(); - ec_vol.add_shard(shard).map_err(VolumeError::Io)?; + if let Err(e) = ec_vol.add_shard(shard) { + // The shard was dropped (its descriptors closed) inside the + // failed add. If this call just created the EcVolume and it + // holds nothing, remove it too — a zero-shard registration + // would advertise a mount that serves no data while pinning + // its descriptors. + let now_empty = ec_vol.shard_count() == 0; + if created && now_empty { + self.ec_volumes.remove(&vid); + } + return Err(VolumeError::Io(e)); + } crate::metrics::VOLUME_GAUGE .with_label_values(&[collection, "ec_shards"]) .inc(); @@ -1779,6 +1798,94 @@ mod tests { } } + /// A refused shard (a 0-byte file beside an index with entries) on the + /// FIRST mount of a volume must not leave the just-created zero-shard + /// EcVolume registered — it would advertise a mount serving no data, + /// pin the .ecx/.ecj descriptors, and make placement's mounted tier + /// prefer this disk. A volume that already holds shards keeps them + /// (the mount RPC's pre-existing first-error-aborts contract). + #[test] + fn test_mount_ec_shards_refused_shard_removes_created_empty_volume() { + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let mut loc = DiskLocation::new( + dir, + dir, + 10, + DiskType::HardDrive, + MinFreeSpace::Percent(1.0), + Vec::new(), + ) + .unwrap(); + + // An index with one 16-byte entry and a 0-byte shard: the mount is + // refused, and the EcVolume created for it must be unregistered. + std::fs::write(format!("{}/pics_9.ecx", dir), [0u8; 16]).unwrap(); + std::fs::write(format!("{}/pics_9.ec00", dir), b"").unwrap(); + let err = loc + .mount_ec_shards(VolumeId(9), "pics", &[0], "") + .expect_err("a 0-byte shard beside an index with entries must refuse the mount"); + assert!( + err.to_string().contains("empty (0 bytes)"), + "want the empty-shard refusal, got: {}", + err + ); + assert!( + loc.find_ec_volume(VolumeId(9)).is_none(), + "a refused first mount must not leave a zero-shard EcVolume registered", + ); + + // With a valid shard mounted, a later refused shard keeps the + // existing registration intact. + std::fs::write(format!("{}/pics_9.ec01", dir), b"good bytes").unwrap(); + loc.mount_ec_shards(VolumeId(9), "pics", &[1], "").unwrap(); + loc.mount_ec_shards(VolumeId(9), "pics", &[0], "") + .expect_err("the 0-byte shard stays refused"); + assert_eq!( + loc.find_ec_volume(VolumeId(9)).map(|v| v.shard_count()), + Some(1), + "an existing volume keeps its valid shards when a later shard is refused", + ); + } + + /// A mount retry re-listing an already mounted shard must keep the + /// existing registration and not bump the ec_shards gauge — the Rust + /// twin of Go's AddEcVolumeShard added=false handling. + #[test] + fn test_mount_ec_shards_duplicate_keeps_registration_and_gauge() { + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let mut loc = DiskLocation::new( + dir, + dir, + 10, + DiskType::HardDrive, + MinFreeSpace::Percent(1.0), + Vec::new(), + ) + .unwrap(); + + // A collection name unique to this test: the gauge is process-global + // and sibling tests running in parallel touch other labels. + std::fs::write(format!("{}/dupmount_11.ec00", dir), b"shard bytes").unwrap(); + let gauge = crate::metrics::VOLUME_GAUGE.with_label_values(&["dupmount", "ec_shards"]); + let before = gauge.get(); + + loc.mount_ec_shards(VolumeId(11), "dupmount", &[0], "").unwrap(); + loc.mount_ec_shards(VolumeId(11), "dupmount", &[0], "") + .expect("a duplicate mount must succeed as a no-op"); + + assert_eq!( + loc.find_ec_volume(VolumeId(11)).map(|v| v.shard_count()), + Some(1), + ); + assert_eq!( + gauge.get(), + before + 1.0, + "the duplicate mount must not bump the ec_shards gauge", + ); + } + #[test] fn test_disk_location_persists_directory_uuid_and_tags() { let tmp = TempDir::new().unwrap(); diff --git a/seaweed-volume/src/storage/erasure_coding/ec_volume.rs b/seaweed-volume/src/storage/erasure_coding/ec_volume.rs index 34c12aaa9..e44659d66 100644 --- a/seaweed-volume/src/storage/erasure_coding/ec_volume.rs +++ b/seaweed-volume/src/storage/erasure_coding/ec_volume.rs @@ -836,6 +836,27 @@ impl EcVolume { )); } shard.open()?; + // A 0-byte shard file beside an index with entries is residue of a + // failed copy or a truncation, not a mountable shard: registering + // it would advertise a size-0 claim that serves nothing and, since + // placement pins re-copies to the owning disk, would keep + // attracting repairs to a file that was never valid. A 0-byte shard + // beside a 0-byte index is different — that is the legitimate + // layout of a volume encoded with no live needles, and it must keep + // mounting. The startup scan already skips 0-byte shard files; this + // covers the mount path. Mirrors AddEcVolumeShard in + // weed/storage/erasure_coding/ec_volume.go. + if shard.file_size() == 0 && self.ecx_file_size > 0 { + let path = shard.file_name(); + shard.close(); + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!( + "ec volume shard {} is empty (0 bytes) but the index has entries: residue of a failed copy, not a mountable shard", + path + ), + )); + } self.shards[id] = Some(shard); Ok(()) } @@ -2101,6 +2122,62 @@ mod tests { assert!(vol.shard_bits().has_shard_id(3)); } + /// A 0-byte shard file beside an index WITH entries is residue of a + /// failed copy: `add_shard` must refuse it and leave the slot + /// unregistered, so no size-0 claim is advertised (and no re-copy is + /// attracted to a file that was never valid). Beside a 0-byte index it + /// is the legitimate empty-volume layout and must keep mounting. The + /// startup scan already skips such files; this covers the mount path. + /// Mirrors TestLoadEcShardRefusesEmptyShardFile in Go. + #[test] + fn test_add_shard_refuses_empty_shard_file() { + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + write_ecx_file( + dir, + "", + VolumeId(1), + &[(NeedleId(1), Offset::from_actual_offset(8), Size(100))], + ); + + let mut vol = EcVolume::new(dir, dir, "", VolumeId(1)).unwrap(); + + let mut shard = EcVolumeShard::new(dir, "", VolumeId(1), 4); + shard.create().unwrap(); + shard.close(); + + let err = vol + .add_shard(EcVolumeShard::new(dir, "", VolumeId(1), 4)) + .expect_err("adding a 0-byte shard file must fail when the index has entries"); + assert!( + err.to_string().contains("empty (0 bytes)"), + "a 0-byte shard should be refused as empty, got: {}", + err + ); + assert_eq!(vol.shard_count(), 0, "a refused shard must not register"); + assert!(!vol.shard_bits().has_shard_id(4)); + } + + /// The legitimate empty-volume layout: a volume encoded with no live + /// needles has a 0-byte .ecx AND 0-byte shards, and mounting it must + /// keep working. + #[test] + fn test_add_shard_accepts_empty_shard_of_empty_volume() { + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + write_ecx_file(dir, "", VolumeId(1), &[]); + + let mut vol = EcVolume::new(dir, dir, "", VolumeId(1)).unwrap(); + + let mut shard = EcVolumeShard::new(dir, "", VolumeId(1), 4); + shard.create().unwrap(); + shard.close(); + + vol.add_shard(EcVolumeShard::new(dir, "", VolumeId(1), 4)) + .expect("a 0-byte shard of an empty volume (0-byte index) must mount"); + assert!(vol.shard_bits().has_shard_id(4)); + } + #[test] fn test_ec_volume_uses_collection_prefixed_vif_config() { let tmp = TempDir::new().unwrap(); diff --git a/weed/storage/disk_location_ec.go b/weed/storage/disk_location_ec.go index 71987b550..4af53e05e 100644 --- a/weed/storage/disk_location_ec.go +++ b/weed/storage/disk_location_ec.go @@ -161,7 +161,28 @@ func (l *DiskLocation) loadEcShardWithIdxDir(collection string, vid needle.Volum } l.ecVolumes[vid] = ecVolume } - ecVolume.AddEcVolumeShard(ecVolumeShard) + added, err := ecVolume.AddEcVolumeShard(ecVolumeShard) + if err != nil { + // The shard could not be registered (e.g. a 0-byte file beside an + // index with entries). Leave nothing behind: close the opened shard, + // and remove the EcVolume if this call just created it and it holds + // no shards — a zero-shard registration would advertise a mount that + // serves no data while pinning its descriptors. + ecVolumeShard.Unmount() // release the gauge the constructor's Mount took + ecVolumeShard.Close() + if !found && len(ecVolume.Shards) == 0 { + delete(l.ecVolumes, vid) + ecVolume.Close() + } + return nil, err + } + if !added { + // Already registered on this disk (a mount retry): the existing + // shard keeps serving; release the duplicate's fd and gauge so + // repeated LoadEcShard calls don't leak either. + ecVolumeShard.Unmount() + ecVolumeShard.Close() + } return ecVolume, nil } diff --git a/weed/storage/disk_location_ec_test.go b/weed/storage/disk_location_ec_test.go index 93401b28b..3ca3765f2 100644 --- a/weed/storage/disk_location_ec_test.go +++ b/weed/storage/disk_location_ec_test.go @@ -3,8 +3,11 @@ package storage import ( "os" "path/filepath" + "strings" "testing" + "github.com/prometheus/client_golang/prometheus/testutil" + "github.com/seaweedfs/seaweedfs/weed/stats" "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" "github.com/seaweedfs/seaweedfs/weed/storage/needle" @@ -769,6 +772,81 @@ func TestLoadAllEcShardsDeletesStaleZeroSizedShards(t *testing.T) { } } +// TestLoadEcShardRefusesEmptyShardFile: the startup scan skips 0-byte shard +// files, but the mount RPC path (MountEcShards -> LoadEcShard) opens the file +// directly. A 0-byte shard beside an index WITH entries is residue of a +// failed copy — registering it would advertise a size-0 claim that serves +// nothing and, with placement pinned to the owning disk, would keep +// attracting re-copies to a file that was never valid. AddEcVolumeShard must +// refuse it and the loader must leave nothing registered. (A 0-byte shard +// beside a 0-byte index is the legitimate empty-volume layout and keeps +// mounting — TestMountEcShards_EmptyEcxMountsSuccessfully.) +func TestLoadEcShardRefusesEmptyShardFile(t *testing.T) { + dir := t.TempDir() + diskLocation := NewDiskLocation(dir, 10, util.MinFreeSpace{}, dir, types.HardDriveType, nil, stats.DefaultDiskIOProbeConfig()) + defer diskLocation.Close() // also stops NewDiskLocation's background goroutine + + // A usable .ecx sits alongside, so without the size gate the load would + // succeed and register the empty shard — the .ecx must not mask the gate. + empty := filepath.Join(dir, "123.ec00") + if f, err := os.Create(empty); err != nil { + t.Fatalf("create %s: %v", empty, err) + } else { + f.Close() + } + if err := os.WriteFile(filepath.Join(dir, "123.ecx"), make([]byte, 16), 0o644); err != nil { + t.Fatalf("seed .ecx: %v", err) + } + + _, err := diskLocation.LoadEcShard("", needle.VolumeId(123), erasure_coding.ShardId(0)) + if err == nil { + t.Fatalf("loading a 0-byte shard file must fail") + } + if !strings.Contains(err.Error(), "empty (0 bytes)") { + t.Fatalf("a 0-byte shard should be refused as empty, got: %v", err) + } + if _, found := diskLocation.FindEcShard(needle.VolumeId(123), erasure_coding.ShardId(0)); found { + t.Errorf("a 0-byte shard file must not register a shard claim") + } + if _, found := diskLocation.FindEcVolume(needle.VolumeId(123)); found { + t.Errorf("a refused shard load must not leave an empty EcVolume registered") + } +} + +// TestLoadEcShardDuplicateReleasesTheNewShard: a mount retry re-loads a shard +// this disk already registered. AddEcVolumeShard keeps the existing shard and +// reports added=false — the loader must then release the duplicate it just +// opened (fd + mount gauge), or every retry leaks both. +func TestLoadEcShardDuplicateReleasesTheNewShard(t *testing.T) { + dir := t.TempDir() + diskLocation := NewDiskLocation(dir, 10, util.MinFreeSpace{}, dir, types.HardDriveType, nil, stats.DefaultDiskIOProbeConfig()) + defer diskLocation.Close() // also stops NewDiskLocation's background goroutine + + if err := os.WriteFile(filepath.Join(dir, "124.ec00"), []byte("shard bytes"), 0o644); err != nil { + t.Fatalf("seed .ec00: %v", err) + } + if err := os.WriteFile(filepath.Join(dir, "124.ecx"), make([]byte, 16), 0o644); err != nil { + t.Fatalf("seed .ecx: %v", err) + } + + gauge := stats.VolumeServerVolumeGauge.WithLabelValues("", "ec_shards") + before := testutil.ToFloat64(gauge) + + if _, err := diskLocation.LoadEcShard("", needle.VolumeId(124), erasure_coding.ShardId(0)); err != nil { + t.Fatalf("first LoadEcShard: %v", err) + } + ecVolume, err := diskLocation.LoadEcShard("", needle.VolumeId(124), erasure_coding.ShardId(0)) + if err != nil { + t.Fatalf("duplicate LoadEcShard must succeed as a no-op: %v", err) + } + if len(ecVolume.Shards) != 1 { + t.Errorf("duplicate load registered %d shards; want 1", len(ecVolume.Shards)) + } + if after := testutil.ToFloat64(gauge); after != before+1 { + t.Errorf("mount gauge at %v after a duplicate load; want %v (the duplicate's Mount must be released)", after, before+1) + } +} + // TestLoadAllEcShardsSplitDirZeroSizedCleanup: the scan merges Directory and // IdxDirectory listings, so a stale zero-sized file in one directory and a // fresh same-named file in the other are different files behind one entry diff --git a/weed/storage/erasure_coding/ec_scrub_local_test.go b/weed/storage/erasure_coding/ec_scrub_local_test.go index af6ccf42c..bd9c5229a 100644 --- a/weed/storage/erasure_coding/ec_scrub_local_test.go +++ b/weed/storage/erasure_coding/ec_scrub_local_test.go @@ -40,7 +40,9 @@ func setupScrubLocalVolume(t *testing.T, shard0 []byte) *erasure_coding.EcVolume if err != nil { t.Fatalf("NewEcVolumeShard: %v", err) } - ecv.AddEcVolumeShard(shard) + if _, err := ecv.AddEcVolumeShard(shard); err != nil { + t.Fatalf("AddEcVolumeShard: %v", err) + } return ecv } diff --git a/weed/storage/erasure_coding/ec_uniform_layout_test.go b/weed/storage/erasure_coding/ec_uniform_layout_test.go index 215fd7bfa..ee5f16b76 100644 --- a/weed/storage/erasure_coding/ec_uniform_layout_test.go +++ b/weed/storage/erasure_coding/ec_uniform_layout_test.go @@ -218,7 +218,9 @@ func TestEcVolumeGeometryFromVif(t *testing.T) { if err != nil { t.Fatalf("NewEcVolumeShard %d: %v", i, err) } - ev.AddEcVolumeShard(shard) + if _, err := ev.AddEcVolumeShard(shard); err != nil { + t.Fatalf("AddEcVolumeShard: %v", err) + } } // Sweep a large extent through the volume's own interval mapping. diff --git a/weed/storage/erasure_coding/ec_volume.go b/weed/storage/erasure_coding/ec_volume.go index cb6463948..d14d6a3ab 100644 --- a/weed/storage/erasure_coding/ec_volume.go +++ b/weed/storage/erasure_coding/ec_volume.go @@ -297,12 +297,25 @@ func NewEcVolume(diskType types.DiskType, dir string, dirIdx string, collection return } -func (ev *EcVolume) AddEcVolumeShard(ecVolumeShard *EcVolumeShard) bool { +func (ev *EcVolume) AddEcVolumeShard(ecVolumeShard *EcVolumeShard) (bool, error) { for _, s := range ev.Shards { if s.ShardId == ecVolumeShard.ShardId { - return false + return false, nil } } + // A 0-byte shard file beside an index with entries is residue of a + // failed copy or a truncation, not a mountable shard: registering it + // would advertise a size-0 claim that serves nothing and, since + // placement pins re-copies to the owning disk, would keep attracting + // repairs to a file that was never valid. A 0-byte shard beside a + // 0-byte index is different — that is the legitimate layout of a + // volume encoded with no live needles, and it must keep mounting. + // The startup scan already skips 0-byte shard files; this covers the + // mount RPC path, which opens the file directly. + if ecVolumeShard.Size() == 0 && ev.ecxFileSize > 0 { + return false, fmt.Errorf("ec volume %d shard %d: shard file is empty (0 bytes) but the index has %d entries: residue of a failed copy, not a mountable shard", + ev.VolumeId, ecVolumeShard.ShardId, ev.ecxFileSize/types.NeedleMapEntrySize) + } ev.Shards = append(ev.Shards, ecVolumeShard) slices.SortFunc(ev.Shards, func(a, b *EcVolumeShard) int { if a.VolumeId != b.VolumeId { @@ -310,7 +323,7 @@ func (ev *EcVolume) AddEcVolumeShard(ecVolumeShard *EcVolumeShard) bool { } return int(a.ShardId - b.ShardId) }) - return true + return true, nil } func (ev *EcVolume) DeleteEcVolumeShard(shardId ShardId) (ecVolumeShard *EcVolumeShard, deleted bool) {