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

@@ -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
 	})

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.

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 {

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)
+	}
+}

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) {

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:

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) {

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",

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{}