From 1202c5ba08a70d115337b447c25eaf90ff67e670 Mon Sep 17 00:00:00 2001 From: Jay Hemnani <193022578+jayhemnani9910@users.noreply.github.com> Date: Wed, 11 Nov 2026 22:13:23 +0530 Subject: [PATCH] filer.copy: compute chunk md5 so files get a real ETag (#12497) * filer.copy: compute chunk md5 so files get a real ETag Chunks uploaded by filer.copy had no ETag, so every multi-chunk file got the md5 of the empty string plus -N. An overwritten file kept its old ETag and a revalidating client got 304 for changed content. * filer.copy: store chunks in offset order Chunks are collected in upload completion order, and the entry ETag hashes the chunk ETags in stored order. Identical files copied with concurrent chunk uploads got different ETags, and a file with swapped equal-sized chunks could keep the old one. Co-Authored-By: Claude Opus 5.5 * filer.copy: test chunk order and chunk md5 on chunked uploads Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Chris Lu Co-authored-by: Claude Opus 5.5 --- weed/command/filer_copy.go | 9 +++ weed/command/filer_copy_test.go | 125 ++++++++++++++++++++++++++++++++ 2 files changed, 134 insertions(+) create mode 100644 weed/command/filer_copy_test.go diff --git a/weed/command/filer_copy.go b/weed/command/filer_copy.go index 5525ebddd4..748a3fcc36 100644 --- a/weed/command/filer_copy.go +++ b/weed/command/filer_copy.go @@ -1,12 +1,14 @@ package command import ( + "cmp" "context" "fmt" "io" "net/http" "os" "path/filepath" + "slices" "strconv" "strings" "sync" @@ -366,6 +368,7 @@ func (worker *FileCopyWorker) uploadFileAsOne(task FileCopyTask, f *os.File) err }, &operation.UploadOption{ Filename: fileName, + WantMd5: true, Cipher: worker.options.cipher, IsInputCompressed: false, MimeType: mimeType, @@ -448,6 +451,7 @@ func (worker *FileCopyWorker) uploadFileInChunks(task FileCopyTask, f *os.File, }, &operation.UploadOption{ Filename: fileName + "-" + strconv.FormatInt(i+1, 10), + WantMd5: true, Cipher: worker.options.cipher, IsInputCompressed: false, MimeType: "", @@ -476,6 +480,10 @@ func (worker *FileCopyWorker) uploadFileInChunks(task FileCopyTask, f *os.File, for chunk := range chunksChan { chunks = append(chunks, chunk) } + // Chunks arrive in completion order, and the ETag hashes them in stored order. + slices.SortFunc(chunks, func(a, b *filer_pb.FileChunk) int { + return cmp.Compare(a.Offset, b.Offset) + }) if uploadError != nil { var fileIds []string @@ -580,6 +588,7 @@ func (worker *FileCopyWorker) saveDataAsChunk(reader io.Reader, name string, off }, &operation.UploadOption{ Filename: name, + WantMd5: true, Cipher: worker.options.cipher, IsInputCompressed: false, MimeType: "", diff --git a/weed/command/filer_copy_test.go b/weed/command/filer_copy_test.go new file mode 100644 index 0000000000..01e6e16547 --- /dev/null +++ b/weed/command/filer_copy_test.go @@ -0,0 +1,125 @@ +package command + +import ( + "context" + "crypto/md5" + "crypto/rand" + "encoding/base64" + "encoding/json" + "fmt" + "io" + "net" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + + "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" +) + +type copyTestFiler struct { + filer_pb.UnimplementedSeaweedFilerServer + volumeHost string + nextKey atomic.Int32 + mu sync.Mutex + created *filer_pb.Entry +} + +func (f *copyTestFiler) AssignVolume(ctx context.Context, req *filer_pb.AssignVolumeRequest) (*filer_pb.AssignVolumeResponse, error) { + return &filer_pb.AssignVolumeResponse{ + FileId: fmt.Sprintf("1,%x637037d6", f.nextKey.Add(1)), + Location: &filer_pb.Location{Url: f.volumeHost, PublicUrl: f.volumeHost}, + }, nil +} + +func (f *copyTestFiler) CreateEntry(ctx context.Context, req *filer_pb.CreateEntryRequest) (*filer_pb.CreateEntryResponse, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.created = req.Entry + return &filer_pb.CreateEntryResponse{}, nil +} + +func TestUploadFileInChunksStoresChunksInOffsetOrder(t *testing.T) { + // The first chunk finishes last, and the server echoes Content-MD5 back + // the way a volume server does. + volume := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + file, header, err := r.FormFile("file") + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + data, _ := io.ReadAll(file) + if strings.HasSuffix(header.Filename, "-1") { + time.Sleep(200 * time.Millisecond) + } + w.Header().Set("Content-MD5", header.Header.Get("Content-MD5")) + w.WriteHeader(http.StatusCreated) + json.NewEncoder(w).Encode(map[string]any{"name": header.Filename, "size": len(data)}) + })) + defer volume.Close() + + fake := ©TestFiler{volumeHost: strings.TrimPrefix(volume.URL, "http://")} + lis, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + srv := grpc.NewServer() + filer_pb.RegisterSeaweedFilerServer(srv, fake) + go srv.Serve(lis) + defer srv.Stop() + + const chunkSize, chunkCount = 64 * 1024, 3 + data := make([]byte, chunkSize*chunkCount) + rand.Read(data) + path := filepath.Join(t.TempDir(), "f.bin") + if err := os.WriteFile(path, data, 0o644); err != nil { + t.Fatal(err) + } + f, err := os.Open(path) + if err != nil { + t.Fatal(err) + } + defer f.Close() + + empty, access, concurrent := "", "url", chunkCount + worker := &FileCopyWorker{ + options: &CopyOptions{ + replication: &empty, + collection: &empty, + diskType: &empty, + concurrentChunks: &concurrent, + grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + volumeServerAccess: &access, + }, + filerAddress: pb.ServerAddress(fmt.Sprintf("127.0.0.1:1.%d", lis.Addr().(*net.TCPAddr).Port)), + } + task := FileCopyTask{destinationUrlPath: "/dir/", fileSize: int64(len(data)), fileMode: 0o644} + if err := worker.uploadFileInChunks(task, f, chunkCount, chunkSize); err != nil { + t.Fatal(err) + } + + fake.mu.Lock() + chunks := fake.created.GetChunks() + fake.mu.Unlock() + if len(chunks) != chunkCount { + t.Fatalf("stored %d chunks, want %d", len(chunks), chunkCount) + } + for i, c := range chunks { + if want := int64(i * chunkSize); c.Offset != want { + t.Fatalf("chunk %d is at offset %d, want %d", i, c.Offset, want) + } + sum := md5.Sum(data[i*chunkSize : (i+1)*chunkSize]) + if want := base64.StdEncoding.EncodeToString(sum[:]); c.ETag != want { + t.Fatalf("chunk %d ETag = %q, want the md5 of its data %q", i, c.ETag, want) + } + } +}