diff --git a/test/s3/remote_cache/README.md b/test/s3/remote_cache/README.md index fde8d3866..516320a2c 100644 --- a/test/s3/remote_cache/README.md +++ b/test/s3/remote_cache/README.md @@ -43,6 +43,7 @@ This tests the full remote caching workflow including singleflight deduplication | Test File | Commands Tested | Test Count | Description | |-----------|----------------|------------|-------------| | `remote_cache_test.go` | Basic caching | 5 tests | Original caching workflow and singleflight tests | +| `remote_cache_copy_test.go` | S3 CopyObject / UploadPartCopy from a remote-only source | 2 tests | Source object lives only in remote storage; CopyObject and UploadPartCopy must cache it locally before persisting the destination so the result is readable | | `command_remote_configure_test.go` | `remote.configure` | 6 tests | Configuration management | | `command_remote_mount_test.go` | `remote.mount`, `remote.unmount`, `remote.mount.buckets` | 10 tests | Mount operations | | `command_remote_cache_test.go` | `remote.cache`, `remote.uncache` | 13 tests | Cache/uncache with filters | @@ -50,7 +51,7 @@ This tests the full remote caching workflow including singleflight deduplication | `command_remote_meta_sync_test.go` | `remote.meta.sync` | 8 tests | Metadata synchronization | | `command_edge_cases_test.go` | All commands | 11 tests | Edge cases and stress tests | -**Total: 65 test cases covering 8 weed shell commands** +**Total: 67 test cases covering 8 weed shell commands and the S3 copy paths for remote-only sources** ### Commands Tested diff --git a/test/s3/remote_cache/remote_cache_copy_test.go b/test/s3/remote_cache/remote_cache_copy_test.go new file mode 100644 index 000000000..efef2d7f3 --- /dev/null +++ b/test/s3/remote_cache/remote_cache_copy_test.go @@ -0,0 +1,174 @@ +package remote_cache + +import ( + "bytes" + "crypto/md5" + "fmt" + "io" + "testing" + "time" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/s3" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestRemoteCacheCopyObject exercises the bug pattern that #9304 originally +// reported and that #7817's tests couldn't catch: CopyObject reads a source +// whose data lives only in remote storage (not yet cached locally), so its +// resolved entry has FileSize > 0 but no chunks. Pre-fix, the handler wrote a +// destination with that same shape, and any GET on the destination returned +// 500 "data integrity error: size N reported but no content". +func TestRemoteCacheCopyObject(t *testing.T) { + checkServersRunning(t) + + srcKey := fmt.Sprintf("copy-src-%d.bin", time.Now().UnixNano()) + dstKey := fmt.Sprintf("copy-dst-%d.bin", time.Now().UnixNano()) + + // 5 MiB so the source has multiple chunks and is large enough to exercise + // the streaming copy path, not just an inline-content shortcut. + srcData := make([]byte, 5*1024*1024) + for i := range srcData { + srcData[i] = byte(i % 256) + } + srcSum := md5.Sum(srcData) + + t.Log("Uploading source object to primary (local)") + uploadToPrimary(t, srcKey, srcData) + + t.Log("Uncaching source: pushes data to remote and removes local chunks") + uncacheLocal(t, srcKey) + + // At this point the source entry on the primary has FileSize > 0, + // no local chunks, and a RemoteEntry pointing at the secondary. + // CopyObject must cache the source before persisting the destination. + t.Log("Issuing CopyObject from remote-only source to a new local destination") + _, err := getPrimaryClient().CopyObject(&s3.CopyObjectInput{ + Bucket: aws.String(testBucket), + Key: aws.String(dstKey), + CopySource: aws.String(testBucket + "/" + srcKey), + }) + require.NoError(t, err, "CopyObject from remote-only source must succeed") + + // HEAD before GET so an obvious size mismatch surfaces with a useful error + // instead of being masked by a body-stream failure. + head, err := getPrimaryClient().HeadObject(&s3.HeadObjectInput{ + Bucket: aws.String(testBucket), + Key: aws.String(dstKey), + }) + require.NoError(t, err, "HEAD on copied destination must succeed") + assert.Equal(t, int64(len(srcData)), aws.Int64Value(head.ContentLength), + "destination ContentLength must match source") + + // The original bug: GET would return 500 with "data integrity error". + // Now the destination is a real local object with chunks, so GET reads + // it back byte-for-byte. + t.Log("Reading destination back; pre-fix this returned 500 'data integrity error'") + resp, err := getPrimaryClient().GetObject(&s3.GetObjectInput{ + Bucket: aws.String(testBucket), + Key: aws.String(dstKey), + }) + require.NoError(t, err, "GET on copied destination must succeed") + defer resp.Body.Close() + + got, err := io.ReadAll(resp.Body) + require.NoError(t, err) + assert.Equal(t, len(srcData), len(got), "destination size must match source") + gotSum := md5.Sum(got) + assert.Equal(t, srcSum, gotSum, "destination bytes must match source bytes") +} + +// TestRemoteCacheCopyObjectPart exercises the same bug pattern via the +// multipart upload-part-copy path (CopyObjectPartHandler), which iterates +// the source entry's chunks just like CopyObjectHandler. A remote-only +// source previously produced a part with size > 0 and no data; pre-fix +// the completed multipart upload was unreadable. +func TestRemoteCacheCopyObjectPart(t *testing.T) { + checkServersRunning(t) + + srcKey := fmt.Sprintf("copypart-src-%d.bin", time.Now().UnixNano()) + dstKey := fmt.Sprintf("copypart-dst-%d.bin", time.Now().UnixNano()) + + // 6 MiB so a single 0-5MiB part covers the lower half and exercises a + // real range-read against the cached chunks (parts must be >= 5 MiB + // per S3's MultipartUpload rules, except the last). + srcData := make([]byte, 6*1024*1024) + for i := range srcData { + srcData[i] = byte((i * 7) % 256) + } + + t.Log("Uploading source object to primary (local)") + uploadToPrimary(t, srcKey, srcData) + + t.Log("Uncaching source: pushes data to remote and removes local chunks") + uncacheLocal(t, srcKey) + + t.Log("Initiating multipart upload on the destination") + create, err := getPrimaryClient().CreateMultipartUpload(&s3.CreateMultipartUploadInput{ + Bucket: aws.String(testBucket), + Key: aws.String(dstKey), + }) + require.NoError(t, err) + uploadID := aws.StringValue(create.UploadId) + defer func() { + // Best-effort abort if the test fails mid-flight. + _, _ = getPrimaryClient().AbortMultipartUpload(&s3.AbortMultipartUploadInput{ + Bucket: aws.String(testBucket), + Key: aws.String(dstKey), + UploadId: aws.String(uploadID), + }) + }() + + t.Log("UploadPartCopy from remote-only source range") + part1Range := fmt.Sprintf("bytes=0-%d", 5*1024*1024-1) + cp1, err := getPrimaryClient().UploadPartCopy(&s3.UploadPartCopyInput{ + Bucket: aws.String(testBucket), + Key: aws.String(dstKey), + UploadId: aws.String(uploadID), + PartNumber: aws.Int64(1), + CopySource: aws.String(testBucket + "/" + srcKey), + CopySourceRange: aws.String(part1Range), + }) + require.NoError(t, err, "UploadPartCopy from remote-only source must succeed") + require.NotNil(t, cp1.CopyPartResult) + + part2Range := fmt.Sprintf("bytes=%d-%d", 5*1024*1024, 6*1024*1024-1) + cp2, err := getPrimaryClient().UploadPartCopy(&s3.UploadPartCopyInput{ + Bucket: aws.String(testBucket), + Key: aws.String(dstKey), + UploadId: aws.String(uploadID), + PartNumber: aws.Int64(2), + CopySource: aws.String(testBucket + "/" + srcKey), + CopySourceRange: aws.String(part2Range), + }) + require.NoError(t, err, "second UploadPartCopy must succeed") + require.NotNil(t, cp2.CopyPartResult) + + t.Log("Completing multipart upload") + _, err = getPrimaryClient().CompleteMultipartUpload(&s3.CompleteMultipartUploadInput{ + Bucket: aws.String(testBucket), + Key: aws.String(dstKey), + UploadId: aws.String(uploadID), + MultipartUpload: &s3.CompletedMultipartUpload{ + Parts: []*s3.CompletedPart{ + {ETag: cp1.CopyPartResult.ETag, PartNumber: aws.Int64(1)}, + {ETag: cp2.CopyPartResult.ETag, PartNumber: aws.Int64(2)}, + }, + }, + }) + require.NoError(t, err, "CompleteMultipartUpload must succeed") + + t.Log("Reading destination back; pre-fix this returned 500 'data integrity error'") + resp, err := getPrimaryClient().GetObject(&s3.GetObjectInput{ + Bucket: aws.String(testBucket), + Key: aws.String(dstKey), + }) + require.NoError(t, err) + defer resp.Body.Close() + + got, err := io.ReadAll(resp.Body) + require.NoError(t, err) + assert.True(t, bytes.Equal(srcData, got), + "destination bytes must match source after multipart copy from remote-only source") +} diff --git a/weed/s3api/s3api_object_handlers.go b/weed/s3api/s3api_object_handlers.go index b0f2bfff8..8e5abaac2 100644 --- a/weed/s3api/s3api_object_handlers.go +++ b/weed/s3api/s3api_object_handlers.go @@ -993,8 +993,10 @@ func (s3a *S3ApiServer) streamFromVolumeServers(w http.ResponseWriter, r *http.R // This handles the case where initial caching attempt timed out or failed if entry.IsInRemoteOnly() { glog.V(1).Infof("streamFromVolumeServers: entry is remote-only, attempting to cache before streaming") - // Try to cache the remote object synchronously (like filer does) - cachedEntry := s3a.cacheRemoteObjectForStreaming(r, entry, bucket, object, versionId) + // Latest-version reads carry an empty query versionId even when the + // entry lives at .versions/v_; resolve from the entry itself. + cacheVersionId := resolvedSourceVersionId(versionId, entry) + cachedEntry := s3a.cacheRemoteObjectForStreaming(r, entry, bucket, object, cacheVersionId) if cachedEntry != nil && len(cachedEntry.GetChunks()) > 0 { chunks = cachedEntry.GetChunks() entry = cachedEntry @@ -3099,21 +3101,27 @@ func (s3a *S3ApiServer) cacheRemoteObjectWithDedup(ctx context.Context, bucket, return entry } -// cacheRemoteObjectForStreaming caches a remote-only object to the local cluster for streaming. -// This is called from streamFromVolumeServers when the initial caching attempt timed out or failed. -// Uses the request context (no artificial timeout) to allow the caching to complete. -// For versioned objects, versionId determines the correct path in .versions/ directory. -func (s3a *S3ApiServer) cacheRemoteObjectForStreaming(r *http.Request, entry *filer_pb.Entry, bucket, object, versionId string) *filer_pb.Entry { - var dir, name string +func (s3a *S3ApiServer) buildVersionedRemoteObjectPath(bucket, object, versionId string) (dir, name string) { if versionId != "" && versionId != "null" { - // This is a specific version - entry is located at /buckets//.versions/v_ normalizedObject := s3_constants.NormalizeObjectKey(object) - dir = s3a.bucketDir(bucket) + "/" + normalizedObject + s3_constants.VersionsFolder - name = s3a.getVersionFileName(versionId) - } else { - // Non-versioned object or "null" version - lives at the main path - dir, name = s3a.buildRemoteObjectPath(bucket, object) + return s3a.bucketDir(bucket) + "/" + normalizedObject + s3_constants.VersionsFolder, s3a.getVersionFileName(versionId) } + return s3a.buildRemoteObjectPath(bucket, object) +} + +// cachedEntryHasLocalData reports whether a cache response carries data the +// copy path can read locally. The streaming caller uses a stricter chunks- +// only check inline since its downstream cannot read from inline Content. +func cachedEntryHasLocalData(entry *filer_pb.Entry) bool { + return entry != nil && (len(entry.GetChunks()) > 0 || len(entry.Content) > 0) +} + +// cacheRemoteObjectForStreaming caches a remote-only object to the local cluster for streaming. +// Uses the request context (no artificial timeout) so the caching can complete. +// Returns the cached entry only when chunks are present; the streaming caller +// is not wired to read inline Content from a cache result here. +func (s3a *S3ApiServer) cacheRemoteObjectForStreaming(r *http.Request, entry *filer_pb.Entry, bucket, object, versionId string) *filer_pb.Entry { + dir, name := s3a.buildVersionedRemoteObjectPath(bucket, object, versionId) glog.V(1).Infof("cacheRemoteObjectForStreaming: caching %s/%s (remote size: %d, versionId: %s)", dir, name, entry.RemoteEntry.RemoteSize, versionId) @@ -3130,3 +3138,50 @@ func (s3a *S3ApiServer) cacheRemoteObjectForStreaming(r *http.Request, entry *fi return nil } + +// cacheRemoteObjectForCopy caches a remote-only source before CopyObject / +// CopyObjectPart reads it; otherwise the copy would write a destination with +// FileSize > 0 but no chunks/content. Bounded so a stuck cache can't hang +// the copy. Returns nil if caching failed or produced no local data. +func (s3a *S3ApiServer) cacheRemoteObjectForCopy(ctx context.Context, bucket, object, versionId string) *filer_pb.Entry { + const cacheTimeout = 30 * time.Second + cacheCtx, cancel := context.WithTimeout(ctx, cacheTimeout) + defer cancel() + + dir, name := s3a.buildVersionedRemoteObjectPath(bucket, object, versionId) + + glog.V(1).Infof("cacheRemoteObjectForCopy: caching %s/%s (versionId: %s)", dir, name, versionId) + + cachedEntry, err := s3a.doCacheRemoteObject(cacheCtx, dir, name) + if err != nil { + if errors.Is(err, context.DeadlineExceeded) { + glog.Warningf("cacheRemoteObjectForCopy: timeout caching %s/%s after %v", dir, name, cacheTimeout) + } else { + glog.Errorf("cacheRemoteObjectForCopy: failed to cache %s/%s: %v", dir, name, err) + } + return nil + } + + if cachedEntryHasLocalData(cachedEntry) { + glog.V(1).Infof("cacheRemoteObjectForCopy: successfully cached %s/%s (chunks=%d, inline=%d)", dir, name, len(cachedEntry.GetChunks()), len(cachedEntry.Content)) + return cachedEntry + } + + return nil +} + +// resolvedSourceVersionId falls back to the version recorded on the entry +// when the request didn't carry one — necessary for latest-version reads +// in versioning-enabled buckets, where the entry lives at .versions/v_. +func resolvedSourceVersionId(requestedVersionId string, entry *filer_pb.Entry) string { + if requestedVersionId != "" { + return requestedVersionId + } + if entry == nil || entry.Extended == nil { + return "" + } + if v, ok := entry.Extended[s3_constants.ExtVersionIdKey]; ok { + return string(v) + } + return "" +} diff --git a/weed/s3api/s3api_object_handlers_copy.go b/weed/s3api/s3api_object_handlers_copy.go index c7e4808d6..2add4214f 100644 --- a/weed/s3api/s3api_object_handlers_copy.go +++ b/weed/s3api/s3api_object_handlers_copy.go @@ -102,6 +102,20 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request return } + // Cache remote-only sources before copying; otherwise the copy path below + // writes a destination with FileSize > 0 but no chunks/content. + if entry.IsInRemoteOnly() { + cacheVersionId := resolvedSourceVersionId(srcVersionId, entry) + cachedEntry := s3a.cacheRemoteObjectForCopy(r.Context(), srcBucket, srcObject, cacheVersionId) + if cachedEntry == nil { + glog.Errorf("CopyObjectHandler: failed to cache remote-only source %s/%s (version %q)", srcBucket, srcObject, cacheVersionId) + w.Header().Set("Retry-After", "5") + s3err.WriteErrorResponse(w, r, s3err.ErrServiceUnavailable) + return + } + entry = cachedEntry + } + sameDestination := srcBucket == dstBucket && srcObject == dstObject if sameDestination && !(replaceMeta || replaceTagging) { s3err.WriteErrorResponse(w, r, s3err.ErrInvalidCopyDest) @@ -691,6 +705,20 @@ func (s3a *S3ApiServer) CopyObjectPartHandler(w http.ResponseWriter, r *http.Req return } + // Cache remote-only sources before copying; the part-copy paths below + // iterate entry.GetChunks() and would otherwise produce an empty part. + if entry.IsInRemoteOnly() { + cacheVersionId := resolvedSourceVersionId(srcVersionId, entry) + cachedEntry := s3a.cacheRemoteObjectForCopy(r.Context(), srcBucket, srcObject, cacheVersionId) + if cachedEntry == nil { + glog.Errorf("CopyObjectPartHandler: failed to cache remote-only source %s/%s (version %q)", srcBucket, srcObject, cacheVersionId) + w.Header().Set("Retry-After", "5") + s3err.WriteErrorResponse(w, r, s3err.ErrServiceUnavailable) + return + } + entry = cachedEntry + } + // Validate conditional copy headers if err := s3a.validateConditionalCopyHeaders(r, entry); err != s3err.ErrNone { s3err.WriteErrorResponse(w, r, err) diff --git a/weed/s3api/s3api_remote_storage_test.go b/weed/s3api/s3api_remote_storage_test.go index 7d5963f5d..bda3f938d 100644 --- a/weed/s3api/s3api_remote_storage_test.go +++ b/weed/s3api/s3api_remote_storage_test.go @@ -271,3 +271,216 @@ func removeDuplicateSlashesTest(s string) string { } return s } + +// TestResolvedSourceVersionId pins that latest-version reads in a versioning- +// enabled bucket fall back to the entry's recorded version id when the +// request carried none, so cache paths target .versions/v_ correctly. +func TestResolvedSourceVersionId(t *testing.T) { + tests := []struct { + name string + requested string + entry *filer_pb.Entry + expected string + }{ + { + name: "explicit request versionId wins", + requested: "abc123", + entry: &filer_pb.Entry{ + Extended: map[string][]byte{ + s3_constants.ExtVersionIdKey: []byte("ignored"), + }, + }, + expected: "abc123", + }, + { + name: "empty request falls back to entry version (latest in versioned bucket)", + requested: "", + entry: &filer_pb.Entry{ + Extended: map[string][]byte{ + s3_constants.ExtVersionIdKey: []byte("xyz789"), + }, + }, + expected: "xyz789", + }, + { + name: "empty request and pre-versioning entry stays empty", + requested: "", + entry: &filer_pb.Entry{ + Extended: map[string][]byte{}, + }, + expected: "", + }, + { + name: "empty request and nil Extended stays empty", + requested: "", + entry: &filer_pb.Entry{Extended: nil}, + expected: "", + }, + { + name: "nil entry tolerated when no request version", + requested: "", + entry: nil, + expected: "", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.expected, resolvedSourceVersionId(tt.requested, tt.entry)) + }) + } +} + +func TestCachedEntryHasLocalData(t *testing.T) { + tests := []struct { + name string + entry *filer_pb.Entry + expected bool + }{ + { + name: "nil entry is not a hit", + entry: nil, + expected: false, + }, + { + name: "entry with chunks is a hit", + entry: &filer_pb.Entry{ + Chunks: []*filer_pb.FileChunk{{FileId: "1,abc", Size: 10}}, + }, + expected: true, + }, + { + name: "entry with inline content is a hit", + entry: &filer_pb.Entry{ + Content: []byte("small file body"), + }, + expected: true, + }, + { + name: "entry with neither chunks nor content is not a hit", + entry: &filer_pb.Entry{}, + expected: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.expected, cachedEntryHasLocalData(tt.entry)) + }) + } +} + +// TestCopyObjectRemoteOnlySourceDetection guards against regressing the +// remote-only source case: such a source used to fall through to +// CopyObject's inline branch with empty Content, producing a destination +// with FileSize > 0 but no chunks. +func TestCopyObjectRemoteOnlySourceDetection(t *testing.T) { + tests := []struct { + name string + entry *filer_pb.Entry + expectRemoteOnly bool + expectInlineBranchHit bool + expectBrokenWithoutFix bool + }{ + { + name: "remote-only object with size and no chunks/content", + entry: &filer_pb.Entry{ + Name: "file-1234-audio.mp3", + Attributes: &filer_pb.FuseAttributes{ + FileSize: 16018804, + }, + Chunks: nil, + Content: nil, + RemoteEntry: &filer_pb.RemoteEntry{ + RemoteSize: 16018804, + }, + }, + expectRemoteOnly: true, + expectInlineBranchHit: true, + expectBrokenWithoutFix: true, + }, + { + name: "local file with chunks - copy works fine, fix does not engage", + entry: &filer_pb.Entry{ + Name: "local.bin", + Attributes: &filer_pb.FuseAttributes{ + FileSize: 1024, + }, + Chunks: []*filer_pb.FileChunk{ + {FileId: "1,abc", Size: 1024, Offset: 0}, + }, + }, + expectRemoteOnly: false, + expectInlineBranchHit: false, + expectBrokenWithoutFix: false, + }, + { + name: "small inline file (no chunks, has Content) - hits inline branch but not broken", + entry: &filer_pb.Entry{ + Name: "tiny.txt", + Attributes: &filer_pb.FuseAttributes{ + FileSize: 5, + }, + Content: []byte("hello"), + }, + expectRemoteOnly: false, + expectInlineBranchHit: true, + expectBrokenWithoutFix: false, + }, + { + name: "remote entry already cached (has chunks) - fix does not engage", + entry: &filer_pb.Entry{ + Name: "cached.dat", + Attributes: &filer_pb.FuseAttributes{ + FileSize: 2048, + }, + Chunks: []*filer_pb.FileChunk{ + {FileId: "2,def", Size: 2048, Offset: 0}, + }, + RemoteEntry: &filer_pb.RemoteEntry{ + RemoteSize: 2048, + }, + }, + expectRemoteOnly: false, + expectInlineBranchHit: false, + expectBrokenWithoutFix: false, + }, + { + // Zero-byte remote objects must not enter the cache branch: + // IsInRemoteOnly requires RemoteSize > 0, so CopyObject's + // pre-existing inline branch handles them with no 503. + name: "zero-byte remote object - fix does not engage, inline branch handles it", + entry: &filer_pb.Entry{ + Name: "empty-on-remote.txt", + Attributes: &filer_pb.FuseAttributes{ + FileSize: 0, + }, + RemoteEntry: &filer_pb.RemoteEntry{ + RemoteSize: 0, + }, + }, + expectRemoteOnly: false, + expectInlineBranchHit: true, + expectBrokenWithoutFix: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.expectRemoteOnly, tt.entry.IsInRemoteOnly()) + + // Mirror the inline branch in s3api_object_handlers_copy.go: + // if entry.Attributes.FileSize == 0 || len(entry.GetChunks()) == 0 + inlineBranchHit := tt.entry.Attributes != nil && + (tt.entry.Attributes.FileSize == 0 || len(tt.entry.GetChunks()) == 0) + assert.Equal(t, tt.expectInlineBranchHit, inlineBranchHit) + + // The broken shape: inline branch fires, no inline content, FileSize > 0. + brokenWithoutFix := inlineBranchHit && + len(tt.entry.Content) == 0 && + tt.entry.Attributes != nil && + tt.entry.Attributes.FileSize > 0 + assert.Equal(t, tt.expectBrokenWithoutFix, brokenWithoutFix) + }) + } +}