diff --git a/weed/server/volume_grpc_copy.go b/weed/server/volume_grpc_copy.go index f8cd66e24..a8cc155a1 100644 --- a/weed/server/volume_grpc_copy.go +++ b/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 { diff --git a/weed/server/volume_grpc_copy_writefile_test.go b/weed/server/volume_grpc_copy_writefile_test.go new file mode 100644 index 000000000..3dcb03011 --- /dev/null +++ b/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) + } +} diff --git a/weed/server/volume_grpc_erasure_coding.go b/weed/server/volume_grpc_erasure_coding.go index 4186fb5e1..3134d9d0d 100644 --- a/weed/server/volume_grpc_erasure_coding.go +++ b/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 {