mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-29 08:20:42 +01:00
Alex Russell
eabffd9294
The deferred delete must only run once the lookup shows no remote-only entry at the old path. A failed lookup shows nothing, and an extra object loses nothing.
415 lines
15 KiB
Go
415 lines
15 KiB
Go
package weed_server
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"maps"
|
|
"path/filepath"
|
|
"time"
|
|
|
|
"github.com/seaweedfs/seaweedfs/weed/filer"
|
|
"github.com/seaweedfs/seaweedfs/weed/glog"
|
|
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
|
"github.com/seaweedfs/seaweedfs/weed/util"
|
|
)
|
|
|
|
// acquireRenamePathLocks holds both paths across commit and notification so a
|
|
// fenced lookup cannot read renamed state under a fence preceding its events.
|
|
// Path order avoids deadlock with a reverse rename; descendants are not
|
|
// locked. The returned func releases both.
|
|
func (fs *FilerServer) acquireRenamePathLocks(intention string, oldPath, newPath util.FullPath) func() {
|
|
firstPath, secondPath := oldPath, newPath
|
|
if secondPath < firstPath {
|
|
firstPath, secondPath = secondPath, firstPath
|
|
}
|
|
firstLock := fs.entryLockTable.AcquireLock(intention, firstPath, util.ExclusiveLock)
|
|
if secondPath == firstPath {
|
|
return func() { fs.entryLockTable.ReleaseLock(firstPath, firstLock) }
|
|
}
|
|
secondLock := fs.entryLockTable.AcquireLock(intention, secondPath, util.ExclusiveLock)
|
|
return func() {
|
|
fs.entryLockTable.ReleaseLock(secondPath, secondLock)
|
|
fs.entryLockTable.ReleaseLock(firstPath, firstLock)
|
|
}
|
|
}
|
|
|
|
func (fs *FilerServer) AtomicRenameEntry(ctx context.Context, req *filer_pb.AtomicRenameEntryRequest) (*filer_pb.AtomicRenameEntryResponse, error) {
|
|
|
|
glog.V(1).Infof("AtomicRenameEntry %v", req)
|
|
|
|
oldParent := util.FullPath(filepath.ToSlash(req.OldDirectory))
|
|
newParent := util.FullPath(filepath.ToSlash(req.NewDirectory))
|
|
|
|
if err := fs.filer.CanRename(ctx, oldParent, newParent, req.OldName); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
defer fs.acquireRenamePathLocks("AtomicRenameEntry", oldParent.Child(req.OldName), newParent.Child(req.NewName))()
|
|
|
|
ctx, err := fs.filer.BeginTransaction(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
oldEntry, err := fs.filer.FindEntry(ctx, oldParent.Child(req.OldName))
|
|
if err != nil {
|
|
fs.filer.RollbackTransaction(ctx)
|
|
return nil, fmt.Errorf("%s/%s not found: %v", req.OldDirectory, req.OldName, err)
|
|
}
|
|
|
|
var metadataEvents []metadataEvent
|
|
var pendingChunkDeletes []*filer_pb.FileChunk
|
|
moveErr := fs.moveEntry(ctx, nil, oldParent, oldEntry, newParent, req.NewName, req.Signatures, false, &metadataEvents, &pendingChunkDeletes)
|
|
if moveErr != nil {
|
|
fs.filer.RollbackTransaction(ctx)
|
|
return nil, fmt.Errorf("%s/%s move error: %v", req.OldDirectory, req.OldName, moveErr)
|
|
} else {
|
|
if commitError := fs.filer.CommitTransaction(ctx); commitError != nil {
|
|
fs.filer.RollbackTransaction(ctx)
|
|
return nil, fmt.Errorf("%s/%s move commit error: %v", req.OldDirectory, req.OldName, commitError)
|
|
}
|
|
}
|
|
// Chunks from an overwritten rename target are only deletable after the
|
|
// rename transaction has committed: anything that fails mid-rename (move,
|
|
// child moves, oldPath delete, CommitTransaction) would otherwise leave
|
|
// live metadata pointing at freshly-deleted chunks.
|
|
if len(pendingChunkDeletes) > 0 {
|
|
fs.filer.DeleteChunksNotRecursive(pendingChunkDeletes)
|
|
}
|
|
for _, event := range metadataEvents {
|
|
event.notify(fs.filer, ctx, req.Signatures)
|
|
}
|
|
|
|
return &filer_pb.AtomicRenameEntryResponse{}, nil
|
|
}
|
|
|
|
func (fs *FilerServer) StreamRenameEntry(req *filer_pb.StreamRenameEntryRequest, stream filer_pb.SeaweedFiler_StreamRenameEntryServer) (err error) {
|
|
|
|
glog.V(1).Infof("StreamRenameEntry %v", req)
|
|
|
|
oldParent := util.FullPath(filepath.ToSlash(req.OldDirectory))
|
|
newParent := util.FullPath(filepath.ToSlash(req.NewDirectory))
|
|
|
|
if err := fs.filer.CanRename(stream.Context(), oldParent, newParent, req.OldName); err != nil {
|
|
return err
|
|
}
|
|
|
|
defer fs.acquireRenamePathLocks("StreamRenameEntry", oldParent.Child(req.OldName), newParent.Child(req.NewName))()
|
|
|
|
ctx := context.Background()
|
|
|
|
ctx, err = fs.filer.BeginTransaction(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
oldEntry, err := fs.filer.FindEntry(ctx, oldParent.Child(req.OldName))
|
|
if err != nil {
|
|
fs.filer.RollbackTransaction(ctx)
|
|
return fmt.Errorf("%s/%s not found: %v", req.OldDirectory, req.OldName, err)
|
|
}
|
|
|
|
if oldEntry.IsDirectory() {
|
|
// follow https://pubs.opengroup.org/onlinepubs/000095399/functions/rename.html
|
|
targetDir := newParent.Child(req.NewName)
|
|
newEntry, err := fs.filer.FindEntry(ctx, targetDir)
|
|
if err == nil {
|
|
if !newEntry.IsDirectory() {
|
|
fs.filer.RollbackTransaction(ctx)
|
|
return fmt.Errorf("%s is not directory", targetDir)
|
|
}
|
|
if entries, _, _ := fs.filer.ListDirectoryEntries(context.Background(), targetDir, "", false, 1, "", "", ""); len(entries) > 0 {
|
|
return fmt.Errorf("%s is not empty", targetDir)
|
|
}
|
|
}
|
|
}
|
|
|
|
var metadataEvents []metadataEvent
|
|
var pendingChunkDeletes []*filer_pb.FileChunk
|
|
moveErr := fs.moveEntry(ctx, stream, oldParent, oldEntry, newParent, req.NewName, req.Signatures, false, &metadataEvents, &pendingChunkDeletes)
|
|
if moveErr != nil {
|
|
fs.filer.RollbackTransaction(ctx)
|
|
return fmt.Errorf("%s/%s move error: %v", req.OldDirectory, req.OldName, moveErr)
|
|
} else {
|
|
if commitError := fs.filer.CommitTransaction(ctx); commitError != nil {
|
|
fs.filer.RollbackTransaction(ctx)
|
|
return fmt.Errorf("%s/%s move commit error: %v", req.OldDirectory, req.OldName, commitError)
|
|
}
|
|
}
|
|
if len(pendingChunkDeletes) > 0 {
|
|
fs.filer.DeleteChunksNotRecursive(pendingChunkDeletes)
|
|
}
|
|
for _, event := range metadataEvents {
|
|
event.notify(fs.filer, ctx, req.Signatures)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
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 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 {
|
|
// a child path is not locked by its directory's rename, so another
|
|
// rename may have put a remote-only entry at the old key meanwhile;
|
|
// the store is asked directly, a lazy fetch would revive the old key.
|
|
// When in doubt the object stays: an extra object loses nothing.
|
|
current, findErr := f.Store.FindEntry(ctx, event.oldEntry.FullPath)
|
|
switch {
|
|
case findErr != nil && !errors.Is(findErr, filer_pb.ErrNotFound):
|
|
glog.WarningfCtx(ctx, "keep remote object of renamed %s: look up its old path: %v", event.oldEntry.FullPath, findErr)
|
|
case findErr == nil && current.IsInRemoteOnly():
|
|
glog.V(0).InfofCtx(ctx, "keep remote object of renamed %s: a remote-only entry has taken its path", event.oldEntry.FullPath)
|
|
default:
|
|
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)
|
|
}
|
|
|
|
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 {
|
|
|
|
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 {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}, signatures, skipTargetLookup, metadataEvents, pendingChunkDeletes); 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 {
|
|
|
|
currentDirPath := oldParent.Child(entry.Name())
|
|
newDirPath := newParent.Child(newName)
|
|
|
|
glog.V(1).Infof("moving folder %s => %s", currentDirPath, newDirPath)
|
|
|
|
lastFileName := ""
|
|
includeLastFile := false
|
|
for {
|
|
|
|
entries, hasMore, err := fs.filer.ListDirectoryEntries(ctx, currentDirPath, lastFileName, includeLastFile, 1024, "", "", "")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
// println("found", len(entries), "entries under", currentDirPath)
|
|
|
|
for _, item := range entries {
|
|
lastFileName = item.Name()
|
|
// 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)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if !hasMore {
|
|
break
|
|
}
|
|
}
|
|
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 {
|
|
|
|
oldPath, newPath := oldParent.Child(entry.Name()), newParent.Child(newName)
|
|
|
|
glog.V(1).Infof("moving entry %s => %s", oldPath, newPath)
|
|
|
|
if oldPath == newPath {
|
|
glog.V(1).Infof("skip moving entry %s => %s", oldPath, newPath)
|
|
return nil
|
|
}
|
|
|
|
sourceEntry := entry.ShallowClone()
|
|
sourceEntry.FullPath = oldPath
|
|
|
|
var existingTarget *filer.Entry
|
|
if !skipTargetLookup {
|
|
if targetEntry, findErr := fs.filer.FindEntry(ctx, newPath); findErr == nil {
|
|
existingTarget = targetEntry.ShallowClone()
|
|
} else if findErr != filer_pb.ErrNotFound {
|
|
return findErr
|
|
}
|
|
}
|
|
if existingTarget != nil {
|
|
switch {
|
|
case existingTarget.IsDirectory() && !entry.IsDirectory():
|
|
return fmt.Errorf("%s: %w", existingTarget.FullPath, filer_pb.ErrExistingIsDirectory)
|
|
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(
|
|
targetCtx,
|
|
newPath,
|
|
false,
|
|
false,
|
|
false,
|
|
false,
|
|
signatures,
|
|
0,
|
|
); deleteErr != nil {
|
|
return deleteErr
|
|
}
|
|
}
|
|
|
|
// add to new directory
|
|
newEntry := &filer.Entry{
|
|
FullPath: newPath,
|
|
Attr: entry.Attr,
|
|
Chunks: entry.GetChunks(),
|
|
Extended: entry.Extended,
|
|
Content: entry.Content,
|
|
HardLinkCounter: entry.HardLinkCounter,
|
|
HardLinkId: entry.HardLinkId,
|
|
Remote: remoteEntry,
|
|
Quota: entry.Quota,
|
|
}
|
|
if skipTargetLookup {
|
|
if newEntry.FullPath.IsLongerFileName(fs.filer.MaxFilenameLength) {
|
|
return filer_pb.ErrEntryNameTooLong
|
|
}
|
|
if createErr := fs.filer.Store.InsertEntryKnownAbsent(filer.WithSuppressedMetadataEvents(ctx), newEntry); createErr != nil {
|
|
return fmt.Errorf("insert entry %s: %v", newEntry.FullPath, createErr)
|
|
}
|
|
} else {
|
|
if createErr := fs.filer.CreateEntry(filer.WithSuppressedMetadataEvents(ctx), newEntry, nil, false, false, signatures, false, fs.filer.MaxFilenameLength); createErr != nil {
|
|
return createErr
|
|
}
|
|
}
|
|
if existingTarget != nil {
|
|
toDelete, err := filer.MinusChunks(ctx, fs.filer.MasterClient.GetLookupFileIdFunction(), existingTarget.GetChunks(), newEntry.GetChunks(), fs.filer.MasterClient)
|
|
if err != nil {
|
|
glog.ErrorfCtx(ctx, "Failed to resolve overwrite target chunks during rename. new: %v, old: %v", newEntry.GetChunks(), existingTarget.GetChunks())
|
|
} else if len(toDelete) > 0 {
|
|
// Defer chunk deletion until after CommitTransaction so that a
|
|
// failure in any subsequent step (child moves, oldPath delete,
|
|
// stream send, or the commit itself) leaves the chunks intact for
|
|
// the rolled-back rename.
|
|
*pendingChunkDeletes = append(*pendingChunkDeletes, toDelete...)
|
|
}
|
|
}
|
|
if stream != nil {
|
|
if err := stream.Send(&filer_pb.StreamRenameEntryResponse{
|
|
Directory: string(oldParent),
|
|
EventNotification: &filer_pb.EventNotification{
|
|
OldEntry: &filer_pb.Entry{
|
|
Name: entry.Name(),
|
|
},
|
|
NewEntry: newEntry.ToProtoEntry(),
|
|
DeleteChunks: false,
|
|
NewParentPath: string(newParent),
|
|
IsFromOtherCluster: false,
|
|
Signatures: nil,
|
|
},
|
|
TsNs: time.Now().UnixNano(),
|
|
}); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
if existingTarget != nil {
|
|
*metadataEvents = append(*metadataEvents, metadataEvent{
|
|
oldEntry: existingTarget,
|
|
deleteChunks: true,
|
|
})
|
|
}
|
|
*metadataEvents = append(*metadataEvents, metadataEvent{
|
|
oldEntry: sourceEntry,
|
|
newEntry: newEntry,
|
|
deleteOldRemote: copiedRemote != nil,
|
|
})
|
|
selfEvent := len(*metadataEvents) - 1
|
|
keepOldRemote := copiedRemote != nil
|
|
|
|
if moveFolderSubEntries != nil {
|
|
if moveChildrenErr := moveFolderSubEntries(); moveChildrenErr != nil {
|
|
return moveChildrenErr
|
|
}
|
|
// removing the old directory on the remote removes every object under
|
|
// its prefix, the copied children's old keys included, which must wait
|
|
// for the commit. Those are deleted one by one after it instead, and
|
|
// nothing else under the prefix is touched.
|
|
for _, event := range (*metadataEvents)[selfEvent+1:] {
|
|
if event.deleteOldRemote {
|
|
keepOldRemote = true
|
|
break
|
|
}
|
|
}
|
|
}
|
|
|
|
// delete old entry
|
|
ctx = context.WithValue(ctx, "OP", "MV")
|
|
deleteCtx := filer.WithSuppressedMetadataEvents(ctx)
|
|
if keepOldRemote {
|
|
deleteCtx = filer.WithKeepRemoteObject(deleteCtx)
|
|
}
|
|
deleteErr := fs.filer.DeleteEntryMetaAndData(deleteCtx, oldPath, false, false, false, false, signatures, 0)
|
|
if deleteErr != nil {
|
|
return deleteErr
|
|
}
|
|
if stream != nil {
|
|
if err := stream.Send(&filer_pb.StreamRenameEntryResponse{
|
|
Directory: string(oldParent),
|
|
EventNotification: &filer_pb.EventNotification{
|
|
OldEntry: &filer_pb.Entry{
|
|
Name: entry.Name(),
|
|
},
|
|
NewEntry: nil,
|
|
DeleteChunks: false,
|
|
NewParentPath: "",
|
|
IsFromOtherCluster: false,
|
|
Signatures: nil,
|
|
},
|
|
TsNs: time.Now().UnixNano(),
|
|
}); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
return nil
|
|
|
|
}
|