From d5cff0a38fd29d01e8500d712fe4ffaf7121f765 Mon Sep 17 00:00:00 2001 From: Alex Russell Date: Sat, 10 Oct 2026 19:39:17 +0200 Subject: [PATCH] filer: copy a remote-only entry on the remote when renaming it Renaming a file under a remote.mount whose content exists only on the remote (never cached, or remote.uncache'd) lost it: the new entry kept the old RemoteEntry, deleting the old entry deleted the old object, and with no chunks nothing ever wrote the new key. filer.remote.sync could not finish the rename either, since the old object was already gone. moveSelfEntry now copies such an entry's object to the new key before it touches anything, and the new entry gets the RemoteEntry of the copy. The copy is a new optional RemoteStorageObjectCopier interface, implemented for S3 with CopyObject up to 5 GB and UploadPartCopy above that; the multipart copy carries the source's metadata and, where the remote supports tagging, its tags. When an existing target is overwritten, its delete keeps the remote object (the copy replaced it) and its event is marked metadata-only so the sync leaves it alone too. A remote-only rename that leaves its mount, or whose remote cannot copy, is refused instead of deleting the only copy. Empty entries and entries with local data rename as before. UpdateFileMetadata builds its copy source with the same helper, which escapes each key segment and keeps the slashes. --- weed/filer/filer_lazy_remote.go | 45 +++ weed/remote_storage/remote_storage.go | 8 + weed/remote_storage/s3/s3_storage_client.go | 156 +++++++++- .../s3/s3_storage_client_copy_test.go | 160 ++++++++++ weed/server/filer_grpc_server_rename.go | 29 +- .../filer_grpc_server_rename_remote_test.go | 282 ++++++++++++++++++ 6 files changed, 677 insertions(+), 3 deletions(-) create mode 100644 weed/remote_storage/s3/s3_storage_client_copy_test.go create mode 100644 weed/server/filer_grpc_server_rename_remote_test.go diff --git a/weed/filer/filer_lazy_remote.go b/weed/filer/filer_lazy_remote.go index 7c642cd6a..0cf4f6eff 100644 --- a/weed/filer/filer_lazy_remote.go +++ b/weed/filer/filer_lazy_remote.go @@ -225,3 +225,48 @@ func (f *Filer) maybeDeleteFromRemote(ctx context.Context, entry *Entry) (bool, glog.V(3).InfofCtx(ctx, "maybeDeleteFromRemote: deleted %s from remote", entry.FullPath) return true, nil } + +// CopyRemoteOnlyEntry copies the remote object of a renamed entry whose +// content exists only on the remote to the key of newPath, and returns the +// RemoteEntry of the copy. The rename deletes the old key along with the old +// entry, and with no local data nothing would write the new key. It returns +// nil when no copy is needed: the entry has local data, which the remote sync +// uploads to the new key, it is empty, which the sync writes as an empty +// object, or it is not under a remote mount. A rename that +// moves such an entry out of its mount, or onto a remote that cannot copy, is +// refused, since it would delete the only copy of the content. +func (f *Filer) CopyRemoteOnlyEntry(ctx context.Context, entry *Entry, oldPath, newPath util.FullPath) (*filer_pb.RemoteEntry, error) { + if f.RemoteStorage == nil || entry.IsDirectory() || !entry.IsInRemoteOnly() || len(entry.Content) > 0 { + return nil, nil + } + + mountDir, remoteLoc := f.RemoteStorage.FindMountDirectory(oldPath) + if remoteLoc == nil { + return nil, nil + } + if newMountDir, _ := f.RemoteStorage.FindMountDirectory(newPath); newMountDir != mountDir { + return nil, fmt.Errorf("%s exists only on the remote of mount %s and cannot be renamed out of it; cache it first", oldPath, mountDir) + } + + remoteConf, found := f.RemoteStorage.GetRemoteStorageConf(remoteLoc.Name) + if !found { + return nil, fmt.Errorf("resolve remote storage client for %s: not found", oldPath) + } + client, clientErr := f.buildRemoteStorageClient(ctx, remoteConf) + if clientErr != nil { + return nil, fmt.Errorf("resolve remote storage client for %s: %w", oldPath, clientErr) + } + copier, ok := client.(remote_storage.RemoteStorageObjectCopier) + if !ok { + return nil, fmt.Errorf("%s exists only on the remote and remote storage type %s cannot copy it to the new name; cache it first", oldPath, remoteConf.Type) + } + + src := MapFullPathToRemoteStorageLocation(mountDir, remoteLoc, oldPath) + dst := MapFullPathToRemoteStorageLocation(mountDir, remoteLoc, newPath) + remoteEntry, err := copier.CopyFile(src, dst) + if err != nil { + return nil, fmt.Errorf("copy remote %s to %s: %w", remote_storage.FormatLocation(src), remote_storage.FormatLocation(dst), err) + } + glog.V(3).InfofCtx(ctx, "CopyRemoteOnlyEntry: copied %s to %s", remote_storage.FormatLocation(src), remote_storage.FormatLocation(dst)) + return remoteEntry, nil +} diff --git a/weed/remote_storage/remote_storage.go b/weed/remote_storage/remote_storage.go index f1845e1b0..0cefe362a 100644 --- a/weed/remote_storage/remote_storage.go +++ b/weed/remote_storage/remote_storage.go @@ -110,6 +110,14 @@ type RemoteStorageStreamReader interface { ReadFileAsStream(ctx context.Context, loc *remote_pb.RemoteStorageLocation, offset int64, size int64) (reader io.ReadCloser, err error) } +// RemoteStorageObjectCopier is an optional interface for remote storage clients +// that copy an object to another key within the storage, without the content +// passing through the client. A rename of an entry whose content exists only +// on the remote needs it. +type RemoteStorageObjectCopier interface { + CopyFile(src *remote_pb.RemoteStorageLocation, dst *remote_pb.RemoteStorageLocation) (remoteEntry *filer_pb.RemoteEntry, err error) +} + // CacheWaitTimeout is how long a read of an uncached remote-only object waits // for the local cache before serving another way: small files wait longer since // their cache completes quickly, large files fail fast for better TTFB. A mount diff --git a/weed/remote_storage/s3/s3_storage_client.go b/weed/remote_storage/s3/s3_storage_client.go index a57a063cf..bcc9464cc 100644 --- a/weed/remote_storage/s3/s3_storage_client.go +++ b/weed/remote_storage/s3/s3_storage_client.go @@ -610,7 +610,7 @@ func (s *s3RemoteStorageClient) UpdateFileMetadata(loc *remote_pb.RemoteStorageL copyInput := &s3.CopyObjectInput{ Bucket: aws.String(loc.Bucket), Key: aws.String(key), - CopySource: aws.String(url.PathEscape(loc.Bucket + "/" + key)), + CopySource: aws.String(s3CopySource(loc.Bucket, key)), MetadataDirective: aws.String(s3.MetadataDirectiveReplace), Metadata: headOut.Metadata, ContentType: headOut.ContentType, @@ -663,6 +663,160 @@ func (s *s3RemoteStorageClient) UpdateFileMetadata(loc *remote_pb.RemoteStorageL } return } + +// s3CopyPartSize is the smallest part a multipart copy uses; parts grow so +// the largest object stays within the 10000-part limit +const s3CopyPartSize = 512 * 1024 * 1024 + +// CopyFile copies the object at src to dst on the server side, with a single +// CopyObject up to the 5 GB limit and a multipart copy above it. The copy +// keeps the source's metadata. +func (s *s3RemoteStorageClient) CopyFile(src *remote_pb.RemoteStorageLocation, dst *remote_pb.RemoteStorageLocation) (remoteEntry *filer_pb.RemoteEntry, err error) { + srcKey, dstKey := src.Path[1:], dst.Path[1:] + headOut, err := s.conn.HeadObject(&s3.HeadObjectInput{ + Bucket: aws.String(src.Bucket), + Key: aws.String(srcKey), + }) + if err != nil { + if reqErr, ok := err.(awserr.RequestFailure); ok && reqErr.StatusCode() == http.StatusNotFound { + return nil, remote_storage.ErrRemoteObjectNotFound + } + return nil, fmt.Errorf("stat s3 %s/%s before copy: %w", src.Bucket, srcKey, err) + } + copySource := aws.String(s3CopySource(src.Bucket, srcKey)) + var storageClass *string + if s.conf.S3StorageClass != "" { + storageClass = aws.String(s.conf.S3StorageClass) + } + + if size := aws.Int64Value(headOut.ContentLength); size <= s3CopyObjectSizeLimit { + if _, err = s.conn.CopyObject(&s3.CopyObjectInput{ + Bucket: aws.String(dst.Bucket), + Key: aws.String(dstKey), + CopySource: copySource, + StorageClass: storageClass, + }); err != nil { + return nil, fmt.Errorf("copy s3 %s/%s to %s/%s: %w", src.Bucket, srcKey, dst.Bucket, dstKey, err) + } + } else if err = s.multipartCopy(headOut, src, copySource, size, dst.Bucket, dstKey, storageClass); err != nil { + return nil, fmt.Errorf("copy s3 %s/%s to %s/%s: %w", src.Bucket, srcKey, dst.Bucket, dstKey, err) + } + + return s.readFileRemoteEntry(dst) +} + +// s3CopySource escapes each segment of the copy source and keeps the slashes, +// which every S3 implementation reads back +func s3CopySource(bucket, key string) string { + segments := strings.Split(bucket+"/"+key, "/") + for i, segment := range segments { + segments[i] = url.PathEscape(segment) + } + return strings.Join(segments, "/") +} + +// s3TaggingHeader encodes tags as the query string x-amz-tagging takes, with +// spaces as %20: a + is read back as a space by some implementations and +// literally by others +func s3TaggingHeader(tagSet []*s3.Tag) string { + escape := func(s string) string { + return strings.ReplaceAll(url.QueryEscape(s), "+", "%20") + } + pairs := make([]string, 0, len(tagSet)) + for _, tag := range tagSet { + pairs = append(pairs, escape(aws.StringValue(tag.Key))+"="+escape(aws.StringValue(tag.Value))) + } + return strings.Join(pairs, "&") +} + +// multipartCopy copies an object above the CopyObject limit part by part. A +// multipart upload does not inherit the source's metadata or tags, so they are +// carried over like CopyObject would: metadata from the source's HeadObject, +// tags when the remote supports tagging. +func (s *s3RemoteStorageClient) multipartCopy(headOut *s3.HeadObjectOutput, src *remote_pb.RemoteStorageLocation, copySource *string, size int64, bucket, key string, storageClass *string) error { + createInput := &s3.CreateMultipartUploadInput{ + Bucket: aws.String(bucket), + Key: aws.String(key), + Metadata: headOut.Metadata, + ContentType: headOut.ContentType, + ContentEncoding: headOut.ContentEncoding, + CacheControl: headOut.CacheControl, + ContentDisposition: headOut.ContentDisposition, + ContentLanguage: headOut.ContentLanguage, + WebsiteRedirectLocation: headOut.WebsiteRedirectLocation, + StorageClass: storageClass, + } + if headOut.Expires != nil { + if expires, parseErr := http.ParseTime(*headOut.Expires); parseErr == nil { + createInput.Expires = aws.Time(expires) + } + } + if s.conf.S3SupportTagging { + tagOut, tagErr := s.conn.GetObjectTagging(&s3.GetObjectTaggingInput{ + Bucket: aws.String(src.Bucket), + Key: aws.String(src.Path[1:]), + }) + if tagErr != nil { + return fmt.Errorf("read tags: %w", tagErr) + } + if len(tagOut.TagSet) > 0 { + createInput.Tagging = aws.String(s3TaggingHeader(tagOut.TagSet)) + } + } + createOut, err := s.conn.CreateMultipartUpload(createInput) + if err != nil { + return err + } + + partSize := int64(s3CopyPartSize) + for partSize*10000 < size { + partSize *= 2 + } + var parts []*s3.CompletedPart + for offset, partNumber := int64(0), int64(1); offset < size; offset, partNumber = offset+partSize, partNumber+1 { + partOut, partErr := s.conn.UploadPartCopy(&s3.UploadPartCopyInput{ + Bucket: aws.String(bucket), + Key: aws.String(key), + UploadId: createOut.UploadId, + PartNumber: aws.Int64(partNumber), + CopySource: copySource, + CopySourceRange: aws.String(fmt.Sprintf("bytes=%d-%d", offset, min(offset+partSize, size)-1)), + }) + if partErr == nil && partOut.CopyPartResult == nil { + partErr = errors.New("no part result") + } + if partErr != nil { + s.abortMultipartUpload(bucket, key, createOut.UploadId) + return fmt.Errorf("copy part %d: %w", partNumber, partErr) + } + parts = append(parts, &s3.CompletedPart{ + ETag: partOut.CopyPartResult.ETag, + PartNumber: aws.Int64(partNumber), + }) + } + + if _, err = s.conn.CompleteMultipartUpload(&s3.CompleteMultipartUploadInput{ + Bucket: aws.String(bucket), + Key: aws.String(key), + UploadId: createOut.UploadId, + MultipartUpload: &s3.CompletedMultipartUpload{Parts: parts}, + }); err != nil { + s.abortMultipartUpload(bucket, key, createOut.UploadId) + return err + } + return nil +} + +func (s *s3RemoteStorageClient) abortMultipartUpload(bucket, key string, uploadId *string) { + if _, err := s.conn.AbortMultipartUpload(&s3.AbortMultipartUploadInput{ + Bucket: aws.String(bucket), + Key: aws.String(key), + UploadId: uploadId, + }); err != nil { + glog.Warningf("abort multipart copy to s3 %s/%s: %v", bucket, key, err) + } +} + func (s *s3RemoteStorageClient) DeleteFile(loc *remote_pb.RemoteStorageLocation) (err error) { _, err = s.conn.DeleteObject(&s3.DeleteObjectInput{ Bucket: aws.String(loc.Bucket), diff --git a/weed/remote_storage/s3/s3_storage_client_copy_test.go b/weed/remote_storage/s3/s3_storage_client_copy_test.go new file mode 100644 index 000000000..f30af7bee --- /dev/null +++ b/weed/remote_storage/s3/s3_storage_client_copy_test.go @@ -0,0 +1,160 @@ +package s3 + +import ( + "errors" + "net/http" + "testing" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/aws/awserr" + awss3 "github.com/aws/aws-sdk-go/service/s3" + "github.com/aws/aws-sdk-go/service/s3/s3iface" + "github.com/seaweedfs/seaweedfs/weed/pb/remote_pb" + "github.com/seaweedfs/seaweedfs/weed/remote_storage" + "github.com/stretchr/testify/require" +) + +// copyObjectMock serves HeadObject for a source of size bytes and records the +// copy calls. +type copyObjectMock struct { + s3iface.S3API + size int64 + missing bool + partErr error + copyInputs []*awss3.CopyObjectInput + partInputs []*awss3.UploadPartCopyInput + createInput *awss3.CreateMultipartUploadInput + tagInputs []*awss3.GetObjectTaggingInput + completed *awss3.CompleteMultipartUploadInput + aborted bool +} + +func (m *copyObjectMock) HeadObject(input *awss3.HeadObjectInput) (*awss3.HeadObjectOutput, error) { + if m.missing { + return nil, awserr.NewRequestFailure(awserr.New("NotFound", "not found", nil), http.StatusNotFound, "") + } + return &awss3.HeadObjectOutput{ + ContentLength: aws.Int64(m.size), + ContentType: aws.String("image/jpeg"), + Expires: aws.String("Wed, 21 Oct 2026 07:28:00 GMT"), + WebsiteRedirectLocation: aws.String("/elsewhere"), + ETag: aws.String(`"etag-` + aws.StringValue(input.Key) + `"`), + }, nil +} + +func (m *copyObjectMock) GetObjectTagging(input *awss3.GetObjectTaggingInput) (*awss3.GetObjectTaggingOutput, error) { + m.tagInputs = append(m.tagInputs, input) + return &awss3.GetObjectTaggingOutput{TagSet: []*awss3.Tag{ + {Key: aws.String("album"), Value: aws.String("2026 & co")}, + {Key: aws.String("a+b"), Value: aws.String("x=y")}, + }}, nil +} + +func (m *copyObjectMock) CopyObject(input *awss3.CopyObjectInput) (*awss3.CopyObjectOutput, error) { + m.copyInputs = append(m.copyInputs, input) + return &awss3.CopyObjectOutput{}, nil +} + +func (m *copyObjectMock) CreateMultipartUpload(input *awss3.CreateMultipartUploadInput) (*awss3.CreateMultipartUploadOutput, error) { + m.createInput = input + return &awss3.CreateMultipartUploadOutput{UploadId: aws.String("upload-1")}, nil +} + +func (m *copyObjectMock) UploadPartCopy(input *awss3.UploadPartCopyInput) (*awss3.UploadPartCopyOutput, error) { + m.partInputs = append(m.partInputs, input) + if m.partErr != nil && len(m.partInputs) == 2 { + return nil, m.partErr + } + return &awss3.UploadPartCopyOutput{CopyPartResult: &awss3.CopyPartResult{ETag: aws.String(`"part"`)}}, nil +} + +func (m *copyObjectMock) CompleteMultipartUpload(input *awss3.CompleteMultipartUploadInput) (*awss3.CompleteMultipartUploadOutput, error) { + m.completed = input + return &awss3.CompleteMultipartUploadOutput{}, nil +} + +func (m *copyObjectMock) AbortMultipartUpload(*awss3.AbortMultipartUploadInput) (*awss3.AbortMultipartUploadOutput, error) { + m.aborted = true + return &awss3.AbortMultipartUploadOutput{}, nil +} + +func newCopyTestClient(mock *copyObjectMock) *s3RemoteStorageClient { + return &s3RemoteStorageClient{conf: &remote_pb.RemoteConf{Name: "test"}, conn: mock} +} + +var ( + copySrc = &remote_pb.RemoteStorageLocation{Name: "test", Bucket: "bucket", Path: "/src/a b.jpg"} + copyDst = &remote_pb.RemoteStorageLocation{Name: "test", Bucket: "bucket", Path: "/dst/a b.jpg"} +) + +func TestS3ClientImplementsObjectCopier(t *testing.T) { + var _ remote_storage.RemoteStorageObjectCopier = &s3RemoteStorageClient{} +} + +func TestS3CopyFileUsesCopyObjectUpToTheLimit(t *testing.T) { + mock := ©ObjectMock{size: s3CopyObjectSizeLimit} + + remoteEntry, err := newCopyTestClient(mock).CopyFile(copySrc, copyDst) + require.NoError(t, err) + + require.Len(t, mock.copyInputs, 1) + require.Equal(t, "bucket/src/a%20b.jpg", aws.StringValue(mock.copyInputs[0].CopySource)) + require.Equal(t, "dst/a b.jpg", aws.StringValue(mock.copyInputs[0].Key)) + require.Nil(t, mock.createInput) + // the RemoteEntry describes the destination + require.Equal(t, `"etag-dst/a b.jpg"`, remoteEntry.RemoteETag) + require.Equal(t, int64(s3CopyObjectSizeLimit), remoteEntry.RemoteSize) +} + +func TestS3CopyFileCopiesLargeObjectsInParts(t *testing.T) { + size := int64(s3CopyObjectSizeLimit) + 1 + mock := ©ObjectMock{size: size} + + _, err := newCopyTestClient(mock).CopyFile(copySrc, copyDst) + require.NoError(t, err) + + require.Empty(t, mock.copyInputs) + require.NotNil(t, mock.createInput) + require.Equal(t, "image/jpeg", aws.StringValue(mock.createInput.ContentType)) + require.Equal(t, "/elsewhere", aws.StringValue(mock.createInput.WebsiteRedirectLocation)) + require.Equal(t, int64(1792567680), aws.TimeValue(mock.createInput.Expires).Unix()) + // tags only where the remote supports tagging + require.Empty(t, mock.tagInputs) + require.Nil(t, mock.createInput.Tagging) + require.Len(t, mock.partInputs, 11) + require.Equal(t, "bytes=0-536870911", aws.StringValue(mock.partInputs[0].CopySourceRange)) + require.Equal(t, "bytes=5368709120-5368709120", aws.StringValue(mock.partInputs[10].CopySourceRange)) + require.Len(t, mock.completed.MultipartUpload.Parts, 11) + require.Equal(t, int64(11), aws.Int64Value(mock.completed.MultipartUpload.Parts[10].PartNumber)) + require.False(t, mock.aborted) +} + +func TestS3CopyFileCarriesTagsIntoAMultipartCopy(t *testing.T) { + mock := ©ObjectMock{size: s3CopyObjectSizeLimit + 1} + client := newCopyTestClient(mock) + client.conf.S3SupportTagging = true + + _, err := client.CopyFile(copySrc, copyDst) + require.NoError(t, err) + + require.Len(t, mock.tagInputs, 1) + require.Equal(t, "src/a b.jpg", aws.StringValue(mock.tagInputs[0].Key)) + require.Equal(t, "album=2026%20%26%20co&a%2Bb=x%3Dy", aws.StringValue(mock.createInput.Tagging)) +} + +func TestS3CopyFileAbortsAFailedMultipartCopy(t *testing.T) { + mock := ©ObjectMock{size: 2 * s3CopyObjectSizeLimit, partErr: errors.New("part failed")} + + _, err := newCopyTestClient(mock).CopyFile(copySrc, copyDst) + require.ErrorContains(t, err, "part failed") + require.True(t, mock.aborted) + require.Nil(t, mock.completed) +} + +func TestS3CopyFileMissingSource(t *testing.T) { + mock := ©ObjectMock{missing: true} + + _, err := newCopyTestClient(mock).CopyFile(copySrc, copyDst) + require.ErrorIs(t, err, remote_storage.ErrRemoteObjectNotFound) + require.Empty(t, mock.copyInputs) +} diff --git a/weed/server/filer_grpc_server_rename.go b/weed/server/filer_grpc_server_rename.go index 020a59c71..ae097ee41 100644 --- a/weed/server/filer_grpc_server_rename.go +++ b/weed/server/filer_grpc_server_rename.go @@ -3,6 +3,7 @@ package weed_server import ( "context" "fmt" + "maps" "path/filepath" "time" @@ -235,8 +236,32 @@ func (fs *FilerServer) moveSelfEntry(ctx context.Context, stream filer_pb.Seawee case !existingTarget.IsDirectory() && entry.IsDirectory(): return fmt.Errorf("%s: %w", existingTarget.FullPath, filer_pb.ErrExistingIsFile) } + } + + // a remote-only entry is copied on the remote before anything changes: + // deleting the old entry deletes the old object + copiedRemote, copyErr := fs.filer.CopyRemoteOnlyEntry(ctx, entry, oldPath, newPath) + if copyErr != nil { + return copyErr + } + remoteEntry := entry.Remote + if copiedRemote != nil { + remoteEntry = copiedRemote + } + + if existingTarget != nil { + targetCtx := filer.WithSuppressedMetadataEvents(ctx) + if copiedRemote != nil { + // the copy has replaced the target's object; keep it + targetCtx = filer.WithKeepRemoteObject(targetCtx) + existingTarget.Extended = maps.Clone(existingTarget.Extended) + if existingTarget.Extended == nil { + existingTarget.Extended = map[string][]byte{} + } + existingTarget.Extended[filer.ExtKeepRemoteObjectKey] = []byte("true") + } if deleteErr := fs.filer.DeleteEntryMetaAndData( - filer.WithSuppressedMetadataEvents(ctx), + targetCtx, newPath, false, false, @@ -258,7 +283,7 @@ func (fs *FilerServer) moveSelfEntry(ctx context.Context, stream filer_pb.Seawee Content: entry.Content, HardLinkCounter: entry.HardLinkCounter, HardLinkId: entry.HardLinkId, - Remote: entry.Remote, + Remote: remoteEntry, Quota: entry.Quota, } if skipTargetLookup { diff --git a/weed/server/filer_grpc_server_rename_remote_test.go b/weed/server/filer_grpc_server_rename_remote_test.go new file mode 100644 index 000000000..27eab9e0a --- /dev/null +++ b/weed/server/filer_grpc_server_rename_remote_test.go @@ -0,0 +1,282 @@ +package weed_server + +import ( + "context" + "errors" + "sync" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/filer" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/pb/remote_pb" + "github.com/seaweedfs/seaweedfs/weed/remote_storage" + "github.com/seaweedfs/seaweedfs/weed/util" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "google.golang.org/protobuf/proto" +) + +// renameRemoteClient records the remote calls a rename makes. It cannot copy; +// copyingRenameRemoteClient adds CopyFile. +type renameRemoteClient struct { + remote_storage.RemoteStorageClient + mu sync.Mutex + calls []string + copyErr error +} + +func (c *renameRemoteClient) record(call string) { + c.mu.Lock() + defer c.mu.Unlock() + c.calls = append(c.calls, call) +} + +func (c *renameRemoteClient) recorded() []string { + c.mu.Lock() + defer c.mu.Unlock() + return append([]string(nil), c.calls...) +} + +func (c *renameRemoteClient) StatFile(*remote_pb.RemoteStorageLocation) (*filer_pb.RemoteEntry, error) { + return nil, remote_storage.ErrRemoteObjectNotFound +} + +func (c *renameRemoteClient) ListDirectory(context.Context, *remote_pb.RemoteStorageLocation, remote_storage.VisitFunc) error { + return nil +} + +func (c *renameRemoteClient) DeleteFile(loc *remote_pb.RemoteStorageLocation) error { + c.record("delete " + loc.Bucket + loc.Path) + return nil +} + +func (c *renameRemoteClient) RemoveDirectory(loc *remote_pb.RemoteStorageLocation) error { + c.record("rmdir " + loc.Bucket + loc.Path) + return nil +} + +type copyingRenameRemoteClient struct { + *renameRemoteClient +} + +func (c copyingRenameRemoteClient) CopyFile(src, dst *remote_pb.RemoteStorageLocation) (*filer_pb.RemoteEntry, error) { + if c.copyErr != nil { + return nil, c.copyErr + } + c.record("copy " + src.Bucket + src.Path + " " + dst.Bucket + dst.Path) + return &filer_pb.RemoteEntry{StorageName: "r1", RemoteSize: 10, RemoteMtime: 1800000000, RemoteETag: `"copy"`}, nil +} + +// newRenameRemoteTestServer mounts /buckets/b, /buckets/c and /data/m on r1, +// through client. +func newRenameRemoteTestServer(t *testing.T, client remote_storage.RemoteStorageClient) (*FilerServer, *renameTestStore) { + t.Helper() + store := newRenameTestStore() + f := newRenameTestFiler(t, store) + f.BuildGuardedRemoteClient = func(context.Context, *remote_pb.RemoteConf, bool) (remote_storage.RemoteStorageClient, error) { + return client, nil + } + mapping, err := proto.Marshal(&remote_pb.RemoteStorageMapping{Mappings: map[string]*remote_pb.RemoteStorageLocation{ + "/buckets/b": {Name: "r1", Bucket: "origin", Path: "/"}, + "/buckets/c": {Name: "r1", Bucket: "other", Path: "/"}, + "/data/m": {Name: "r1", Bucket: "origin", Path: "/m"}, + }}) + require.NoError(t, err) + conf, err := proto.Marshal(&remote_pb.RemoteConf{Name: "r1", Type: "s3"}) + require.NoError(t, err) + for name, content := range map[string][]byte{filer.REMOTE_STORAGE_MOUNT_FILE: mapping, "r1" + filer.REMOTE_STORAGE_CONF_SUFFIX: conf} { + entry := newFileEntry(filer.DirectoryEtcRemote+"/"+name, 1) + entry.Content = content + store.entries[string(entry.FullPath)] = entry + } + require.NoError(t, f.RemoteStorage.LoadRemoteStorageConfigurationsAndMapping(f)) + for _, dir := range []string{"/buckets", "/buckets/b", "/buckets/b/src", "/buckets/b/dst", "/buckets/c", "/data", "/data/m"} { + store.entries[dir] = newDirectoryEntry(dir, 2) + } + return &FilerServer{filer: f, option: &FilerOption{}, entryLockTable: util.NewLockTable[util.FullPath]()}, store +} + +func remoteOnlyRenameEntry(path string, inode uint64) *filer.Entry { + entry := newFileEntry(path, inode) + entry.FileSize = 10 + entry.Remote = &filer_pb.RemoteEntry{StorageName: "r1", RemoteSize: 10, RemoteMtime: 1700000000, RemoteETag: `"orig"`} + return entry +} + +func renameFile(server *FilerServer, oldPath, newPath util.FullPath) error { + oldDir, oldName := oldPath.DirAndName() + newDir, newName := newPath.DirAndName() + _, err := server.AtomicRenameEntry(context.Background(), &filer_pb.AtomicRenameEntryRequest{ + OldDirectory: oldDir, + OldName: oldName, + NewDirectory: newDir, + NewName: newName, + }) + return err +} + +func TestRenameRemoteOnlyEntryCopiesBeforeDeletingTheOldObject(t *testing.T) { + client := copyingRenameRemoteClient{&renameRemoteClient{}} + server, store := newRenameRemoteTestServer(t, client) + store.entries["/buckets/b/src/a.jpg"] = remoteOnlyRenameEntry("/buckets/b/src/a.jpg", 101) + + require.NoError(t, renameFile(server, "/buckets/b/src/a.jpg", "/buckets/b/dst/a.jpg")) + + assert.Equal(t, []string{"copy origin/src/a.jpg origin/dst/a.jpg", "delete origin/src/a.jpg"}, client.recorded()) + _, err := store.FindEntry(context.Background(), "/buckets/b/src/a.jpg") + assert.ErrorIs(t, err, filer_pb.ErrNotFound) + dst, err := store.FindEntry(context.Background(), "/buckets/b/dst/a.jpg") + require.NoError(t, err) + assert.Equal(t, `"copy"`, dst.Remote.RemoteETag) + assert.Equal(t, uint64(101), dst.Attr.Inode) +} + +func TestRenameRemoteOnlyEntryOverTargetKeepsTheCopy(t *testing.T) { + client := copyingRenameRemoteClient{&renameRemoteClient{}} + server, store := newRenameRemoteTestServer(t, client) + store.entries["/buckets/b/src/a.jpg"] = remoteOnlyRenameEntry("/buckets/b/src/a.jpg", 101) + store.entries["/buckets/b/dst/a.jpg"] = remoteOnlyRenameEntry("/buckets/b/dst/a.jpg", 202) + queue := &captureQueue{} + swapNotificationQueue(t, queue) + + require.NoError(t, renameFile(server, "/buckets/b/src/a.jpg", "/buckets/b/dst/a.jpg")) + + // the target's delete must not take the copy that replaced its object + assert.Equal(t, []string{"copy origin/src/a.jpg origin/dst/a.jpg", "delete origin/src/a.jpg"}, client.recorded()) + events := queue.snapshot() + require.Len(t, events, 2) + assert.True(t, filer.IsMetadataOnlyDelete(events[0].notification.OldEntry), "the target's delete event must leave the remote object alone") + assert.False(t, filer.IsMetadataOnlyDelete(events[1].notification.OldEntry)) + dst, err := store.FindEntry(context.Background(), "/buckets/b/dst/a.jpg") + require.NoError(t, err) + assert.Equal(t, uint64(101), dst.Attr.Inode) + assert.Equal(t, `"copy"`, dst.Remote.RemoteETag) +} + +func TestRenameRemoteOnlyEntryCopyFailureChangesNothing(t *testing.T) { + client := copyingRenameRemoteClient{&renameRemoteClient{copyErr: errors.New("copy refused")}} + server, store := newRenameRemoteTestServer(t, client) + store.entries["/buckets/b/src/a.jpg"] = remoteOnlyRenameEntry("/buckets/b/src/a.jpg", 101) + store.entries["/buckets/b/dst/a.jpg"] = remoteOnlyRenameEntry("/buckets/b/dst/a.jpg", 202) + + err := renameFile(server, "/buckets/b/src/a.jpg", "/buckets/b/dst/a.jpg") + require.ErrorContains(t, err, "copy refused") + + assert.Empty(t, client.recorded()) + src, err := store.FindEntry(context.Background(), "/buckets/b/src/a.jpg") + require.NoError(t, err) + assert.Equal(t, `"orig"`, src.Remote.RemoteETag) + dst, err := store.FindEntry(context.Background(), "/buckets/b/dst/a.jpg") + require.NoError(t, err) + assert.Equal(t, uint64(202), dst.Attr.Inode) +} + +func TestRenameRemoteOnlyEntryWithoutCopierIsRefused(t *testing.T) { + client := &renameRemoteClient{} + server, store := newRenameRemoteTestServer(t, client) + store.entries["/buckets/b/src/a.jpg"] = remoteOnlyRenameEntry("/buckets/b/src/a.jpg", 101) + + err := renameFile(server, "/buckets/b/src/a.jpg", "/buckets/b/dst/a.jpg") + require.ErrorContains(t, err, "cannot copy") + + assert.Empty(t, client.recorded()) + _, err = store.FindEntry(context.Background(), "/buckets/b/src/a.jpg") + require.NoError(t, err) + _, err = store.FindEntry(context.Background(), "/buckets/b/dst/a.jpg") + assert.ErrorIs(t, err, filer_pb.ErrNotFound) +} + +func TestRenameRemoteOnlyEntryOutOfItsMountIsRefused(t *testing.T) { + for _, paths := range [][2]util.FullPath{ + {"/buckets/b/src/a.jpg", "/buckets/c/a.jpg"}, + {"/data/m/a.jpg", "/data/a.jpg"}, + } { + t.Run(string(paths[1]), func(t *testing.T) { + client := copyingRenameRemoteClient{&renameRemoteClient{}} + server, store := newRenameRemoteTestServer(t, client) + store.entries[string(paths[0])] = remoteOnlyRenameEntry(string(paths[0]), 101) + + err := renameFile(server, paths[0], paths[1]) + require.ErrorContains(t, err, "cannot be renamed out of it") + + assert.Empty(t, client.recorded()) + _, err = store.FindEntry(context.Background(), paths[0]) + require.NoError(t, err) + _, err = store.FindEntry(context.Background(), paths[1]) + assert.ErrorIs(t, err, filer_pb.ErrNotFound) + }) + } +} + +func TestRenameEmptyRemoteEntryDoesNotCopy(t *testing.T) { + // nothing to lose: no copy, and no refusal even out of the mount or + // without a copier; the sync writes an empty object at a new key + client := &renameRemoteClient{} + server, store := newRenameRemoteTestServer(t, client) + for _, path := range []string{"/buckets/b/src/empty", "/data/m/empty"} { + entry := remoteOnlyRenameEntry(path, 101) + entry.FileSize = 0 + entry.Remote.RemoteSize = 0 + store.entries[path] = entry + } + + require.NoError(t, renameFile(server, "/buckets/b/src/empty", "/buckets/b/dst/empty")) + require.NoError(t, renameFile(server, "/data/m/empty", "/data/empty")) + + assert.Equal(t, []string{"delete origin/src/empty", "delete origin/m/empty"}, client.recorded()) +} + +func TestRenameRemoteOnlyEntryFailingAfterTheCopyKeepsTheContent(t *testing.T) { + client := copyingRenameRemoteClient{&renameRemoteClient{}} + server, store := newRenameRemoteTestServer(t, client) + store.entries["/buckets/b/src/a.jpg"] = remoteOnlyRenameEntry("/buckets/b/src/a.jpg", 101) + store.commitErr = errors.New("commit failed") + + err := renameFile(server, "/buckets/b/src/a.jpg", "/buckets/b/dst/a.jpg") + require.ErrorContains(t, err, "commit failed") + + // remote changes are not rolled back with the store: the old object is + // gone, but only after the copy at the new key was written + assert.Equal(t, []string{"copy origin/src/a.jpg origin/dst/a.jpg", "delete origin/src/a.jpg"}, client.recorded()) +} + +func TestRenameEntryWithLocalDataDoesNotCopy(t *testing.T) { + client := copyingRenameRemoteClient{&renameRemoteClient{}} + server, store := newRenameRemoteTestServer(t, client) + entry := remoteOnlyRenameEntry("/data/m/a.jpg", 101) + entry.Chunks = []*filer_pb.FileChunk{{FileId: "1,0123456789", Size: 10}} + store.entries["/data/m/a.jpg"] = entry + + // even out of the mount: the old object goes, the content stays local + require.NoError(t, renameFile(server, "/data/m/a.jpg", "/data/a.jpg")) + + assert.Equal(t, []string{"delete origin/m/a.jpg"}, client.recorded()) + dst, err := store.FindEntry(context.Background(), "/data/a.jpg") + require.NoError(t, err) + assert.Equal(t, `"orig"`, dst.Remote.RemoteETag) + assert.Len(t, dst.GetChunks(), 1) +} + +func TestRenameDirectoryCopiesRemoteOnlyChildren(t *testing.T) { + client := copyingRenameRemoteClient{&renameRemoteClient{}} + server, store := newRenameRemoteTestServer(t, client) + store.entries["/buckets/b/src/a.jpg"] = remoteOnlyRenameEntry("/buckets/b/src/a.jpg", 101) + local := remoteOnlyRenameEntry("/buckets/b/src/b.jpg", 102) + local.Content = []byte("0123456789") + store.entries["/buckets/b/src/b.jpg"] = local + + require.NoError(t, renameFile(server, "/buckets/b/src", "/buckets/b/moved")) + + assert.Equal(t, []string{ + "copy origin/src/a.jpg origin/moved/a.jpg", + "delete origin/src/a.jpg", + "delete origin/src/b.jpg", + "rmdir origin/src", + }, client.recorded()) + moved, err := store.FindEntry(context.Background(), "/buckets/b/moved/a.jpg") + require.NoError(t, err) + assert.Equal(t, `"copy"`, moved.Remote.RemoteETag) + moved, err = store.FindEntry(context.Background(), "/buckets/b/moved/b.jpg") + require.NoError(t, err) + assert.Equal(t, `"orig"`, moved.Remote.RemoteETag) +}