fix(ec_distribute): remove partial files on copy stream error (#9543)

* fix(ec_distribute): remove partial files on copy stream error

writeToFile opens the destination with O_TRUNC and streams into it. On
a mid-stream receive / write / cancellation error it returned the
failure but left the destination behind in whatever state had been
written so far — typically 0 bytes when the source errored before
sending any FileContent. VolumeEcShardsCopy distributes .ecx by
calling doCopyFile, so this same stub-leaving behaviour produced the
0-byte .ecx files seen on EC encoding failures: the source claims a
non-zero ModifiedTsNs (so the existing "source not found" cleanup
doesn't fire), the stream then errors immediately, and the receiver
ends up with a 0-byte .ecx that downstream code mistook for a valid
empty index.

Clean up the partial file on every error path that returns from the
streaming loop (receive, write, and cancellation). Skip cleanup when
isAppend=true so resumable appends keep their existing content. As
defense in depth, VolumeEcShardsCopy also stats the .ecx after copy
and removes / errors on a 0-byte result so the orchestrator can pick
a different source.

The Rust volume server has only the source side of CopyFile (no
client-side stream-to-disk consumer) and no .ecx subsystem yet, so
this fix has no Rust mirror.

* fix(ec_distribute): close file before remove, fail fast on stat error

Address review feedback:

- writeToFile's mid-stream removeIncomplete called os.Remove while the
  destination file handle was still open. On Windows os.Remove fails
  while a handle is open, so the cleanup wouldn't run there. Wrap the
  handle close in a once-only helper, call it from removeIncomplete
  and from the existing "source not found" cleanup, and keep a deferred
  close as the safety net for the normal-return path.
- VolumeEcShardsCopy's post-copy .ecx check silently passed when
  os.Stat returned an error: doCopyFile had reported success but if
  the file was already gone, unreadable, or somehow a directory, the
  orchestrator only learned at mount time with no useful context.
  Treat any non-nil stat error and any directory result as a copy
  failure here and surface it immediately.

weed/server/volume_grpc_copy.go

@@ -342,7 +342,37 @@ func writeToFile(client volume_server_pb.VolumeServer_CopyFileClient, fileName s
 	if err != nil {
 		return modifiedTsNs, fmt.Errorf("open file %s: %w", fileName, err)
 	}
-	defer dst.Close()
+	// Track the destination handle through a closer that runs at most once.
+	// On Windows os.Remove fails while the file is still open, so any path
+	// that wants to delete the file we just created must close the handle
+	// first. The deferred call here is the safety net for normal returns.
+	dstClosed := false
+	closeDst := func() {
+		if dstClosed {
+			return
+		}
+		dstClosed = true
+		_ = dst.Close()
+	}
+	defer closeDst()
+
+	// removeIncomplete deletes the partially-written file we just opened
+	// with O_TRUNC. Used on stream / write / cancellation errors so a
+	// caller (notably VolumeEcShardsCopy distributing .ecx) doesn't end
+	// up with a 0-byte stub that downstream code mistakes for a valid
+	// empty file. Skip in isAppend mode — the existing content is not
+	// ours to remove, and resumable appends rely on partial state.
+	removeIncomplete := func(reason string) {
+		if isAppend {
+			return
+		}
+		closeDst()
+		if removeErr := os.Remove(fileName); removeErr != nil && !os.IsNotExist(removeErr) {
+			glog.Warningf("failed to remove incomplete file %s after %s: %v", fileName, reason, removeErr)
+		} else if removeErr == nil {
+			glog.V(1).Infof("removed incomplete file %s after %s", fileName, reason)
+		}
+	}
 
 	var progressedBytes int64
 	for {
@@ -354,14 +384,17 @@ func writeToFile(client volume_server_pb.VolumeServer_CopyFileClient, fileName s
 			modifiedTsNs = resp.ModifiedTsNs
 		}
 		if receiveErr != nil {
+			removeIncomplete("receive error")
 			return modifiedTsNs, fmt.Errorf("receiving %s: %w", fileName, receiveErr)
 		}
 		if _, writeErr := dst.Write(resp.FileContent); writeErr != nil {
+			removeIncomplete("write error")
 			return modifiedTsNs, fmt.Errorf("write file %s: %w", fileName, writeErr)
 		}
 		progressedBytes += int64(len(resp.FileContent))
 		if progressFn != nil {
 			if !progressFn(progressedBytes) {
+				removeIncomplete("progress cancelled")
 				return modifiedTsNs, fmt.Errorf("interrupted copy operation")
 			}
 		}
@@ -372,6 +405,7 @@ func writeToFile(client volume_server_pb.VolumeServer_CopyFileClient, fileName s
 	// Note: We check modifiedTsNs (not progressedBytes) because an empty source file
 	// is valid and should result in an empty destination file.
 	if modifiedTsNs == 0 && !isAppend {
+		closeDst()
 		if removeErr := os.Remove(fileName); removeErr != nil {
 			glog.V(1).Infof("failed to remove empty file %s: %v", fileName, removeErr)
 		} else {

weed/server/volume_grpc_copy_writefile_test.go

@@ -0,0 +1,145 @@
+package weed_server
+
+import (
+	"context"
+	"errors"
+	"io"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"google.golang.org/grpc/metadata"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+// fakeCopyFileStream is a synthetic VolumeServer_CopyFileClient used to
+// drive writeToFile's failure paths in tests. The pre-fix code left a
+// partial / 0-byte destination file on the disk when the stream errored
+// mid-copy; with the fix, writeToFile now removes the incomplete file
+// so callers (notably VolumeEcShardsCopy distributing .ecx) don't end
+// up with stubs that mount-time code mistakes for valid empty indexes.
+type fakeCopyFileStream struct {
+	responses []*volume_server_pb.CopyFileResponse
+	finalErr  error
+	index     int
+}
+
+func (s *fakeCopyFileStream) Recv() (*volume_server_pb.CopyFileResponse, error) {
+	if s.index >= len(s.responses) {
+		if s.finalErr != nil {
+			return nil, s.finalErr
+		}
+		return nil, io.EOF
+	}
+	r := s.responses[s.index]
+	s.index++
+	return r, nil
+}
+
+func (s *fakeCopyFileStream) Header() (metadata.MD, error) { return metadata.MD{}, nil }
+func (s *fakeCopyFileStream) Trailer() metadata.MD         { return metadata.MD{} }
+func (s *fakeCopyFileStream) CloseSend() error             { return nil }
+func (s *fakeCopyFileStream) Context() context.Context     { return context.Background() }
+func (s *fakeCopyFileStream) SendMsg(any) error            { return nil }
+func (s *fakeCopyFileStream) RecvMsg(any) error            { return nil }
+
+func TestWriteToFile_RemovesPartialFileOnStreamError(t *testing.T) {
+	dir := t.TempDir()
+	dst := filepath.Join(dir, "vol_42.ecx")
+
+	stream := &fakeCopyFileStream{
+		responses: []*volume_server_pb.CopyFileResponse{
+			// Real bytes flow first — modifiedTsNs is non-zero so the
+			// existing "source file not found" cleanup at the bottom of
+			// writeToFile does NOT fire; the new mid-stream cleanup is
+			// the only path that can remove the file.
+			{FileContent: []byte("partial data"), ModifiedTsNs: 1234567890},
+		},
+		finalErr: errors.New("simulated mid-stream failure"),
+	}
+
+	_, err := writeToFile(stream, dst, util.NewWriteThrottler(0), false, nil)
+	if err == nil {
+		t.Fatalf("writeToFile should propagate the stream error")
+	}
+
+	if _, statErr := os.Stat(dst); !os.IsNotExist(statErr) {
+		t.Errorf("incomplete file should be removed; stat err = %v", statErr)
+	}
+}
+
+func TestWriteToFile_RemovesEmptyFileOnImmediateStreamError(t *testing.T) {
+	dir := t.TempDir()
+	dst := filepath.Join(dir, "vol_42.ecx")
+
+	stream := &fakeCopyFileStream{
+		// No FileContent at all; stream errors on the first Recv.
+		// progressedBytes == 0 and modifiedTsNs == 0, so without the
+		// mid-stream cleanup this would leave a 0-byte file from the
+		// O_TRUNC at OpenFile time.
+		finalErr: errors.New("simulated immediate failure"),
+	}
+
+	_, err := writeToFile(stream, dst, util.NewWriteThrottler(0), false, nil)
+	if err == nil {
+		t.Fatalf("writeToFile should propagate the stream error")
+	}
+
+	if _, statErr := os.Stat(dst); !os.IsNotExist(statErr) {
+		t.Errorf("0-byte file should be removed; stat err = %v", statErr)
+	}
+}
+
+func TestWriteToFile_PreservesAppendModeOnError(t *testing.T) {
+	dir := t.TempDir()
+	dst := filepath.Join(dir, "vol_42.ecj")
+
+	// Pre-existing content the caller owns — isAppend=true tells
+	// writeToFile not to touch it on cleanup.
+	if err := os.WriteFile(dst, []byte("pre-existing journal data"), 0o644); err != nil {
+		t.Fatalf("seed file: %v", err)
+	}
+
+	stream := &fakeCopyFileStream{
+		finalErr: errors.New("simulated failure"),
+	}
+
+	_, err := writeToFile(stream, dst, util.NewWriteThrottler(0), true, nil)
+	if err == nil {
+		t.Fatalf("writeToFile should propagate the stream error")
+	}
+
+	info, statErr := os.Stat(dst)
+	if statErr != nil {
+		t.Fatalf("append-mode file should be preserved on error; stat: %v", statErr)
+	}
+	if info.Size() == 0 {
+		t.Errorf("append-mode file unexpectedly truncated to 0 bytes")
+	}
+}
+
+func TestWriteToFile_SucceedsOnCleanStream(t *testing.T) {
+	dir := t.TempDir()
+	dst := filepath.Join(dir, "vol_42.ecx")
+
+	want := []byte("hello ecx index")
+	stream := &fakeCopyFileStream{
+		responses: []*volume_server_pb.CopyFileResponse{
+			{FileContent: want, ModifiedTsNs: 9},
+		},
+	}
+
+	if _, err := writeToFile(stream, dst, util.NewWriteThrottler(0), false, nil); err != nil {
+		t.Fatalf("writeToFile failed on clean stream: %v", err)
+	}
+
+	got, err := os.ReadFile(dst)
+	if err != nil {
+		t.Fatalf("read back: %v", err)
+	}
+	if string(got) != string(want) {
+		t.Errorf("contents = %q, want %q", got, want)
+	}
+}

weed/server/volume_grpc_erasure_coding.go

@@ -289,6 +289,32 @@ func (vs *VolumeServer) VolumeEcShardsCopy(ctx context.Context, req *volume_serv
 			if _, err := vs.doCopyFile(client, true, req.Collection, req.VolumeId, math.MaxUint32, math.MaxInt64, indexBaseFileName, ".ecx", false, false, nil); err != nil {
 				return err
 			}
+			// Defense in depth: writeToFile now removes partial files on
+			// stream error, but a source that genuinely held a 0-byte
+			// .ecx (e.g. a corrupted upstream replica) would otherwise
+			// leave a 0-byte file here and the mount path would reject
+			// it later. Catch that at distribute time so the orchestrator
+			// can pick a different source rather than learning about it
+			// at mount.
+			// Stat failure must not silently pass. doCopyFile reported
+			// success, but if the file is gone, unreadable, or a directory
+			// somehow, the orchestrator should learn now — at mount time
+			// the operator only sees "no .ecx found" with no useful context
+			// about which step actually failed.
+			ecxPath := indexBaseFileName + ".ecx"
+			info, statErr := os.Stat(ecxPath)
+			if statErr != nil {
+				return fmt.Errorf("VolumeEcShardsCopy volume %d: stat copied .ecx %s: %w", req.VolumeId, ecxPath, statErr)
+			}
+			if info.IsDir() {
+				return fmt.Errorf("VolumeEcShardsCopy volume %d: copied .ecx path %s is a directory", req.VolumeId, ecxPath)
+			}
+			if info.Size() == 0 {
+				if removeErr := os.Remove(ecxPath); removeErr != nil && !os.IsNotExist(removeErr) {
+					glog.Warningf("VolumeEcShardsCopy volume %d: remove 0-byte .ecx %s: %v", req.VolumeId, ecxPath, removeErr)
+				}
+				return fmt.Errorf("VolumeEcShardsCopy volume %d: source .ecx is 0 bytes", req.VolumeId)
+			}
 		}
 
 		if req.CopyEcjFile {