filer: delete a copied rename's old object only after the commit

Review follow-ups:

- The old key of a copied remote-only entry, and the old directory of a
  rename that copied one, are now deleted once the rename has committed.
  A rename that fails after the copy leaves the source entry and its
  object in place, plus an extra object at the new key.
- Every part of a multipart copy is conditioned on the ETag the copy
  started from, so an object overwritten mid-copy fails the copy
  instead of mixing versions.
- Both copy paths keep the source's server-side encryption settings.
- filer.remote.sync no longer deletes a key that a remote-only entry
  holds by the time the delete event is processed: a file deleted and
  then replaced by a remote-only rename has its content only there.

weed/command/filer_remote_sync_dir.go

@@ -195,7 +195,7 @@ func (option *RemoteSyncOptions) makeEventProcessor(remoteStorage *remote_pb.Rem
 			return updateLocalEntry(option, message.NewParentPath, message.NewEntry, remoteEntry)
 		}
 		if filer_pb.IsDelete(resp) {
-			return processDeleteEvent(client, mountedDir, remoteStorageMountLocation, resp)
+			return processDeleteEvent(filerSource, client, mountedDir, remoteStorageMountLocation, resp)
 		}
 		if message.OldEntry != nil && message.NewEntry != nil {
 			return processUpdateEvent(option, filerSource, *option.storageClass, client, mountedDir, remoteStorageMountLocation, resp)
@@ -212,8 +212,11 @@ func (option *RemoteSyncOptions) makeEventProcessor(remoteStorage *remote_pb.Rem
 // event the inline delete already handled), and GCS reports that case as
 // ErrRemoteObjectNotFound where S3 and Azure answer an idempotent success.
 // Returning the error would pin the sync offset on an event that has nothing
-// left to do.
+// left to do. A file whose path a remote-only entry has taken since keeps its
+// object: a rename copied that entry's content to this key, and it exists
+// nowhere else.
 func processDeleteEvent(
+	filerSource filer_pb.FilerClient,
 	client remote_storage.RemoteStorageClient,
 	mountedDir string,
 	remoteStorageMountLocation *remote_pb.RemoteStorageLocation,
@@ -232,6 +235,16 @@ func processDeleteEvent(
 	}
 	glog.V(2).Infof("delete: %+v", resp)
 	dest := toRemoteStorageLocation(util.FullPath(mountedDir), util.NewFullPath(resp.Directory, message.OldEntry.Name), remoteStorageMountLocation)
+	if !message.OldEntry.IsDirectory {
+		current, err := currentEntry(filerSource, resp.Directory, message.OldEntry.Name)
+		if err != nil {
+			return err
+		}
+		if isRemoteOnly(current) {
+			glog.V(0).Infof("keep %s: a remote-only entry has taken its path since", remote_storage.FormatLocation(dest))
+			return nil
+		}
+	}
 	if message.OldEntry.IsDirectory {
 		glog.V(0).Infof("rmdir  %s", remote_storage.FormatLocation(dest))
 		return client.RemoveDirectory(dest)

weed/command/filer_remote_sync_dir_test.go

@@ -999,7 +999,7 @@ func TestDeleteEventAbsentRemoteObjectIsSuccess(t *testing.T) {
 
 	t.Run("absent object", func(t *testing.T) {
 		remote := &recordingRemote{deleteErr: remote_storage.ErrRemoteObjectNotFound}
-		if err := processDeleteEvent(remote, mountedDir, mountLoc, resp); err != nil {
+		if err := processDeleteEvent(&stubFilerClient{}, remote, mountedDir, mountLoc, resp); err != nil {
 			t.Fatalf("err = %v, want nil: deleting an absent object is complete", err)
 		}
 		if len(remote.deletes) != 1 || !proto.Equal(remote.deletes[0], wantDelete) {
@@ -1011,7 +1011,7 @@ func TestDeleteEventAbsentRemoteObjectIsSuccess(t *testing.T) {
 		marked := proto.Clone(resp).(*filer_pb.SubscribeMetadataResponse)
 		marked.EventNotification.OldEntry.Extended = map[string][]byte{filer.ExtKeepRemoteObjectKey: []byte("true")}
 		remote := &recordingRemote{}
-		if err := processDeleteEvent(remote, mountedDir, mountLoc, marked); err != nil {
+		if err := processDeleteEvent(&stubFilerClient{}, remote, mountedDir, mountLoc, marked); err != nil {
 			t.Fatalf("err = %v", err)
 		}
 		if len(remote.deletes) != 0 {
@@ -1019,9 +1019,41 @@ func TestDeleteEventAbsentRemoteObjectIsSuccess(t *testing.T) {
 		}
 	})
 
+	t.Run("a remote-only entry that took the path keeps the object", func(t *testing.T) {
+		// deleted, then a remote-only file renamed onto the path: the rename
+		// copied its content to this key before the sync got to the delete
+		remote := &recordingRemote{}
+		filerClient := &stubFilerClient{entry: &filer_pb.Entry{
+			Name:        "state.db.tmp",
+			Attributes:  &filer_pb.FuseAttributes{FileSize: 10},
+			RemoteEntry: &filer_pb.RemoteEntry{RemoteSize: 10},
+		}}
+		if err := processDeleteEvent(filerClient, remote, mountedDir, mountLoc, resp); err != nil {
+			t.Fatalf("err = %v", err)
+		}
+		if len(remote.deletes) != 0 {
+			t.Errorf("deletes = %+v, want none", remote.deletes)
+		}
+	})
+
+	t.Run("an entry with local data at the path does not stop the delete", func(t *testing.T) {
+		remote := &recordingRemote{}
+		filerClient := &stubFilerClient{entry: &filer_pb.Entry{
+			Name:        "state.db.tmp",
+			Content:     []byte("local"),
+			RemoteEntry: &filer_pb.RemoteEntry{RemoteSize: 10},
+		}}
+		if err := processDeleteEvent(filerClient, remote, mountedDir, mountLoc, resp); err != nil {
+			t.Fatalf("err = %v", err)
+		}
+		if len(remote.deletes) != 1 {
+			t.Errorf("deletes = %+v, want one", remote.deletes)
+		}
+	})
+
 	t.Run("other failure still fails", func(t *testing.T) {
 		remote := &recordingRemote{deleteErr: errors.New("AccessDenied: Access Denied")}
-		if err := processDeleteEvent(remote, mountedDir, mountLoc, resp); err == nil {
+		if err := processDeleteEvent(&stubFilerClient{}, remote, mountedDir, mountLoc, resp); err == nil {
 			t.Fatal("err = nil, want the delete failure")
 		}
 	})

weed/filer/filer_lazy_remote.go

@@ -226,6 +226,13 @@ func (f *Filer) maybeDeleteFromRemote(ctx context.Context, entry *Entry) (bool,
 	return true, nil
 }
 
+// DeleteFromRemote deletes the object, or for a directory every object, the
+// entry maps to on its mounted remote storage.
+func (f *Filer) DeleteFromRemote(ctx context.Context, entry *Entry) error {
+	_, err := f.maybeDeleteFromRemote(ctx, entry)
+	return err
+}
+
 // 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

weed/remote_storage/s3/s3_storage_client.go

@@ -690,11 +690,14 @@ func (s *s3RemoteStorageClient) CopyFile(src *remote_pb.RemoteStorageLocation, d
 	}
 
 	if size := aws.Int64Value(headOut.ContentLength); size <= s3CopyObjectSizeLimit {
+		// a copy takes the bucket's default encryption unless told otherwise
 		if _, err = s.conn.CopyObject(&s3.CopyObjectInput{
-			Bucket:       aws.String(dst.Bucket),
-			Key:          aws.String(dstKey),
-			CopySource:   copySource,
-			StorageClass: storageClass,
+			Bucket:               aws.String(dst.Bucket),
+			Key:                  aws.String(dstKey),
+			CopySource:           copySource,
+			StorageClass:         storageClass,
+			ServerSideEncryption: headOut.ServerSideEncryption,
+			SSEKMSKeyId:          headOut.SSEKMSKeyId,
 		}); err != nil {
 			return nil, fmt.Errorf("copy s3 %s/%s to %s/%s: %w", src.Bucket, srcKey, dst.Bucket, dstKey, err)
 		}
@@ -744,6 +747,8 @@ func (s *s3RemoteStorageClient) multipartCopy(headOut *s3.HeadObjectOutput, src
 		ContentDisposition:      headOut.ContentDisposition,
 		ContentLanguage:         headOut.ContentLanguage,
 		WebsiteRedirectLocation: headOut.WebsiteRedirectLocation,
+		ServerSideEncryption:    headOut.ServerSideEncryption,
+		SSEKMSKeyId:             headOut.SSEKMSKeyId,
 		StorageClass:            storageClass,
 	}
 	if headOut.Expires != nil {
@@ -781,6 +786,9 @@ func (s *s3RemoteStorageClient) multipartCopy(headOut *s3.HeadObjectOutput, src
 			PartNumber:      aws.Int64(partNumber),
 			CopySource:      copySource,
 			CopySourceRange: aws.String(fmt.Sprintf("bytes=%d-%d", offset, min(offset+partSize, size)-1)),
+			// every part from the object the HeadObject saw, not a mix of
+			// versions if it is overwritten mid-copy
+			CopySourceIfMatch: headOut.ETag,
 		})
 		if partErr == nil && partOut.CopyPartResult == nil {
 			partErr = errors.New("no part result")

weed/remote_storage/s3/s3_storage_client_copy_test.go

@@ -38,6 +38,8 @@ func (m *copyObjectMock) HeadObject(input *awss3.HeadObjectInput) (*awss3.HeadOb
 		ContentType:             aws.String("image/jpeg"),
 		Expires:                 aws.String("Wed, 21 Oct 2026 07:28:00 GMT"),
 		WebsiteRedirectLocation: aws.String("/elsewhere"),
+		ServerSideEncryption:    aws.String("aws:kms"),
+		SSEKMSKeyId:             aws.String("key-1"),
 		ETag:                    aws.String(`"etag-` + aws.StringValue(input.Key) + `"`),
 	}, nil
 }
@@ -100,6 +102,8 @@ func TestS3CopyFileUsesCopyObjectUpToTheLimit(t *testing.T) {
 	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.Equal(t, "aws:kms", aws.StringValue(mock.copyInputs[0].ServerSideEncryption))
+	require.Equal(t, "key-1", aws.StringValue(mock.copyInputs[0].SSEKMSKeyId))
 	require.Nil(t, mock.createInput)
 	// the RemoteEntry describes the destination
 	require.Equal(t, `"etag-dst/a b.jpg"`, remoteEntry.RemoteETag)
@@ -117,6 +121,8 @@ func TestS3CopyFileCopiesLargeObjectsInParts(t *testing.T) {
 	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, "aws:kms", aws.StringValue(mock.createInput.ServerSideEncryption))
+	require.Equal(t, "key-1", aws.StringValue(mock.createInput.SSEKMSKeyId))
 	require.Equal(t, int64(1792567680), aws.TimeValue(mock.createInput.Expires).Unix())
 	// tags only where the remote supports tagging
 	require.Empty(t, mock.tagInputs)
@@ -124,6 +130,9 @@ func TestS3CopyFileCopiesLargeObjectsInParts(t *testing.T) {
 	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))
+	for _, part := range mock.partInputs {
+		require.Equal(t, `"etag-src/a b.jpg"`, aws.StringValue(part.CopySourceIfMatch))
+	}
 	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)

weed/server/filer_grpc_server_rename.go

@@ -150,9 +150,18 @@ type metadataEvent struct {
 	oldEntry     *filer.Entry
 	newEntry     *filer.Entry
 	deleteChunks bool
+	// the old entry's remote object is deleted once the rename has committed:
+	// it, or an object under it, was copied to the new key, and a rename that
+	// fails must not lose it
+	deleteOldRemote bool
 }
 
 func (event metadataEvent) notify(f *filer.Filer, ctx context.Context, signatures []int32) {
+	if event.deleteOldRemote {
+		if err := f.DeleteFromRemote(ctx, event.oldEntry); err != nil {
+			glog.WarningfCtx(ctx, "delete remote object of renamed %s: %v", event.oldEntry.FullPath, err)
+		}
+	}
 	f.NotifyUpdateEvent(ctx, event.oldEntry, event.newEntry, event.deleteChunks, false, signatures)
 }
 
@@ -336,19 +345,33 @@ func (fs *FilerServer) moveSelfEntry(ctx context.Context, stream filer_pb.Seawee
 		})
 	}
 	*metadataEvents = append(*metadataEvents, metadataEvent{
-		oldEntry: sourceEntry,
-		newEntry: newEntry,
+		oldEntry:        sourceEntry,
+		newEntry:        newEntry,
+		deleteOldRemote: copiedRemote != nil,
 	})
+	selfEvent := len(*metadataEvents) - 1
 
 	if moveFolderSubEntries != nil {
 		if moveChildrenErr := moveFolderSubEntries(); moveChildrenErr != nil {
 			return moveChildrenErr
 		}
+		// removing the old directory on the remote removes every object under
+		// it, so it waits for the commit too once a child was copied
+		for _, event := range (*metadataEvents)[selfEvent+1:] {
+			if event.deleteOldRemote {
+				(*metadataEvents)[selfEvent].deleteOldRemote = true
+				break
+			}
+		}
 	}
 
 	// delete old entry
 	ctx = context.WithValue(ctx, "OP", "MV")
-	deleteErr := fs.filer.DeleteEntryMetaAndData(filer.WithSuppressedMetadataEvents(ctx), oldPath, false, false, false, false, signatures, 0)
+	deleteCtx := filer.WithSuppressedMetadataEvents(ctx)
+	if (*metadataEvents)[selfEvent].deleteOldRemote {
+		deleteCtx = filer.WithKeepRemoteObject(deleteCtx)
+	}
+	deleteErr := fs.filer.DeleteEntryMetaAndData(deleteCtx, oldPath, false, false, false, false, signatures, 0)
 	if deleteErr != nil {
 		return deleteErr
 	}

weed/server/filer_grpc_server_rename_remote_test.go

@@ -226,18 +226,26 @@ func TestRenameEmptyRemoteEntryDoesNotCopy(t *testing.T) {
 	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")
+func TestRenameRemoteOnlyEntryFailingAfterTheCopyKeepsTheSource(t *testing.T) {
+	for _, src := range []util.FullPath{"/buckets/b/src/a.jpg", "/buckets/b/src"} {
+		t.Run(string(src), func(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")
+			err := renameFile(server, src, "/buckets/b/moved")
+			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())
+			// the old object, and the old directory holding it, are deleted
+			// only once the rename has committed
+			dst := "origin/moved"
+			if src == "/buckets/b/src" {
+				dst = "origin/moved/a.jpg"
+			}
+			assert.Equal(t, []string{"copy origin/src/a.jpg " + dst}, client.recorded())
+		})
+	}
 }
 
 func TestRenameEntryWithLocalDataDoesNotCopy(t *testing.T) {
@@ -267,11 +275,13 @@ func TestRenameDirectoryCopiesRemoteOnlyChildren(t *testing.T) {
 
 	require.NoError(t, renameFile(server, "/buckets/b/src", "/buckets/b/moved"))
 
+	// b.jpg's old object goes with its entry; the old directory and a.jpg's
+	// old object wait for the commit
 	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",
+		"delete origin/src/a.jpg",
 	}, client.recorded())
 	moved, err := store.FindEntry(context.Background(), "/buckets/b/moved/a.jpg")
 	require.NoError(t, err)