diff --git a/weed/s3api/s3api_object_handlers_copy.go b/weed/s3api/s3api_object_handlers_copy.go index 89d22db766..fbbf2b2fd5 100644 --- a/weed/s3api/s3api_object_handlers_copy.go +++ b/weed/s3api/s3api_object_handlers_copy.go @@ -2,17 +2,21 @@ package s3api import ( "bytes" + "cmp" "context" + "crypto/md5" "crypto/rand" "encoding/base64" "encoding/xml" "errors" "fmt" + "hash" "io" "net/http" "net/url" "os" "reflect" + "slices" "strconv" "strings" "time" @@ -1146,7 +1150,9 @@ func (s3a *S3ApiServer) CopyObjectPartHandler(w http.ResponseWriter, r *http.Req return } - if uploadEntryHasSSE(uploadEntry) || sourceEntryIsEncrypted(entry) || uploadEntryHasChecksum(uploadEntry) { + // Inline-Content sources have no chunks for the raw-byte path to copy; + // the plaintext stream serves their range instead. + if uploadEntryHasSSE(uploadEntry) || sourceEntryIsEncrypted(entry) || uploadEntryHasChecksum(uploadEntry) || len(entry.Content) > 0 { etag, sseMetadata, errCode := s3a.copyObjectPartViaReencryption(r, entry, startOffset, endOffset, dstBucket, dstObject, uploadID, partID, uploadEntry) if errCode != s3err.ErrNone { s3err.WriteErrorResponse(w, r, errCode) @@ -1204,15 +1210,24 @@ func (s3a *S3ApiServer) CopyObjectPartHandler(w http.ResponseWriter, r *http.Req // For zero-size files or invalid ranges, create an empty part with size 0 dstEntry.Attributes.FileSize = 0 dstEntry.Chunks = nil + dstEntry.Attributes.Md5 = md5.New().Sum(nil) } else { // Copy chunks that overlap with the range - dstChunks, err := s3a.copyChunksForRange(entry, startOffset, endOffset, dstAssignPath) + dstChunks, partMd5, err := s3a.copyChunksForRange(entry, startOffset, endOffset, dstAssignPath) if err != nil { glog.Errorf("CopyObjectPartHandler copy chunks error: %v", err) s3err.WriteErrorResponse(w, r, s3a.mapCopyErrorToS3Error(err)) return } dstEntry.Chunks = dstChunks + dstEntry.Attributes.Md5 = partMd5 + } + + // Mirror putToFiler: the part entry carries the MD5 of its bytes so + // CopyPartResult, ListParts and complete's composite ETag all describe + // the copied range rather than the source chunks. + if len(dstEntry.Attributes.Md5) > 0 { + dstEntry.Extended[s3_constants.ExtETagKey] = []byte(fmt.Sprintf("%x", dstEntry.Attributes.Md5)) } // Save the part entry to the multipart uploads folder @@ -1479,7 +1494,7 @@ func (s3a *S3ApiServer) copySingleChunk(chunk *filer_pb.FileChunk, dstPath strin // the source volume for compressed bytes, so a gzipped chunk is // forwarded to the destination without anyone having to decompress. if canStreamCopyChunk(chunk) { - if err := s3a.streamCopyChunkRange(context.Background(), srcUrl, fileId, 0, int64(chunk.Size), true /*isFullChunk*/, assignResult); err != nil { + if err := s3a.streamCopyChunkRange(context.Background(), srcUrl, fileId, 0, int64(chunk.Size), true /*isFullChunk*/, assignResult, nil); err != nil { return nil, fmt.Errorf("stream chunk: %w", err) } return dstChunk, nil @@ -1534,9 +1549,22 @@ func (s3a *S3ApiServer) copySingleChunkForRange(originalChunk, rangeChunk *filer // part-size = chunk-size assemble pattern hits this branch. if canStreamCopyChunk(originalChunk) { isFullChunk := offsetInChunk == 0 && rangeChunk.Size == originalChunk.Size - if err := s3a.streamCopyChunkRange(context.Background(), srcUrl, fileId, offsetInChunk, int64(rangeChunk.Size), isFullChunk, assignResult); err != nil { + // A partial range stores different bytes than the source chunk, so + // its ETag has to be recomputed; a whole-chunk copy reuses it. + var digest hash.Hash + if !isFullChunk { + digest = md5.New() + } + if err := s3a.streamCopyChunkRange(context.Background(), srcUrl, fileId, offsetInChunk, int64(rangeChunk.Size), isFullChunk, assignResult, digest); err != nil { return nil, fmt.Errorf("stream chunk range: %w", err) } + if digest != nil { + dstChunk.ETag = base64.StdEncoding.EncodeToString(digest.Sum(nil)) + // The range fetch returns raw bytes and the destination stores + // them uncompressed, so the source chunk's gzip flag must not + // carry over onto a partial copy. + dstChunk.IsCompressed = false + } return dstChunk, nil } @@ -1609,8 +1637,16 @@ func parseRangeHeader(rangeHeader string, fileSize int64) (startOffset, endOffse return startOffset, endOffset, nil } -// copyChunksForRange copies chunks that overlap with the specified range -func (s3a *S3ApiServer) copyChunksForRange(entry *filer_pb.Entry, startOffset, endOffset int64, dstPath string) ([]*filer_pb.FileChunk, error) { +// partMd5Result carries the whole-part MD5 the copiedRangeMd5 goroutine +// computes beside the per-chunk copies. +type partMd5Result struct { + sum []byte + err error +} + +// copyChunksForRange copies chunks that overlap with the specified range and +// returns the MD5 of the copied bytes for the part entry's Attributes.Md5. +func (s3a *S3ApiServer) copyChunksForRange(entry *filer_pb.Entry, startOffset, endOffset int64, dstPath string) ([]*filer_pb.FileChunk, []byte, error) { var relevantChunks []*filer_pb.FileChunk var originalChunks []*filer_pb.FileChunk @@ -1641,6 +1677,36 @@ func (s3a *S3ApiServer) copyChunksForRange(entry *filer_pb.Entry, startOffset, e } } + // The part's Attributes.Md5 must cover the copied range in byte order, + // and md5 cannot combine per-chunk digests. A single chunk's own digest + // is the answer only when it supplies the whole range; anything else — + // several chunks, or gaps a sparse source zero-fills — needs an ordered + // pass the concurrent copies cannot feed, so a second stream of the + // source range hashes beside them. + partSize := endOffset - startOffset + 1 + singleChunkCovers := len(relevantChunks) == 1 && + relevantChunks[0].Offset == 0 && int64(relevantChunks[0].Size) == partSize + + md5Ctx, cancelMd5 := context.WithCancel(context.Background()) + defer cancelMd5() + + var partMd5 []byte + var partMd5Ch chan partMd5Result + switch { + case singleChunkCovers: + // resolved below from the copied chunk's own digest + case len(entry.GetChunks()) == 0: + // no chunks to read at all: the part is partSize zero bytes (the + // stream helper returns empty for a chunkless entry, not zero-fill) + partMd5 = md5Zeros(partSize) + default: + partMd5Ch = make(chan partMd5Result, 1) + go func() { + sum, err := s3a.copiedRangeMd5(md5Ctx, entry, startOffset, endOffset) + partMd5Ch <- partMd5Result{sum, err} + }() + } + // Copy the relevant chunks using a specialized method for range copies dstChunks := make([]*filer_pb.FileChunk, len(relevantChunks)) const defaultChunkCopyConcurrency = 4 @@ -1664,11 +1730,79 @@ func (s3a *S3ApiServer) copyChunksForRange(entry *filer_pb.Entry, startOffset, e // Wait for all operations to complete and check for errors for i := 0; i < len(relevantChunks); i++ { if err := <-errChan; err != nil { - return nil, err + return nil, nil, err } } - return dstChunks, nil + // Reads resolve overlapping chunks by ModifiedTsNs, and the concurrent + // copies stamped arbitrary finish times: rank the destination chunks by + // the source's timestamps so the part resolves overlaps the way the + // source did (and the way copiedRangeMd5 hashed it). + if len(dstChunks) > 1 { + order := make([]int, len(dstChunks)) + for i := range order { + order[i] = i + } + slices.SortStableFunc(order, func(a, b int) int { + return cmp.Compare(originalChunks[a].ModifiedTsNs, originalChunks[b].ModifiedTsNs) + }) + now := time.Now().UnixNano() + for rank, i := range order { + dstChunks[i].ModifiedTsNs = now + int64(rank) + } + } + + if partMd5Ch != nil { + if res := <-partMd5Ch; res.err != nil { + return nil, nil, fmt.Errorf("md5 of copied range: %w", res.err) + } else { + partMd5 = res.sum + } + } else if singleChunkCovers { + if partMd5 = util.Base64Md5ToBytes(dstChunks[0].ETag); len(partMd5) != md5.Size { + // TUS and mount writes leave chunks without an ETag; hash the + // copied range when the inherited digest is missing or malformed. + sum, err := s3a.copiedRangeMd5(md5Ctx, entry, startOffset, endOffset) + if err != nil { + return nil, nil, fmt.Errorf("md5 of copied range: %w", err) + } + partMd5 = sum + } + } + + return dstChunks, partMd5, nil +} + +// md5Zeros hashes size zero bytes without a volume read — a copied range +// that overlaps no source chunk assembles to all zeros. +func md5Zeros(size int64) []byte { + h := md5.New() + var zeros [1 << 16]byte + for size > 0 { + n := min(size, int64(len(zeros))) + h.Write(zeros[:n]) + size -= n + } + return h.Sum(nil) +} + +// copiedRangeMd5 streams the source range once and hashes its plaintext — +// exactly the bytes the non-SSE part copy assembles on the destination. +func (s3a *S3ApiServer) copiedRangeMd5(ctx context.Context, entry *filer_pb.Entry, startOffset, endOffset int64) ([]byte, error) { + size := endOffset - startOffset + 1 + // The copy walks the chunks, so the hash must too: on an entry that also + // carries inline Content the stream helper would prefer Content instead. + chunked := &filer_pb.Entry{Chunks: entry.GetChunks()} + reader, err := s3a.getEncryptedStreamFromVolumes(ctx, chunked, startOffset, size) + if err != nil { + return nil, err + } + defer reader.Close() + h := md5.New() + if _, err := io.Copy(h, reader); err != nil { + return nil, err + } + return h.Sum(nil), nil } // Helper methods for copy operations to avoid code duplication diff --git a/weed/s3api/s3api_object_handlers_copy_bench_test.go b/weed/s3api/s3api_object_handlers_copy_bench_test.go index 31fb773839..f2011b9fe8 100644 --- a/weed/s3api/s3api_object_handlers_copy_bench_test.go +++ b/weed/s3api/s3api_object_handlers_copy_bench_test.go @@ -160,7 +160,7 @@ func BenchmarkCopyChunk_Streamed(b *testing.B) { for i := 0; i < b.N; i++ { if err := s3a.streamCopyChunkRange(context.Background(), env.srcSrv.URL, env.assign.FileId, 0, int64(size), - true /*isFullChunk*/, env.assign); err != nil { + true /*isFullChunk*/, env.assign, nil); err != nil { b.Fatalf("stream: %v", err) } } diff --git a/weed/s3api/s3api_object_handlers_copy_etag_test.go b/weed/s3api/s3api_object_handlers_copy_etag_test.go index 8aeda9ccd4..9860a03971 100644 --- a/weed/s3api/s3api_object_handlers_copy_etag_test.go +++ b/weed/s3api/s3api_object_handlers_copy_etag_test.go @@ -1,6 +1,11 @@ package s3api import ( + "bytes" + "crypto/md5" + "encoding/base64" + "fmt" + "net/http" "net/http/httptest" "strings" "testing" @@ -9,6 +14,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" "github.com/seaweedfs/seaweedfs/weed/s3api/s3err" + util_http "github.com/seaweedfs/seaweedfs/weed/util/http" ) func TestCopyEntryETagPrefersStoredExtendedETag(t *testing.T) { @@ -145,3 +151,292 @@ func TestParseRangeHeaderRejectsUnsatisfiableRange(t *testing.T) { }) } } + +// A part copied with x-amz-copy-source-range must report the MD5 of the copied +// bytes: per destination chunk for a partial-chunk copy, and the whole-part MD5 +// the part entry's Attributes.Md5 carries for CopyPartResult / ListParts / +// complete's composite ETag. +func TestCopyChunksForRangeCopiedETags(t *testing.T) { + volume := startFakeVolumeServer(t) + filerImpl := &ambiguousPutFiler{volume: volume, entries: map[string]*filer_pb.Entry{}} + s3a := newPutTestServer(t, startFakeFiler(t, filerImpl)) + util_http.InitGlobalHttpClient() + + const chunkSize = 4 << 20 + payload := make([]byte, 2*chunkSize) + for i := range payload { + payload[i] = byte(i*31 + 7) + } + b64md5 := func(data []byte) string { + sum := md5.Sum(data) + return base64.StdEncoding.EncodeToString(sum[:]) + } + fid0 := fmt.Sprintf("3,%016x%08x", 101, 202) + fid1 := fmt.Sprintf("3,%016x%08x", 103, 204) + volume.stored[fid0] = payload[:chunkSize] + volume.stored[fid1] = payload[chunkSize:] + entry := &filer_pb.Entry{ + Name: "src", + Attributes: &filer_pb.FuseAttributes{ + FileSize: uint64(len(payload)), + }, + Chunks: []*filer_pb.FileChunk{ + {FileId: fid0, Offset: 0, Size: chunkSize, ETag: b64md5(payload[:chunkSize])}, + {FileId: fid1, Offset: chunkSize, Size: chunkSize, ETag: b64md5(payload[chunkSize:])}, + }, + } + + // A range spanning the first chunk fully plus 1 MiB of the second. + multiEnd := int64(chunkSize + 1<<20 - 1) + dstChunks, partMd5, err := s3a.copyChunksForRange(entry, 0, multiEnd, "/dst") + if err != nil { + t.Fatalf("copyChunksForRange: %v", err) + } + if len(dstChunks) != 2 { + t.Fatalf("copyChunksForRange returned %d chunks, want 2", len(dstChunks)) + } + // A whole-chunk copy forwards the stored bytes, so the source ETag stays valid. + if got, want := dstChunks[0].ETag, entry.Chunks[0].ETag; got != want { + t.Errorf("whole-chunk copy ETag = %q, want source chunk ETag %q", got, want) + } + // A partial copy stores only the range bytes; its ETag must be their MD5. + if got, want := dstChunks[1].ETag, b64md5(payload[chunkSize:multiEnd+1]); got != want { + t.Errorf("partial-chunk copy ETag = %q, want copied-bytes MD5 %q", got, want) + } + // The copied bytes landed on the destination volume (inside the multipart framing). + if stored := volume.stored[dstChunks[1].GetFileIdString()]; !bytes.Contains(stored, payload[chunkSize:multiEnd+1]) { + t.Error("destination volume does not hold the copied range bytes") + } + // The whole-part MD5 hashes every copied byte once, in range order. + wantPartMd5 := md5.Sum(payload[:multiEnd+1]) + if !bytes.Equal(partMd5, wantPartMd5[:]) { + t.Errorf("part Md5 = %x, want %x", partMd5, wantPartMd5) + } + partEntry := &filer_pb.Entry{ + Attributes: &filer_pb.FuseAttributes{Md5: partMd5}, + Chunks: dstChunks, + } + if got, want := copyEntryETag(partEntry), fmt.Sprintf("%x", wantPartMd5); got != want { + t.Errorf("copyEntryETag = %q, want %q", got, want) + } + + // A range inside one chunk: the part's ETag is the MD5 of those bytes. + innerChunks, innerMd5, err := s3a.copyChunksForRange(entry, 100, 199, "/dst") + if err != nil { + t.Fatalf("copyChunksForRange inner range: %v", err) + } + if len(innerChunks) != 1 { + t.Fatalf("inner range returned %d chunks, want 1", len(innerChunks)) + } + if got, want := innerChunks[0].ETag, b64md5(payload[100:200]); got != want { + t.Errorf("inner-range chunk ETag = %q, want copied-bytes MD5 %q", got, want) + } + wantInnerMd5 := md5.Sum(payload[100:200]) + if !bytes.Equal(innerMd5, wantInnerMd5[:]) { + t.Errorf("inner-range part Md5 = %x, want %x", innerMd5, wantInnerMd5) + } +} + +// A range over a sparse source zero-fills the gaps, so the part MD5 has to +// hash the logical range — not just the one chunk that overlaps it. +func TestCopyChunksForRangeSparseETags(t *testing.T) { + volume := startFakeVolumeServer(t) + filerImpl := &ambiguousPutFiler{volume: volume, entries: map[string]*filer_pb.Entry{}} + s3a := newPutTestServer(t, startFakeFiler(t, filerImpl)) + util_http.InitGlobalHttpClient() + + fid := fmt.Sprintf("3,%016x%08x", 105, 206) + volume.stored[fid] = []byte("abc") + sum := md5.Sum([]byte("abc")) + entry := &filer_pb.Entry{ + Name: "sparse", + Attributes: &filer_pb.FuseAttributes{ + FileSize: 6, + }, + Chunks: []*filer_pb.FileChunk{ + {FileId: fid, Offset: 3, Size: 3, ETag: base64.StdEncoding.EncodeToString(sum[:])}, + }, + } + + // bytes 0-5 read as "\x00\x00\x00abc"; the part MD5 must cover the zeros too. + dstChunks, partMd5, err := s3a.copyChunksForRange(entry, 0, 5, "/dst") + if err != nil { + t.Fatalf("copyChunksForRange sparse: %v", err) + } + if len(dstChunks) != 1 || dstChunks[0].Offset != 3 { + t.Fatalf("sparse copy chunks = %+v, want one chunk at offset 3", dstChunks) + } + wantMd5 := md5.Sum([]byte("\x00\x00\x00abc")) + if !bytes.Equal(partMd5, wantMd5[:]) { + t.Errorf("sparse part Md5 = %x, want %x (zero-filled range)", partMd5, wantMd5) + } + + // A range over a pure hole copies nothing but still reads as zeros. + holeChunks, holeMd5, err := s3a.copyChunksForRange(entry, 0, 2, "/dst") + if err != nil { + t.Fatalf("copyChunksForRange hole: %v", err) + } + if len(holeChunks) != 0 { + t.Fatalf("hole copy returned %d chunks, want 0", len(holeChunks)) + } + wantHoleMd5 := md5.Sum([]byte{0, 0, 0}) + if !bytes.Equal(holeMd5, wantHoleMd5[:]) { + t.Errorf("hole part Md5 = %x, want %x (all zeros)", holeMd5, wantHoleMd5) + } +} + +// Destination chunks must resolve overlaps the way the source did: reads +// pick winners by ModifiedTsNs, so the copied chunks keep the source's +// precedence even though they were uploaded concurrently. +func TestCopyChunksForRangePreservesSourcePrecedence(t *testing.T) { + volume := startFakeVolumeServer(t) + filerImpl := &ambiguousPutFiler{volume: volume, entries: map[string]*filer_pb.Entry{}} + s3a := newPutTestServer(t, startFakeFiler(t, filerImpl)) + util_http.InitGlobalHttpClient() + + etagOf := func(b []byte) string { + sum := md5.Sum(b) + return base64.StdEncoding.EncodeToString(sum[:]) + } + fidA := fmt.Sprintf("3,%016x%08x", 107, 208) + fidB := fmt.Sprintf("3,%016x%08x", 109, 210) + volume.stored[fidA] = []byte("aaaa") + volume.stored[fidB] = []byte("bbbb") + + // A covers [0,4), B covers [2,6); B's range is copied into a chunk that + // overlaps A's copy on [2,4), and the source says A wins there. + entry := &filer_pb.Entry{ + Name: "overlap", + Attributes: &filer_pb.FuseAttributes{FileSize: 6}, + Chunks: []*filer_pb.FileChunk{ + {FileId: fidA, Offset: 0, Size: 4, ModifiedTsNs: 20, ETag: etagOf([]byte("aaaa"))}, + {FileId: fidB, Offset: 2, Size: 4, ModifiedTsNs: 10, ETag: etagOf([]byte("bbbb"))}, + }, + } + dstChunks, partMd5, err := s3a.copyChunksForRange(entry, 0, 5, "/dst") + if err != nil { + t.Fatalf("copyChunksForRange overlap: %v", err) + } + if len(dstChunks) != 2 { + t.Fatalf("overlap copy returned %d chunks, want 2", len(dstChunks)) + } + if dstChunks[0].ModifiedTsNs <= dstChunks[1].ModifiedTsNs { + t.Errorf("destination precedence inverted: chunk from ts=20 got %d, chunk from ts=10 got %d", + dstChunks[0].ModifiedTsNs, dstChunks[1].ModifiedTsNs) + } + // The source resolves to "aaaa"+"bb"; the part MD5 must match that view. + wantMd5 := md5.Sum([]byte("aaaabb")) + if !bytes.Equal(partMd5, wantMd5[:]) { + t.Errorf("overlap part Md5 = %x, want %x", partMd5, wantMd5) + } +} + +// An inline-Content source has no chunks for the raw-byte copy; the plaintext +// stream path must write the range's real bytes and tag the part with their +// MD5, not a zero-fill digest. +func TestCopyObjectPartInlineContent(t *testing.T) { + volume := startFakeVolumeServer(t) + filerImpl := &ambiguousPutFiler{volume: volume, apply: true, entries: map[string]*filer_pb.Entry{}} + s3a := newPutTestServer(t, startFakeFiler(t, filerImpl)) + util_http.InitGlobalHttpClient() + + content := []byte("inline-content") + entry := &filer_pb.Entry{ + Name: "src", + Attributes: &filer_pb.FuseAttributes{FileSize: uint64(len(content))}, + Content: content, + } + r := httptest.NewRequest(http.MethodPut, "/b/dst?partNumber=1&uploadId=u1", nil) + etag, _, code := s3a.copyObjectPartViaReencryption(r, entry, 2, 5, "b", "dst", "u1", 1, &filer_pb.Entry{}) + if code != s3err.ErrNone { + t.Fatalf("copyObjectPartViaReencryption: %v", code) + } + wantMd5 := md5.Sum(content[2:6]) + if got := strings.Trim(etag, `"`); got != fmt.Sprintf("%x", wantMd5) { + t.Errorf("etag = %q, want %x", got, wantMd5) + } + var partEntry *filer_pb.Entry + for key, e := range filerImpl.entries { + if strings.Contains(key, "u1") { + partEntry = e + } + } + if partEntry == nil || len(partEntry.Chunks) == 0 { + t.Fatalf("no part entry with chunks stored: %+v", filerImpl.entries) + } + found := false + for _, stored := range volume.stored { + if bytes.Contains(stored, content[2:6]) { + found = true + } + } + if !found { + t.Errorf("destination volume does not hold the copied content bytes") + } +} + +// An empty part must carry the empty-body MD5 on the stored entry too, so +// ListParts and complete's composite ETag see the same digest CopyPartResult +// reported. +func TestWriteEmptyCopyPartStoresMd5(t *testing.T) { + volume := startFakeVolumeServer(t) + filerImpl := &ambiguousPutFiler{volume: volume, apply: true, entries: map[string]*filer_pb.Entry{}} + s3a := newPutTestServer(t, startFakeFiler(t, filerImpl)) + util_http.InitGlobalHttpClient() + + etag, code := s3a.writeEmptyCopyPart("b", "u1", 3) + if code != s3err.ErrNone { + t.Fatalf("writeEmptyCopyPart: %v", code) + } + const emptyMD5Hex = "d41d8cd98f00b204e9800998ecf8427e" + if etag != emptyMD5Hex { + t.Fatalf("writeEmptyCopyPart etag = %q, want %q", etag, emptyMD5Hex) + } + var partEntry *filer_pb.Entry + for key, e := range filerImpl.entries { + if strings.Contains(key, "u1") { + partEntry = e + } + } + if partEntry == nil { + t.Fatalf("no part entry stored: %+v", filerImpl.entries) + } + if got := string(partEntry.Extended[s3_constants.ExtETagKey]); got != emptyMD5Hex { + t.Errorf("stored ExtETagKey = %q, want %q", got, emptyMD5Hex) + } + if got := fmt.Sprintf("%x", partEntry.Attributes.Md5); got != emptyMD5Hex { + t.Errorf("stored Attributes.Md5 = %q, want %q", got, emptyMD5Hex) + } +} + +// A whole-range chunk copy normally reuses the source chunk's ETag as the +// part MD5; TUS and mount writes leave chunks without one, so the part has +// to hash the copied bytes instead of saving an empty digest. +func TestCopyChunksForRangeMissingChunkETag(t *testing.T) { + volume := startFakeVolumeServer(t) + filerImpl := &ambiguousPutFiler{volume: volume, entries: map[string]*filer_pb.Entry{}} + s3a := newPutTestServer(t, startFakeFiler(t, filerImpl)) + util_http.InitGlobalHttpClient() + + fid := fmt.Sprintf("3,%016x%08x", 111, 212) + volume.stored[fid] = []byte("etagless") + entry := &filer_pb.Entry{ + Name: "src", + Attributes: &filer_pb.FuseAttributes{FileSize: 8}, + Chunks: []*filer_pb.FileChunk{ + {FileId: fid, Offset: 0, Size: 8}, + }, + } + + dstChunks, partMd5, err := s3a.copyChunksForRange(entry, 0, 7, "/dst") + if err != nil { + t.Fatalf("copyChunksForRange: %v", err) + } + if len(dstChunks) != 1 { + t.Fatalf("copyChunksForRange returned %d chunks, want 1", len(dstChunks)) + } + wantMd5 := md5.Sum([]byte("etagless")) + if !bytes.Equal(partMd5, wantMd5[:]) { + t.Errorf("part Md5 = %x, want %x (hashed copied range)", partMd5, wantMd5) + } +} diff --git a/weed/s3api/s3api_object_handlers_copy_part_sse.go b/weed/s3api/s3api_object_handlers_copy_part_sse.go index 7a4968fa9f..f8baa8782d 100644 --- a/weed/s3api/s3api_object_handlers_copy_part_sse.go +++ b/weed/s3api/s3api_object_handlers_copy_part_sse.go @@ -2,6 +2,7 @@ package s3api import ( "context" + "crypto/md5" "encoding/base64" "encoding/json" "errors" @@ -419,14 +420,19 @@ func (s3a *S3ApiServer) writeEmptyCopyPart(dstBucket, uploadID string, partID in return "", s3err.ErrInternalError } } + const emptyMD5Hex = "d41d8cd98f00b204e9800998ecf8427e" if err := s3a.mkFile(uploadDir, partName, nil, func(e *filer_pb.Entry) { if e.Attributes == nil { e.Attributes = &filer_pb.FuseAttributes{} } e.Attributes.FileSize = 0 + e.Attributes.Md5 = md5.New().Sum(nil) + if e.Extended == nil { + e.Extended = map[string][]byte{} + } + e.Extended[s3_constants.ExtETagKey] = []byte(emptyMD5Hex) }); err != nil { return "", s3err.ErrInternalError } - const emptyMD5Hex = "d41d8cd98f00b204e9800998ecf8427e" return emptyMD5Hex, s3err.ErrNone } diff --git a/weed/s3api/s3api_object_handlers_copy_stream.go b/weed/s3api/s3api_object_handlers_copy_stream.go index acba56caff..910ec3aca5 100644 --- a/weed/s3api/s3api_object_handlers_copy_stream.go +++ b/weed/s3api/s3api_object_handlers_copy_stream.go @@ -73,12 +73,19 @@ func canStreamCopyChunk(chunk *filer_pb.FileChunk) bool { // Trusting a hint that disagrees with the wire bytes (e.g. caller says // "compressed" but the volume returned raw bytes for a partial range) // would label raw bytes as gzip and corrupt reads on the destination. +// +// digestWriter, when non-nil, receives the plaintext bytes as they are +// forwarded so the caller can derive the destination chunk's ETag. It is +// not fed when the wire bytes are gzip — only possible on the full-chunk +// pass-through, whose logical content still matches the source chunk's +// own ETag and needs no recomputation. func (s3a *S3ApiServer) streamCopyChunkRange( ctx context.Context, srcUrl, srcFileId string, offset, size int64, isFullChunk bool, assignResult *filer_pb.AssignVolumeResponse, + digestWriter io.Writer, ) error { // Range header takes int64, but mirror the int32 bound from // downloadChunkData so the two paths reject the same inputs. @@ -166,7 +173,11 @@ func (s3a *S3ApiServer) streamCopyChunkRange( return } - if _, err := io.Copy(fw, srcResp.Body); err != nil { + src := io.Reader(srcResp.Body) + if digestWriter != nil && !bodyIsGzipped { + src = io.TeeReader(srcResp.Body, digestWriter) + } + if _, err := io.Copy(fw, src); err != nil { if shouldLogStreamError(err) { glog.V(2).Infof("stream copy %s offset=%d size=%d: %v", srcUrl, offset, size, err) diff --git a/weed/s3api/s3api_object_handlers_readonly_test.go b/weed/s3api/s3api_object_handlers_readonly_test.go index 3fd4c6f91c..09074644f7 100644 --- a/weed/s3api/s3api_object_handlers_readonly_test.go +++ b/weed/s3api/s3api_object_handlers_readonly_test.go @@ -87,7 +87,7 @@ func TestCopyChunksReadOnlyIsAccessDenied(t *testing.T) { t.Fatalf("copyChunks err %v maps to %v, want %v", err, code, s3err.ErrAccessDenied) } - _, err = s3a.copyChunksForRange(entry, 0, 7, "/buckets/b/dst") + _, _, err = s3a.copyChunksForRange(entry, 0, 7, "/buckets/b/dst") if code := s3a.mapCopyErrorToS3Error(err); code != s3err.ErrAccessDenied { t.Fatalf("copyChunksForRange err %v maps to %v, want %v", err, code, s3err.ErrAccessDenied) }