From 94f8e2caf9a52293ebb8cc47c162d250d0ef89aa Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Thu, 13 Aug 2026 21:38:22 -0700 Subject: [PATCH] EC: handle zero-sized shard files uniformly (moves, rebuilds, startup cleanup) (#10753) * volume_move: treat zero-sized EC shards as absent in move verification A zero-sized shard file is residue of a failed operation (issue 10730), not a shard - but VerifyEcShards only checked presence, so a copy that landed as an empty file passed verification and the source was deleted behind it. Size zero now reads as absent, with a distinct error naming the zero-sized shard so the operator can tell a broken copy from a missing one. * storage: exclude zero-sized EC shards from rebuilds and clean up stale ones The reproducer in issue 10730: a zero-sized shard file left by a failed operation was selected as a Reed-Solomon input and failed the whole rebuild with an input size mismatch, because input discovery checked existence, not substance. - RebuildEcFiles treats a zero-sized shard file as missing and regenerates over it in place (the reclassified-corrupt path: temp file beside the residue, atomic rename). - The startup/rescan shard loader, which always skipped zero-sized files, now deletes them once they are older than an hour - young enough files can be an in-flight copy's just-created file, since the same scan runs from LoadNewVolumes while serving. Regression tests: a rebuild with one emptied shard regenerates it byte-identical; the loader deletes a stale zero-sized shard and leaves a fresh one alone. * storage: age-check each zero-shard cleanup candidate individually The shard scan merges the data and idx directory listings, so the age-checked entry and a deletion candidate can be different files sharing one name - a stale zero-sized file in one directory next to a fresh same-named file in the other (possibly an in-flight copy's just-created one) could get the fresh file deleted. Each candidate's own modification time now decides, both directories are handled in one pass, and the split-directory case is pinned by a test. --- weed/operation/volume_move/ec_move.go | 17 ++++- weed/operation/volume_move/ec_move_test.go | 22 +++++- weed/storage/disk_location_ec.go | 34 +++++++++ weed/storage/disk_location_ec_test.go | 72 +++++++++++++++++++ .../storage/erasure_coding/ec_decoder_test.go | 14 ++-- weed/storage/erasure_coding/ec_encoder.go | 35 ++++++--- .../erasure_coding/ec_rebuild_safety_test.go | 45 ++++++++++++ .../needle/needle_parse_upload_alloc_test.go | 10 +-- weed/storage/volume_io_error_test.go | 10 +-- 9 files changed, 228 insertions(+), 31 deletions(-) diff --git a/weed/operation/volume_move/ec_move.go b/weed/operation/volume_move/ec_move.go index 4c0929702..bd06d55a3 100644 --- a/weed/operation/volume_move/ec_move.go +++ b/weed/operation/volume_move/ec_move.go @@ -139,17 +139,28 @@ func (m *Mover) VerifyEcShards(ctx context.Context, volumeId needle.VolumeId, se if err != nil { return fmt.Errorf("verify EC shard(s) on %s for volume %d: %v", server, volumeId, err) } - var bits erasure_coding.ShardBits + var bits, zeroSized erasure_coding.ShardBits for _, s := range resp.EcShardInfos { if s.VolumeId != uint32(volumeId) || s.ShardId >= erasure_coding.MaxShardCount { continue } + // A zero-sized shard is residue of a failed operation, not a + // shard; counting it as present would let a broken copy pass + // verification and the source be deleted behind it. + if s.Size <= 0 { + zeroSized = zeroSized.Set(erasure_coding.ShardId(s.ShardId)) + continue + } bits = bits.Set(erasure_coding.ShardId(s.ShardId)) } for _, sid := range shardIds { - if !bits.Has(sid) { - return fmt.Errorf("%s missing EC shard %d.%d after copy/mount; keeping source", server, volumeId, sid) + if bits.Has(sid) { + continue } + if zeroSized.Has(sid) { + return fmt.Errorf("%s has a zero-sized EC shard %d.%d after copy/mount; keeping source", server, volumeId, sid) + } + return fmt.Errorf("%s missing EC shard %d.%d after copy/mount; keeping source", server, volumeId, sid) } return nil }) diff --git a/weed/operation/volume_move/ec_move_test.go b/weed/operation/volume_move/ec_move_test.go index c71bd01fa..2e4756ff6 100644 --- a/weed/operation/volume_move/ec_move_test.go +++ b/weed/operation/volume_move/ec_move_test.go @@ -24,7 +24,7 @@ func ecMove(shardIds ...erasure_coding.ShardId) EcShardMove { func dstShards(shardIds ...uint32) []*volume_server_pb.EcShardInfo { var infos []*volume_server_pb.EcShardInfo for _, sid := range shardIds { - infos = append(infos, &volume_server_pb.EcShardInfo{VolumeId: 7, ShardId: sid}) + infos = append(infos, &volume_server_pb.EcShardInfo{VolumeId: 7, ShardId: sid, Size: 1024}) } return infos } @@ -71,6 +71,26 @@ func TestMoveEcShardsVerifyFailureKeepsSource(t *testing.T) { } } +func TestMoveEcShardsZeroSizedDestinationKeepsSource(t *testing.T) { + // A zero-sized shard on the destination is residue of a failed operation + // (seaweedfs issue 10730); verification must not count it as a delivered + // shard, or the source is deleted behind a broken copy. + cluster := newFakeCluster() + cluster.ecShards[string(dstAddr)] = dstShards(3, 4) + cluster.ecShards[string(dstAddr)][1].Size = 0 // shard 4 landed as an empty file + + err := cluster.mover().MoveEcShards(context.Background(), ecMove(3, 4), EcMoveOptions{}) + if err == nil || !strings.Contains(err.Error(), "zero-sized EC shard 7.4") { + t.Fatalf("expected zero-sized-shard rejection, got: %v", err) + } + + for _, call := range cluster.callList() { + if call == "src:8080 VolumeEcShardsUnmount" || call == "src:8080 VolumeEcShardsDelete" { + t.Fatalf("source touched despite zero-sized destination shard: %v", cluster.callList()) + } + } +} + func TestMoveEcShardsRejectsSameServer(t *testing.T) { // The second target is the same server written with an explicit grpc port; // the guard must see through the representation difference. diff --git a/weed/storage/disk_location_ec.go b/weed/storage/disk_location_ec.go index 40b7ec987..c306fcd11 100644 --- a/weed/storage/disk_location_ec.go +++ b/weed/storage/disk_location_ec.go @@ -7,6 +7,7 @@ import ( "regexp" "strconv" "strings" + "time" "slices" @@ -210,6 +211,11 @@ func (l *DiskLocation) loadEcShards(shards []string, collection string, vid need return nil } +// staleZeroShardAge guards the zero-sized-shard cleanup in loadAllEcShards: a +// just-created file of an in-flight VolumeEcShardsCopy is legitimately empty +// for a moment, while failed-operation residue is old by the next scan. +const staleZeroShardAge = time.Hour + func (l *DiskLocation) loadAllEcShards(onShardLoad func(collection string, vid needle.VolumeId, shardId erasure_coding.ShardId, ecVolume *erasure_coding.EcVolume)) (err error) { dirEntries, err := os.ReadDir(l.Directory) @@ -257,6 +263,34 @@ func (l *DiskLocation) loadAllEcShards(onShardLoad func(collection string, vid n continue } + // A zero-sized shard file is residue of a failed operation (never + // loaded, but its presence poisons later rebuilds, which select + // inputs from the directory). Delete it once it is old enough that + // it cannot be an in-flight copy's just-created file — this scan + // also runs from LoadNewVolumes while the server is serving. The + // scan merges the Directory and IdxDirectory listings, so the entry + // and a candidate path can be different files with one name: each + // candidate's own age decides, and a same-named fresh file (possibly + // an in-flight copy's just-created one) always survives. + if re.MatchString(ext) && info.Size() == 0 { + for _, dir := range []string{l.Directory, l.IdxDirectory} { + p := path.Join(dir, name) + fi, statErr := os.Stat(p) + if statErr != nil || fi.IsDir() || fi.Size() != 0 { + continue + } + if time.Since(fi.ModTime()) <= staleZeroShardAge { + continue + } + if rmErr := os.Remove(p); rmErr != nil { + glog.Warningf("remove zero-sized ec shard %s: %v", p, rmErr) + } else { + glog.Warningf("removed zero-sized ec shard %s (residue of a failed operation)", p) + } + } + continue + } + // 0 byte files should be only appearing erroneously for ec data files // so we ignore them if re.MatchString(ext) && info.Size() > 0 { diff --git a/weed/storage/disk_location_ec_test.go b/weed/storage/disk_location_ec_test.go index f62c9beb6..6fcc7b591 100644 --- a/weed/storage/disk_location_ec_test.go +++ b/weed/storage/disk_location_ec_test.go @@ -5,10 +5,12 @@ import ( "path/filepath" "testing" + "github.com/seaweedfs/seaweedfs/weed/stats" "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/util" + "time" ) // closeEcVolumes closes all EC volumes in the given DiskLocation to release file handles. @@ -747,3 +749,73 @@ func TestLoadExistingVolumeSkipsVifWhenEcxPresent(t *testing.T) { }) } } + +// TestLoadAllEcShardsDeletesStaleZeroSizedShards: zero-sized shard files that +// are old enough to be failed-operation residue (issue 10730) are deleted by +// the scan; a fresh zero-sized file (possibly an in-flight copy's just-created +// file) is left alone. +func TestLoadAllEcShardsDeletesStaleZeroSizedShards(t *testing.T) { + dir := t.TempDir() + diskLocation := NewDiskLocation(dir, 10, util.MinFreeSpace{}, dir, types.HardDriveType, nil, stats.DefaultDiskIOProbeConfig()) + + stale := filepath.Join(dir, "123.ec00") + fresh := filepath.Join(dir, "123.ec01") + for _, p := range []string{stale, fresh} { + if f, err := os.Create(p); err != nil { + t.Fatalf("create %s: %v", p, err) + } else { + f.Close() + } + } + oldTime := time.Now().Add(-2 * staleZeroShardAge) + if err := os.Chtimes(stale, oldTime, oldTime); err != nil { + t.Fatalf("chtimes: %v", err) + } + + if err := diskLocation.loadAllEcShards(nil); err != nil { + t.Fatalf("loadAllEcShards: %v", err) + } + + if _, err := os.Stat(stale); !os.IsNotExist(err) { + t.Errorf("stale zero-sized shard %s not deleted (err=%v)", stale, err) + } + if _, err := os.Stat(fresh); err != nil { + t.Errorf("fresh zero-sized shard %s must survive the scan: %v", fresh, err) + } +} + +// 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 +// name. Each candidate's own age must decide: the stale one is deleted, the +// fresh one (possibly an in-flight copy's just-created file) survives. +func TestLoadAllEcShardsSplitDirZeroSizedCleanup(t *testing.T) { + dataDir := t.TempDir() + idxDir := t.TempDir() + diskLocation := NewDiskLocation(dataDir, 10, util.MinFreeSpace{}, idxDir, types.HardDriveType, nil, stats.DefaultDiskIOProbeConfig()) + + fresh := filepath.Join(dataDir, "124.ec00") + stale := filepath.Join(idxDir, "124.ec00") + for _, p := range []string{fresh, stale} { + if f, err := os.Create(p); err != nil { + t.Fatalf("create %s: %v", p, err) + } else { + f.Close() + } + } + oldTime := time.Now().Add(-2 * staleZeroShardAge) + if err := os.Chtimes(stale, oldTime, oldTime); err != nil { + t.Fatalf("chtimes: %v", err) + } + + if err := diskLocation.loadAllEcShards(nil); err != nil { + t.Fatalf("loadAllEcShards: %v", err) + } + + if _, err := os.Stat(fresh); err != nil { + t.Errorf("fresh zero-sized shard %s must survive the scan: %v", fresh, err) + } + if _, err := os.Stat(stale); !os.IsNotExist(err) { + t.Errorf("stale zero-sized shard %s not deleted (err=%v)", stale, err) + } +} diff --git a/weed/storage/erasure_coding/ec_decoder_test.go b/weed/storage/erasure_coding/ec_decoder_test.go index bbdbb353d..8c655d7fa 100644 --- a/weed/storage/erasure_coding/ec_decoder_test.go +++ b/weed/storage/erasure_coding/ec_decoder_test.go @@ -542,13 +542,13 @@ func TestEcxFileDeletionWithSeparateHandles(t *testing.T) { // are journaled to .ecj and tracked in an in-memory set — so the // durability chain decode relies on is: // -// 1. DeleteNeedleFromEcx appends the needle id to .ecj and fsyncs it. -// 2. Runtime reads via FindNeedleFromEcx consult the in-memory set and -// return TombstoneFileSize even though the sealed .ecx record on -// disk still shows the original size. -// 3. ec.decode later closes the EcVolume and calls RebuildEcxFile on -// the now-quiescent files, which walks .ecj and writes tombstones -// into .ecx. CopyFile then reads the rebuilt .ecx. +// 1. DeleteNeedleFromEcx appends the needle id to .ecj and fsyncs it. +// 2. Runtime reads via FindNeedleFromEcx consult the in-memory set and +// return TombstoneFileSize even though the sealed .ecx record on +// disk still shows the original size. +// 3. ec.decode later closes the EcVolume and calls RebuildEcxFile on +// the now-quiescent files, which walks .ecj and writes tombstones +// into .ecx. CopyFile then reads the rebuilt .ecx. // // This test exercises the full chain on a tempdir fixture. func TestEcVolumeDeleteDurableToJournal(t *testing.T) { diff --git a/weed/storage/erasure_coding/ec_encoder.go b/weed/storage/erasure_coding/ec_encoder.go index 8167c567e..927a50d86 100644 --- a/weed/storage/erasure_coding/ec_encoder.go +++ b/weed/storage/erasure_coding/ec_encoder.go @@ -167,21 +167,33 @@ func generateMissingEcFiles(baseFileName string, bufferSize int, largeBlockSize shardPaths := make([]string, ctx.Total()) // non-empty for present shards (also the in-place output for a reclassified-corrupt shard) inputFiles := make([]*os.File, ctx.Total()) presentCount := 0 + var zeroSized []int for shardId := 0; shardId < ctx.Total(); shardId++ { ext := ctx.ToExt(shardId) shardPath := findShardFile(baseFileName, ext, additionalDirs) - if shardPath != "" { - shardHasData[shardId] = true - shardPaths[shardId] = shardPath - inputFiles[shardId], err = os.OpenFile(shardPath, os.O_RDONLY, 0) - if err != nil { - return nil, err - } - defer inputFiles[shardId].Close() - presentCount++ - } else { + if shardPath == "" { generatedShardIds = append(generatedShardIds, uint32(shardId)) + continue } + if fi, statErr := os.Stat(shardPath); statErr == nil && fi.Size() == 0 { + // A zero-sized shard file is residue of a failed operation, not a + // shard; feeding it to Reed-Solomon fails the whole rebuild with a + // size mismatch. Treat it as missing and regenerate over it in + // place, like a reclassified-corrupt shard. + glog.Warningf("shard %d for %s is zero-sized at %s; excluding from rebuild inputs and regenerating", shardId, baseFileName, shardPath) + shardPaths[shardId] = shardPath + zeroSized = append(zeroSized, shardId) + generatedShardIds = append(generatedShardIds, uint32(shardId)) + continue + } + shardHasData[shardId] = true + shardPaths[shardId] = shardPath + inputFiles[shardId], err = os.OpenFile(shardPath, os.O_RDONLY, 0) + if err != nil { + return nil, err + } + defer inputFiles[shardId].Close() + presentCount++ } // Bitrot verify-and-exclude: when a generation-0 checksum sidecar is present @@ -190,6 +202,9 @@ func generateMissingEcFiles(baseFileName string, bufferSize int, largeBlockSize // silently consuming corrupt bytes. corruptOwned marks shards whose // (corrupt) original file must be replaced in place at its discovered path. corruptOwned := make([]bool, ctx.Total()) + for _, shardId := range zeroSized { + corruptOwned[shardId] = true + } prot, status := loadRebuildSidecar(baseFileName, ctx, additionalDirs) switch status { case BitrotInvalid: diff --git a/weed/storage/erasure_coding/ec_rebuild_safety_test.go b/weed/storage/erasure_coding/ec_rebuild_safety_test.go index 6184ec72b..99b912470 100644 --- a/weed/storage/erasure_coding/ec_rebuild_safety_test.go +++ b/weed/storage/erasure_coding/ec_rebuild_safety_test.go @@ -215,6 +215,51 @@ func TestRebuildEcFiles_HappyPathRebuildsByteIdentical(t *testing.T) { } } +// TestRebuildEcFiles_ZeroSizedShardRegeneratedInPlace: a zero-sized shard file +// (residue of a failed operation, issue 10730) must be excluded from the +// rebuild inputs and regenerated byte-identical over the empty file, instead +// of failing the whole rebuild with an input size mismatch. +func TestRebuildEcFiles_ZeroSizedShardRegeneratedInPlace(t *testing.T) { + dir := t.TempDir() + base := filepath.Join(dir, "vol") + ctx := NewDefaultECContext("", 0) + writeRandomDat(t, base, 7000) + + if _, err := generateEcFiles(base, 256*1024, ErasureCodingLargeBlockSize, ErasureCodingSmallBlockSize, ctx); err != nil { + t.Fatalf("generateEcFiles: %v", err) + } + + const emptied = 2 + want, err := os.ReadFile(base + ToExt(emptied)) + if err != nil { + t.Fatalf("read shard %d: %v", emptied, err) + } + if err := os.Truncate(base+ToExt(emptied), 0); err != nil { + t.Fatalf("truncate shard: %v", err) + } + + generated, err := RebuildEcFiles(base, ctx, true) + if err != nil { + t.Fatalf("RebuildEcFiles with a zero-sized shard: %v", err) + } + found := false + for _, id := range generated { + if id == emptied { + found = true + } + } + if !found { + t.Errorf("shard %d not reported as regenerated: %v", emptied, generated) + } + got, err := os.ReadFile(base + ToExt(emptied)) + if err != nil { + t.Fatalf("read regenerated shard: %v", err) + } + if string(got) != string(want) { + t.Errorf("regenerated shard %d differs from original (%d vs %d bytes)", emptied, len(got), len(want)) + } +} + // TestRebuildEcFiles_CustomRatioRebuildsByteIdentical: the same for a 9+3 ratio, // confirming no 10+4 assumption leaks into the destructive rebuild path. func TestRebuildEcFiles_CustomRatioRebuildsByteIdentical(t *testing.T) { diff --git a/weed/storage/needle/needle_parse_upload_alloc_test.go b/weed/storage/needle/needle_parse_upload_alloc_test.go index 3b65c289a..2d33ddc2d 100644 --- a/weed/storage/needle/needle_parse_upload_alloc_test.go +++ b/weed/storage/needle/needle_parse_upload_alloc_test.go @@ -27,11 +27,11 @@ func TestEagerPreGrow(t *testing.T) { const sizeLimit = int64(256 * 1024 * 1024) cases := []struct { - name string - startCap int - cl int64 - wantMinCap int - wantNoop bool // post: cap stays at startCap + name string + startCap int + cl int64 + wantMinCap int + wantNoop bool // post: cap stays at startCap }{ { name: "small content-length grows exactly", diff --git a/weed/storage/volume_io_error_test.go b/weed/storage/volume_io_error_test.go index 855dfdd9c..4d16af712 100644 --- a/weed/storage/volume_io_error_test.go +++ b/weed/storage/volume_io_error_test.go @@ -22,11 +22,11 @@ func (eioBackend) ReadAt(p []byte, off int64) (int, error) { return 0, syscall.E func (eioBackend) WriteAt(p []byte, off int64) (int, error) { return len(p), nil } -func (eioBackend) Truncate(int64) error { return nil } -func (eioBackend) Close() error { return nil } -func (eioBackend) GetStat() (int64, time.Time, error) { return 0, time.Time{}, nil } -func (eioBackend) Name() string { return "eio" } -func (eioBackend) Sync() error { return nil } +func (eioBackend) Truncate(int64) error { return nil } +func (eioBackend) Close() error { return nil } +func (eioBackend) GetStat() (int64, time.Time, error) { return 0, time.Time{}, nil } +func (eioBackend) Name() string { return "eio" } +func (eioBackend) Sync() error { return nil } func TestCheckReadWriteErrorTracksConsecutiveEIO(t *testing.T) { v := &Volume{}