mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-13 17:48:29 +02:00
filer: copy over a remote-only rename target only after the commit
Copying the source object onto the target's key before the transaction committed destroyed the target's only content, which a rolled-back commit cannot restore. When the target is remote-only the copy now runs after the commit: the new entry keeps the old key's stamp until the copy lands and re-stamps it, the target's object is kept meanwhile, and a failed copy keeps the old object so remote.sync can finish it. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
1 parent
eabffd9294
commit
912b027e6f
3 files changed
+132
-51
No files matched your search
@@ -243,33 +243,10 @@ func (f *Filer) DeleteFromRemote(ctx context.Context, entry *Entry) error {
|
||||
// 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
|
||||
src, dst, copier, err := f.RemoteCopyLocations(ctx, entry, oldPath, newPath)
|
||||
if err != nil || copier == nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
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)
|
||||
@@ -277,3 +254,38 @@ func (f *Filer) CopyRemoteOnlyEntry(ctx context.Context, entry *Entry, oldPath,
|
||||
glog.V(3).InfofCtx(ctx, "CopyRemoteOnlyEntry: copied %s to %s", remote_storage.FormatLocation(src), remote_storage.FormatLocation(dst))
|
||||
return remoteEntry, nil
|
||||
}
|
||||
|
||||
// RemoteCopyLocations resolves the object locations a remote-only rename copy
|
||||
// of entry between oldPath and newPath would use, and the client able to copy
|
||||
// it. A nil copier means the entry needs no copy; an error means it needs one
|
||||
// that cannot be served, which a rename must refuse before committing.
|
||||
func (f *Filer) RemoteCopyLocations(ctx context.Context, entry *Entry, oldPath, newPath util.FullPath) (src, dst *remote_pb.RemoteStorageLocation, copier remote_storage.RemoteStorageObjectCopier, err error) {
|
||||
if f.RemoteStorage == nil || entry.IsDirectory() || !entry.IsInRemoteOnly() || len(entry.Content) > 0 {
|
||||
return nil, nil, nil, nil
|
||||
}
|
||||
|
||||
mountDir, remoteLoc := f.RemoteStorage.FindMountDirectory(oldPath)
|
||||
if remoteLoc == nil {
|
||||
return nil, nil, nil, nil
|
||||
}
|
||||
if newMountDir, _ := f.RemoteStorage.FindMountDirectory(newPath); newMountDir != mountDir {
|
||||
return nil, nil, 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, nil, nil, fmt.Errorf("resolve remote storage client for %s: not found", oldPath)
|
||||
}
|
||||
client, clientErr := f.buildRemoteStorageClient(ctx, remoteConf)
|
||||
if clientErr != nil {
|
||||
return nil, nil, nil, fmt.Errorf("resolve remote storage client for %s: %w", oldPath, clientErr)
|
||||
}
|
||||
copier, ok := client.(remote_storage.RemoteStorageObjectCopier)
|
||||
if !ok {
|
||||
return nil, nil, 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)
|
||||
return src, dst, copier, nil
|
||||
}
|
||||
@@ -60,7 +60,8 @@ func (fs *FilerServer) AtomicRenameEntry(ctx context.Context, req *filer_pb.Atom
|
||||
|
||||
var metadataEvents []metadataEvent
|
||||
var pendingChunkDeletes []*filer_pb.FileChunk
|
||||
moveErr := fs.moveEntry(ctx, nil, oldParent, oldEntry, newParent, req.NewName, req.Signatures, false, &metadataEvents, &pendingChunkDeletes)
|
||||
var pendingRemoteCopies []*pendingRemoteCopy
|
||||
moveErr := fs.moveEntry(ctx, nil, oldParent, oldEntry, newParent, req.NewName, req.Signatures, false, &metadataEvents, &pendingChunkDeletes, &pendingRemoteCopies)
|
||||
if moveErr != nil {
|
||||
fs.filer.RollbackTransaction(ctx)
|
||||
return nil, fmt.Errorf("%s/%s move error: %v", req.OldDirectory, req.OldName, moveErr)
|
||||
@@ -77,7 +78,14 @@ func (fs *FilerServer) AtomicRenameEntry(ctx context.Context, req *filer_pb.Atom
|
||||
if len(pendingChunkDeletes) > 0 {
|
||||
fs.filer.DeleteChunksNotRecursive(pendingChunkDeletes)
|
||||
}
|
||||
for _, event := range metadataEvents {
|
||||
fs.finishRemoteCopies(ctx, pendingRemoteCopies)
|
||||
for i := range metadataEvents {
|
||||
event := &metadataEvents[i]
|
||||
for _, pending := range pendingRemoteCopies {
|
||||
if pending.failed && event.oldEntry.FullPath == pending.oldPath {
|
||||
event.deleteOldRemote = false
|
||||
}
|
||||
}
|
||||
event.notify(fs.filer, ctx, req.Signatures)
|
||||
}
|
||||
|
||||
@@ -127,7 +135,8 @@ func (fs *FilerServer) StreamRenameEntry(req *filer_pb.StreamRenameEntryRequest,
|
||||
|
||||
var metadataEvents []metadataEvent
|
||||
var pendingChunkDeletes []*filer_pb.FileChunk
|
||||
moveErr := fs.moveEntry(ctx, stream, oldParent, oldEntry, newParent, req.NewName, req.Signatures, false, &metadataEvents, &pendingChunkDeletes)
|
||||
var pendingRemoteCopies []*pendingRemoteCopy
|
||||
moveErr := fs.moveEntry(ctx, stream, oldParent, oldEntry, newParent, req.NewName, req.Signatures, false, &metadataEvents, &pendingChunkDeletes, &pendingRemoteCopies)
|
||||
if moveErr != nil {
|
||||
fs.filer.RollbackTransaction(ctx)
|
||||
return fmt.Errorf("%s/%s move error: %v", req.OldDirectory, req.OldName, moveErr)
|
||||
@@ -140,7 +149,14 @@ func (fs *FilerServer) StreamRenameEntry(req *filer_pb.StreamRenameEntryRequest,
|
||||
if len(pendingChunkDeletes) > 0 {
|
||||
fs.filer.DeleteChunksNotRecursive(pendingChunkDeletes)
|
||||
}
|
||||
for _, event := range metadataEvents {
|
||||
fs.finishRemoteCopies(ctx, pendingRemoteCopies)
|
||||
for i := range metadataEvents {
|
||||
event := &metadataEvents[i]
|
||||
for _, pending := range pendingRemoteCopies {
|
||||
if pending.failed && event.oldEntry.FullPath == pending.oldPath {
|
||||
event.deleteOldRemote = false
|
||||
}
|
||||
}
|
||||
event.notify(fs.filer, ctx, req.Signatures)
|
||||
}
|
||||
|
||||
@@ -156,6 +172,42 @@ type metadataEvent struct {
|
||||
deleteOldRemote bool
|
||||
}
|
||||
|
||||
// pendingRemoteCopy is a remote-only rename over a remote-only target whose
|
||||
// copy runs after the commit: the store transaction cannot roll a remote
|
||||
// write back, so the new entry keeps the old key's stamp until the copy lands.
|
||||
type pendingRemoteCopy struct {
|
||||
source *filer.Entry
|
||||
oldPath util.FullPath
|
||||
newPath util.FullPath
|
||||
failed bool
|
||||
}
|
||||
|
||||
// finishRemoteCopies runs the deferred remote copies after the commit. A
|
||||
// failure leaves the new entry pointing at the old key, whose object the
|
||||
// caller then keeps: remote.sync completes the copy the same way it does a
|
||||
// rename whose event it finds unfinished.
|
||||
func (fs *FilerServer) finishRemoteCopies(ctx context.Context, pendings []*pendingRemoteCopy) {
|
||||
for _, pending := range pendings {
|
||||
remoteEntry, err := fs.filer.CopyRemoteOnlyEntry(ctx, pending.source, pending.oldPath, pending.newPath)
|
||||
if err != nil {
|
||||
glog.WarningfCtx(ctx, "remote rename copy of %s: %v", pending.newPath, err)
|
||||
pending.failed = true
|
||||
continue
|
||||
}
|
||||
entry, findErr := fs.filer.FindEntry(ctx, pending.newPath)
|
||||
if findErr != nil {
|
||||
glog.WarningfCtx(ctx, "stamp remote rename copy of %s: %v", pending.newPath, findErr)
|
||||
pending.failed = true
|
||||
continue
|
||||
}
|
||||
entry.Remote = remoteEntry
|
||||
if updateErr := fs.filer.UpdateEntry(ctx, entry, entry, false); updateErr != nil {
|
||||
glog.WarningfCtx(ctx, "stamp remote rename copy of %s: %v", pending.newPath, updateErr)
|
||||
pending.failed = true
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (event metadataEvent) notify(f *filer.Filer, ctx context.Context, signatures []int32) {
|
||||
if event.deleteOldRemote {
|
||||
// a child path is not locked by its directory's rename, so another
|
||||
@@ -177,23 +229,23 @@ func (event metadataEvent) notify(f *filer.Filer, ctx context.Context, signature
|
||||
f.NotifyUpdateEvent(ctx, event.oldEntry, event.newEntry, event.deleteChunks, false, signatures)
|
||||
}
|
||||
|
||||
func (fs *FilerServer) moveEntry(ctx context.Context, stream filer_pb.SeaweedFiler_StreamRenameEntryServer, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string, signatures []int32, skipTargetLookup bool, metadataEvents *[]metadataEvent, pendingChunkDeletes *[]*filer_pb.FileChunk) error {
|
||||
func (fs *FilerServer) moveEntry(ctx context.Context, stream filer_pb.SeaweedFiler_StreamRenameEntryServer, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string, signatures []int32, skipTargetLookup bool, metadataEvents *[]metadataEvent, pendingChunkDeletes *[]*filer_pb.FileChunk, pendingRemoteCopies *[]*pendingRemoteCopy) error {
|
||||
|
||||
if err := fs.moveSelfEntry(ctx, stream, oldParent, entry, newParent, newName, func() error {
|
||||
if entry.IsDirectory() {
|
||||
if err := fs.moveFolderSubEntries(ctx, stream, oldParent, entry, newParent, newName, signatures, metadataEvents, pendingChunkDeletes); err != nil {
|
||||
if err := fs.moveFolderSubEntries(ctx, stream, oldParent, entry, newParent, newName, signatures, metadataEvents, pendingChunkDeletes, pendingRemoteCopies); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}, signatures, skipTargetLookup, metadataEvents, pendingChunkDeletes); err != nil {
|
||||
}, signatures, skipTargetLookup, metadataEvents, pendingChunkDeletes, pendingRemoteCopies); err != nil {
|
||||
return fmt.Errorf("fail to move %s => %s: %v", oldParent.Child(entry.Name()), newParent.Child(newName), err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (fs *FilerServer) moveFolderSubEntries(ctx context.Context, stream filer_pb.SeaweedFiler_StreamRenameEntryServer, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string, signatures []int32, metadataEvents *[]metadataEvent, pendingChunkDeletes *[]*filer_pb.FileChunk) error {
|
||||
func (fs *FilerServer) moveFolderSubEntries(ctx context.Context, stream filer_pb.SeaweedFiler_StreamRenameEntryServer, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string, signatures []int32, metadataEvents *[]metadataEvent, pendingChunkDeletes *[]*filer_pb.FileChunk, pendingRemoteCopies *[]*pendingRemoteCopy) error {
|
||||
|
||||
currentDirPath := oldParent.Child(entry.Name())
|
||||
newDirPath := newParent.Child(newName)
|
||||
@@ -216,7 +268,7 @@ func (fs *FilerServer) moveFolderSubEntries(ctx context.Context, stream filer_pb
|
||||
// println("processing", lastFileName)
|
||||
newChildPath := newDirPath.Child(item.Name())
|
||||
skipTarget := fs.filer.Store.SameActualStore(newDirPath, newChildPath)
|
||||
err := fs.moveEntry(ctx, stream, currentDirPath, item, newDirPath, item.Name(), signatures, skipTarget, metadataEvents, pendingChunkDeletes)
|
||||
err := fs.moveEntry(ctx, stream, currentDirPath, item, newDirPath, item.Name(), signatures, skipTarget, metadataEvents, pendingChunkDeletes, pendingRemoteCopies)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -228,7 +280,7 @@ func (fs *FilerServer) moveFolderSubEntries(ctx context.Context, stream filer_pb
|
||||
return nil
|
||||
}
|
||||
|
||||
func (fs *FilerServer) moveSelfEntry(ctx context.Context, stream filer_pb.SeaweedFiler_StreamRenameEntryServer, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string, moveFolderSubEntries func() error, signatures []int32, skipTargetLookup bool, metadataEvents *[]metadataEvent, pendingChunkDeletes *[]*filer_pb.FileChunk) error {
|
||||
func (fs *FilerServer) moveSelfEntry(ctx context.Context, stream filer_pb.SeaweedFiler_StreamRenameEntryServer, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string, moveFolderSubEntries func() error, signatures []int32, skipTargetLookup bool, metadataEvents *[]metadataEvent, pendingChunkDeletes *[]*filer_pb.FileChunk, pendingRemoteCopies *[]*pendingRemoteCopy) error {
|
||||
|
||||
oldPath, newPath := oldParent.Child(entry.Name()), newParent.Child(newName)
|
||||
|
||||
@@ -260,10 +312,26 @@ func (fs *FilerServer) moveSelfEntry(ctx context.Context, stream filer_pb.Seawee
|
||||
}
|
||||
|
||||
// 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
|
||||
// deleting the old entry deletes the old object. Over a remote-only
|
||||
// target the copy runs after the commit instead — it overwrites the
|
||||
// target's only content, which a rolled-back commit cannot restore —
|
||||
// and the new entry reads the old key until the copy stamps it.
|
||||
deferredCopy := false
|
||||
var copiedRemote *filer_pb.RemoteEntry
|
||||
if existingTarget != nil && existingTarget.IsInRemoteOnly() && entry.IsInRemoteOnly() {
|
||||
if _, _, copier, copyErr := fs.filer.RemoteCopyLocations(ctx, entry, oldPath, newPath); copyErr != nil {
|
||||
return copyErr
|
||||
} else if copier != nil {
|
||||
deferredCopy = true
|
||||
*pendingRemoteCopies = append(*pendingRemoteCopies, &pendingRemoteCopy{source: sourceEntry, oldPath: oldPath, newPath: newPath})
|
||||
}
|
||||
}
|
||||
if !deferredCopy {
|
||||
var copyErr error
|
||||
copiedRemote, copyErr = fs.filer.CopyRemoteOnlyEntry(ctx, entry, oldPath, newPath)
|
||||
if copyErr != nil {
|
||||
return copyErr
|
||||
}
|
||||
}
|
||||
remoteEntry := entry.Remote
|
||||
if copiedRemote != nil {
|
||||
@@ -272,8 +340,8 @@ func (fs *FilerServer) moveSelfEntry(ctx context.Context, stream filer_pb.Seawee
|
||||
|
||||
if existingTarget != nil {
|
||||
targetCtx := filer.WithSuppressedMetadataEvents(ctx)
|
||||
if copiedRemote != nil {
|
||||
// the copy has replaced the target's object; keep it
|
||||
if copiedRemote != nil || deferredCopy {
|
||||
// the copy has replaced or still needs the target's object; keep it
|
||||
targetCtx = filer.WithKeepRemoteObject(targetCtx)
|
||||
existingTarget.Extended = maps.Clone(existingTarget.Extended)
|
||||
if existingTarget.Extended == nil {
|
||||
@@ -359,10 +427,10 @@ func (fs *FilerServer) moveSelfEntry(ctx context.Context, stream filer_pb.Seawee
|
||||
*metadataEvents = append(*metadataEvents, metadataEvent{
|
||||
oldEntry: sourceEntry,
|
||||
newEntry: newEntry,
|
||||
deleteOldRemote: copiedRemote != nil,
|
||||
deleteOldRemote: copiedRemote != nil || deferredCopy,
|
||||
})
|
||||
selfEvent := len(*metadataEvents) - 1
|
||||
keepOldRemote := copiedRemote != nil
|
||||
keepOldRemote := copiedRemote != nil || deferredCopy
|
||||
|
||||
if moveFolderSubEntries != nil {
|
||||
if moveChildrenErr := moveFolderSubEntries(); moveChildrenErr != nil {
|
||||
|
||||
@@ -153,22 +153,23 @@ func TestRenameRemoteOnlyEntryOverTargetKeepsTheCopy(t *testing.T) {
|
||||
assert.Equal(t, `"copy"`, dst.Remote.RemoteETag)
|
||||
}
|
||||
|
||||
func TestRenameRemoteOnlyEntryCopyFailureChangesNothing(t *testing.T) {
|
||||
func TestRenameRemoteOnlyEntryCopyFailureKeepsTheOldObject(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")
|
||||
require.NoError(t, renameFile(server, "/buckets/b/src/a.jpg", "/buckets/b/dst/a.jpg"))
|
||||
|
||||
// the copy runs after the commit over a remote-only target, so its
|
||||
// failure cannot roll anything back: the entry still stamps the old
|
||||
// object, which is kept for remote.sync to finish the copy
|
||||
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)
|
||||
_, 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, uint64(202), dst.Attr.Inode)
|
||||
assert.Equal(t, `"orig"`, dst.Remote.RemoteETag)
|
||||
}
|
||||
|
||||
func TestRenameRemoteOnlyEntryWithoutCopierIsRefused(t *testing.T) {
|
||||
|
||||
Reference in new issue
Block a user