From f410d975c7e52bcc568e4fa023f97d9523ae6bca Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Mon, 1 Jun 2026 19:22:14 -0700 Subject: [PATCH] fix(ec): resolve EC data-shard count from the volume's .vif on reboot (#9779) * fix(ec): resolve EC data-shard count from the volume's .vif on reboot A volume server never loads a cluster EC config into memory, so startup decisions that assumed 10 data shards mishandled volumes whose .vif records a different ratio: - validateEcVolume sized the expected shard against 10 data shards and required >=10 local shards, so a volume with a non-default ratio and a coexisting .dat could be wiped on reboot. Read the ratio from the .vif. - pruneIncompleteEcWithSiblingDat used the hardcoded 10-shard threshold, so a full data set for a non-default ratio with a healthy sibling .dat was wiped as a partial leftover. Use the EcVolume's .vif-derived ratio. Behavior is unchanged for the standard 10+4 layout (the .vif resolves to 10). Adds storage-level reboot tests. * ec: avoid per-call allocations in ecDataShardsFromVif Address review: the helper runs once per EC volume at startup. Replace the slice+map dedup of the two dirs with direct conditional checks via a small ecDataShardsFromVifDir helper, eliminating the heap allocations and GC pressure when loading many volumes. --- weed/storage/disk_location_ec.go | 57 ++- .../store_ec_9plus3_reboot_repro_test.go | 324 ++++++++++++++++++ weed/storage/store_ec_reconcile.go | 14 +- 3 files changed, 385 insertions(+), 10 deletions(-) create mode 100644 weed/storage/store_ec_9plus3_reboot_repro_test.go diff --git a/weed/storage/disk_location_ec.go b/weed/storage/disk_location_ec.go index c41afed38..c6e4c1824 100644 --- a/weed/storage/disk_location_ec.go +++ b/weed/storage/disk_location_ec.go @@ -13,6 +13,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/volume_info" ) var ( @@ -466,13 +467,19 @@ func (l *DiskLocation) validateEcVolume(collection string, vid needle.VolumeId) var expectedShardSize int64 = -1 datExists := false + // Resolve the data-shard count from the volume's own .vif, which is + // written at encode time and travels with the volume. The volume + // server never loads the cluster EC config into memory, so the OSS + // default (10) would size a custom-ratio volume's shards (e.g. 9+3) + // for 10 data shards, fail the size check, and wrongly delete a + // healthy volume on reboot. Fall back to the default only when the + // .vif carries no EC shard config. + dataShards := l.ecDataShardsFromVif(collection, vid) + // If .dat file exists, compute exact expected shard size from it. - // Pass the build's default data-shard count; calculateExpectedShardSize - // takes it as a parameter so tests / enterprise builds can supply - // their own. if datFileInfo, err := os.Stat(datFileName); err == nil { datExists = true - expectedShardSize = calculateExpectedShardSize(datFileInfo.Size(), erasure_coding.DataShardsCount) + expectedShardSize = calculateExpectedShardSize(datFileInfo.Size(), dataShards) } else if !os.IsNotExist(err) { // If stat fails with unexpected error (permission, I/O), fail validation // Don't treat this as "distributed EC" - it could be a temporary error @@ -525,17 +532,51 @@ func (l *DiskLocation) validateEcVolume(collection string, vid needle.VolumeId) return true } - // If .dat file exists, we need at least DataShardsCount shards locally - // Otherwise it's an incomplete EC encoding that should be cleaned up - if shardCount < erasure_coding.DataShardsCount { + // If .dat file exists, we need at least DataShards shards locally + // for the volume's configured ratio. Otherwise it's an incomplete + // EC encoding that should be cleaned up. + if shardCount < dataShards { glog.Warningf("EC volume %d has .dat file but only %d shards (need at least %d for local EC)", - vid, shardCount, erasure_coding.DataShardsCount) + vid, shardCount, dataShards) return false } return true } +// ecDataShardsFromVif resolves the data-shard count for an EC volume from +// its own .vif (EcShardConfig), checking the data dir then the idx dir. The +// .vif is the source of truth for custom ratios on the volume server, which +// never holds the cluster EC config in memory. Falls back to the default +// ratio when the .vif carries no EC shard config. +func (l *DiskLocation) ecDataShardsFromVif(collection string, vid needle.VolumeId) int { + // At most two dirs to check; avoid slice/map allocations on this + // per-volume startup path. + if l.Directory != "" { + if ds := ecDataShardsFromVifDir(collection, l.Directory, vid); ds > 0 { + return ds + } + } + if l.IdxDirectory != "" && l.IdxDirectory != l.Directory { + if ds := ecDataShardsFromVifDir(collection, l.IdxDirectory, vid); ds > 0 { + return ds + } + } + return erasure_coding.DataShardsCount +} + +// ecDataShardsFromVifDir returns the .vif EcShardConfig data-shard count for +// (collection, vid) under dir, or 0 when absent / not custom. +func ecDataShardsFromVifDir(collection, dir string, vid needle.VolumeId) int { + vifName := erasure_coding.EcShardFileName(collection, dir, int(vid)) + ".vif" + if vi, _, found, _ := volume_info.MaybeLoadVolumeInfo(vifName); found && vi.EcShardConfig != nil { + if ds := int(vi.EcShardConfig.DataShards); ds > 0 { + return ds + } + } + return 0 +} + // removeEcVolumeFiles removes all EC-related files for a volume func (l *DiskLocation) removeEcVolumeFiles(collection string, vid needle.VolumeId) { baseFileName := erasure_coding.EcShardFileName(collection, l.Directory, int(vid)) diff --git a/weed/storage/store_ec_9plus3_reboot_repro_test.go b/weed/storage/store_ec_9plus3_reboot_repro_test.go new file mode 100644 index 000000000..f657160c1 --- /dev/null +++ b/weed/storage/store_ec_9plus3_reboot_repro_test.go @@ -0,0 +1,324 @@ +package storage + +import ( + "os" + "path/filepath" + "strconv" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" + "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/types" + "github.com/seaweedfs/seaweedfs/weed/storage/volume_info" + "github.com/seaweedfs/seaweedfs/weed/util" +) + +// drainStoreChans consumes the heartbeat-style channels so NewStore's +// notify handler never blocks during loading. +func drainStoreChans(t *testing.T, store *Store) { + t.Helper() + done := make(chan struct{}) + go func() { + for { + select { + case <-store.NewVolumesChan: + case <-store.NewEcShardsChan: + case <-store.DeletedVolumesChan: + case <-store.DeletedEcShardsChan: + case <-store.StateUpdateChan: + case <-done: + return + } + } + }() + t.Cleanup(func() { + store.Close() + close(done) + }) +} + +// TestEc9plus3MultiDiskRebootLoadsAllShards mirrors the reported setup: +// enterprise EC 9+3, one volume server with 6 disks, a distributed volume's +// 12 shards spread across the disks while the .ecx/.ecj/.vif sidecars live on +// a single disk (the distribute "owner"). After a reboot (NewStore), every +// shard must load — the orphan-shard reconciler is supposed to mirror the +// sidecars onto the other disks and mount their shards. +func TestEc9plus3MultiDiskRebootLoadsAllShards(t *testing.T) { + // Deliberately do NOT call SetGlobalECConfig: a real volume server never + // loads the cluster EC config into memory, so the 9+3 ratio must come + // from each volume's .vif. Relying on the .vif keeps these faithful to + // production and guards the fixes that read the ratio from the .vif. + + const dataShards, parityShards = 9, 3 + const totalShards = dataShards + parityShards // 12 + + tempDir := t.TempDir() + const diskCount = 6 + dirs := make([]string, diskCount) + for i := range dirs { + dirs[i] = filepath.Join(tempDir, "disk"+strconv.Itoa(i)) + if err := os.MkdirAll(dirs[i], 0o755); err != nil { + t.Fatalf("mkdir %s: %v", dirs[i], err) + } + } + + collection := "loki" + vid := needle.VolumeId(7) + const datSize int64 = 64 * 1024 * 1024 + expectedShardSize := calculateExpectedShardSize(datSize, dataShards) + + // 12 shards, 2 per disk: disk0→{0,1}, disk1→{2,3}, ... disk5→{10,11}. + shardToDisk := func(shardId int) int { return shardId / 2 } + + for s := 0; s < totalShards; s++ { + dir := dirs[shardToDisk(s)] + base := erasure_coding.EcShardFileName(collection, dir, int(vid)) + f, err := os.Create(base + erasure_coding.ToExt(s)) + if err != nil { + t.Fatalf("create shard %d: %v", s, err) + } + if err := f.Truncate(expectedShardSize); err != nil { + f.Close() + t.Fatalf("truncate shard %d: %v", s, err) + } + f.Close() + } + + // Sidecars live ONLY on disk0 (the distribute owner). No .dat anywhere — + // this is a fully distributed EC volume. + base0 := erasure_coding.EcShardFileName(collection, dirs[0], int(vid)) + if err := os.WriteFile(base0+".ecx", make([]byte, 20), 0o644); err != nil { + t.Fatalf("write .ecx: %v", err) + } + if err := os.WriteFile(base0+".ecj", nil, 0o644); err != nil { + t.Fatalf("write .ecj: %v", err) + } + if err := volume_info.SaveVolumeInfo(base0+".vif", &volume_server_pb.VolumeInfo{ + Version: uint32(needle.Version3), + DatFileSize: datSize, + EcShardConfig: &volume_server_pb.EcShardConfig{ + DataShards: dataShards, + ParityShards: parityShards, + }, + }); err != nil { + t.Fatalf("save .vif: %v", err) + } + + diskTypes := make([]types.DiskType, diskCount) + maxCounts := make([]int32, diskCount) + minFree := make([]util.MinFreeSpace, diskCount) + for i := range diskTypes { + diskTypes[i] = types.HardDriveType + maxCounts[i] = 100 + } + + store := NewStore(nil, "localhost", 8080, 18080, "http://localhost:8080", "store-id", + dirs, maxCounts, minFree, "", NeedleMapInMemory, diskTypes, nil, 3) + drainStoreChans(t, store) + + missing := []int{} + for s := 0; s < totalShards; s++ { + loc := store.Locations[shardToDisk(s)] + if _, found := loc.FindEcShard(vid, erasure_coding.ShardId(s)); !found { + missing = append(missing, s) + } + } + if len(missing) > 0 { + t.Errorf("after reboot, %d/%d shards missing for 9+3 volume %d: shards %v not loaded", + len(missing), totalShards, vid, missing) + } + + // Also confirm the heartbeat would report all 12 shards (master's view). + hb := store.CollectErasureCodingHeartbeat() + reported := map[int]bool{} + for _, msg := range hb.EcShards { + if needle.VolumeId(msg.Id) != vid { + continue + } + si := erasure_coding.ShardsInfoFromVolumeEcShardInformationMessage(msg) + for _, id := range si.Ids() { + reported[int(id)] = true + } + } + if len(reported) != totalShards { + t.Errorf("heartbeat reports %d/%d shards for volume %d (master would show %d missing)", + len(reported), totalShards, vid, totalShards-len(reported)) + } +} + +// TestEc9plus3ValidateKeepsVolumeWithDat reproduces the reboot-time deletion +// path for a custom 9+3 ratio. When a .dat coexists with the EC shards on a +// disk (a just-encoded volume whose original .dat has not been deleted yet, or +// a local-EC layout), startup loading runs validateEcVolume, which derives the +// expected per-shard size from the .dat size and the DATA-shard count. +// +// At 4.29 that count was the compiled-in OSS default (10); on a 9+3 volume the +// shards are sized for 9 data shards, so the size check fails, validation +// returns false, and handleFoundEcxFile deletes every EC file for the volume — +// i.e. the shards "go missing" on the next reboot. HEAD derives the count from +// GetECConfig(collection), so the volume survives. +func TestEc9plus3ValidateKeepsVolumeWithDat(t *testing.T) { + // Deliberately do NOT call SetGlobalECConfig: a real volume server never + // loads the cluster EC config into memory, so the 9+3 ratio must come + // from each volume's .vif. Relying on the .vif keeps these faithful to + // production and guards the fixes that read the ratio from the .vif. + + const dataShards, parityShards = 9, 3 + const totalShards = dataShards + parityShards + + // The 4.29 bug is a wrong expected shard size; prove the two ratios really + // do disagree for a realistic .dat size, so the size check would reject. + const datSize int64 = 64 * 1024 * 1024 + sizeFor9 := calculateExpectedShardSize(datSize, dataShards) + sizeFor10 := calculateExpectedShardSize(datSize, erasure_coding.DataShardsCount) // 10 + if sizeFor9 == sizeFor10 { + t.Fatalf("test premise broken: shard size for 9 and 10 data shards must differ (got %d)", sizeFor9) + } + + dir := t.TempDir() + collection := "loki" + vid := needle.VolumeId(42) + base := erasure_coding.EcShardFileName(collection, dir, int(vid)) + + // .dat sized as the encode source. + if err := os.WriteFile(base+".dat", make([]byte, datSize), 0o644); err != nil { + t.Fatalf("write .dat: %v", err) + } + // 12 shards, each sized for the real 9-data-shard layout. + for s := 0; s < totalShards; s++ { + f, err := os.Create(base + erasure_coding.ToExt(s)) + if err != nil { + t.Fatalf("create shard %d: %v", s, err) + } + if err := f.Truncate(sizeFor9); err != nil { + f.Close() + t.Fatalf("truncate shard %d: %v", s, err) + } + f.Close() + } + if err := os.WriteFile(base+".ecx", make([]byte, 20), 0o644); err != nil { + t.Fatalf("write .ecx: %v", err) + } + if err := os.WriteFile(base+".ecj", nil, 0o644); err != nil { + t.Fatalf("write .ecj: %v", err) + } + if err := volume_info.SaveVolumeInfo(base+".vif", &volume_server_pb.VolumeInfo{ + Version: uint32(needle.Version3), + DatFileSize: datSize, + EcShardConfig: &volume_server_pb.EcShardConfig{ + DataShards: dataShards, + ParityShards: parityShards, + }, + }); err != nil { + t.Fatalf("save .vif: %v", err) + } + + store := NewStore(nil, "localhost", 8080, 18080, "http://localhost:8080", "store-id", + []string{dir}, []int32{100}, []util.MinFreeSpace{{}}, "", + NeedleMapInMemory, []types.DiskType{types.HardDriveType}, nil, 3) + drainStoreChans(t, store) + + // HEAD: validateEcVolume uses the 9+3 ratio, the size check passes, and the + // shards stay on disk + get loaded. 4.29 would have deleted them here. + for s := 0; s < totalShards; s++ { + if _, err := os.Stat(base + erasure_coding.ToExt(s)); err != nil { + t.Errorf("shard %d was deleted at startup (ratio-mismatch cleanup): %v", s, err) + } + } + loc := store.Locations[0] + loaded := 0 + for s := 0; s < totalShards; s++ { + if _, found := loc.FindEcShard(vid, erasure_coding.ShardId(s)); found { + loaded++ + } + } + if loaded != totalShards { + t.Errorf("loaded %d/%d shards for 9+3 volume with .dat present; expected all to survive validation", loaded, totalShards) + } +} + +// TestEc9plus3PruneKeepsFullDataShardSet reproduces a CURRENT (HEAD) custom-ratio +// bug. pruneIncompleteEcWithSiblingDat deletes EC shards on a disk when the disk +// holds fewer than DataShards shards AND a healthy .dat for the same volume lives +// on a sibling disk (issue 9478 cleanup). The threshold is hardcoded to the OSS +// default DataShardsCount (10), so for a 9+3 volume a disk holding a COMPLETE +// 9-data-shard set (fully recoverable on its own) is treated as a partial +// leftover and wiped on the next reboot. +// +// With the volume's real ratio (9), 9 >= 9, so the set must be kept. +func TestEc9plus3PruneKeepsFullDataShardSet(t *testing.T) { + // Deliberately do NOT call SetGlobalECConfig: a real volume server never + // loads the cluster EC config into memory, so the 9+3 ratio must come + // from each volume's .vif. Relying on the .vif keeps these faithful to + // production and guards the fixes that read the ratio from the .vif. + + const dataShards, parityShards = 9, 3 + + tempDir := t.TempDir() + dir0 := filepath.Join(tempDir, "disk0") // 9 data shards + index sidecars + dir1 := filepath.Join(tempDir, "disk1") // healthy .dat for the same volume + for _, d := range []string{dir0, dir1} { + if err := os.MkdirAll(d, 0o755); err != nil { + t.Fatalf("mkdir %s: %v", d, err) + } + } + + collection := "loki" + vid := needle.VolumeId(77) + const datSize int64 = 64 * 1024 * 1024 + shardSize := calculateExpectedShardSize(datSize, dataShards) + + base0 := erasure_coding.EcShardFileName(collection, dir0, int(vid)) + for s := 0; s < dataShards; s++ { // shards 0..8 — a complete data set + f, err := os.Create(base0 + erasure_coding.ToExt(s)) + if err != nil { + t.Fatalf("create shard %d: %v", s, err) + } + if err := f.Truncate(shardSize); err != nil { + f.Close() + t.Fatalf("truncate shard %d: %v", s, err) + } + f.Close() + } + if err := os.WriteFile(base0+".ecx", make([]byte, 20), 0o644); err != nil { + t.Fatalf("write .ecx: %v", err) + } + if err := os.WriteFile(base0+".ecj", nil, 0o644); err != nil { + t.Fatalf("write .ecj: %v", err) + } + if err := volume_info.SaveVolumeInfo(base0+".vif", &volume_server_pb.VolumeInfo{ + Version: uint32(needle.Version3), + DatFileSize: datSize, + EcShardConfig: &volume_server_pb.EcShardConfig{ + DataShards: dataShards, + ParityShards: parityShards, + }, + }); err != nil { + t.Fatalf("save .vif: %v", err) + } + + // Sibling disk holds a credible .dat (>= recorded datFileSize). + base1 := erasure_coding.EcShardFileName(collection, dir1, int(vid)) + if err := os.WriteFile(base1+".dat", make([]byte, datSize), 0o644); err != nil { + t.Fatalf("write .dat: %v", err) + } + + store := NewStore(nil, "localhost", 8080, 18080, "http://localhost:8080", "store-id", + []string{dir0, dir1}, []int32{100, 100}, []util.MinFreeSpace{{}, {}}, "", + NeedleMapInMemory, []types.DiskType{types.HardDriveType, types.HardDriveType}, nil, 3) + drainStoreChans(t, store) + + loc0 := store.Locations[0] + kept := 0 + for s := 0; s < dataShards; s++ { + if _, found := loc0.FindEcShard(vid, erasure_coding.ShardId(s)); found { + kept++ + } + } + if kept != dataShards { + t.Errorf("a complete %d-data-shard set was pruned at reboot: only %d/%d shards survived "+ + "(pruneIncompleteEcWithSiblingDat used the hardcoded 10-shard threshold instead of the volume's %d)", + dataShards, kept, dataShards, dataShards) + } +} diff --git a/weed/storage/store_ec_reconcile.go b/weed/storage/store_ec_reconcile.go index 391a4d451..a5b36146d 100644 --- a/weed/storage/store_ec_reconcile.go +++ b/weed/storage/store_ec_reconcile.go @@ -247,12 +247,21 @@ func (s *Store) pruneIncompleteEcWithSiblingDat() { messages []*master_pb.VolumeEcShardInformationMessage datDir string shardCount int + dataShards int } var victims []victim loc.ecVolumesLock.RLock() for vid, ev := range loc.ecVolumes { shardCount := len(ev.Shards) - if shardCount >= erasure_coding.DataShardsCount { + // Use the volume's configured data-shard count, not the OSS + // default: a disk holding a full data set for a custom ratio + // (e.g. 9 shards of a 9+3 volume) is independently recoverable + // and must not be mistaken for a partial leftover and wiped. + dataShards := erasure_coding.DataShardsCount + if ev.ECContext != nil && ev.ECContext.DataShards > 0 { + dataShards = ev.ECContext.DataShards + } + if shardCount >= dataShards { continue } key := ecKeyForReconcile{collection: ev.Collection, vid: vid} @@ -279,13 +288,14 @@ func (s *Store) pruneIncompleteEcWithSiblingDat() { messages: ev.ToVolumeEcShardInformationMessage(uint32(diskId)), datDir: owner.location.Directory, shardCount: shardCount, + dataShards: dataShards, }) } loc.ecVolumesLock.RUnlock() for _, v := range victims { glog.Warningf("ec volume %d (collection=%q) on %s has only %d shards (need %d) while a healthy .dat exists on sibling disk %s; cleaning up leftover EC files (issue 9478)", - v.vid, v.collection, loc.Directory, v.shardCount, erasure_coding.DataShardsCount, v.datDir) + v.vid, v.collection, loc.Directory, v.shardCount, v.dataShards, v.datDir) loc.unloadEcVolume(v.vid) loc.removeEcVolumeFiles(v.collection, v.vid) for _, msg := range v.messages {