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 <noreply@anthropic.com>

* filer.copy: test chunk order and chunk md5 on chunked uploads

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Chris Lu <chrislusf@users.noreply.github.com>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
authored and GitHub committed 2026-11-11 08:43:23 -08:00
1 parent b9866bae4f
commit 1202c5ba08
2 files changed
+134

No files matched your search

+9
View File
@@ -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: "",
+125
View File
@@ -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 := &copyTestFiler{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)
}
}
}