diff --git a/weed/command/filer_remote_sync_dir.go b/weed/command/filer_remote_sync_dir.go index 81b5f916c4..1214422b48 100644 --- a/weed/command/filer_remote_sync_dir.go +++ b/weed/command/filer_remote_sync_dir.go @@ -569,10 +569,7 @@ func deleteRenameSource(filerClient filer_pb.FilerClient, client remote_storage. // copy written as one stream can legitimately carry a different ETag from a // multipart original). func describes(stamp, object *filer_pb.RemoteEntry) bool { - if stamp == nil || object == nil || stamp.RemoteSize != object.RemoteSize { - return false - } - return stamp.RemoteETag == "" || object.RemoteETag == "" || stamp.RemoteETag == object.RemoteETag + return remote_storage.Describes(stamp, object) } // openRemoteObject streams the object when the client can, and reads it whole diff --git a/weed/filer/filer_lazy_remote.go b/weed/filer/filer_lazy_remote.go index 036ac5f5dc..2e8bc4d3f8 100644 --- a/weed/filer/filer_lazy_remote.go +++ b/weed/filer/filer_lazy_remote.go @@ -202,13 +202,31 @@ func (f *Filer) retractLazyRemoteEntry(ctx context.Context, entry *Entry) { } } -func (f *Filer) maybeDeleteFromRemote(ctx context.Context, entry *Entry) (bool, error) { - if entry == nil || f.RemoteStorage == nil { - return false, nil - } - +// remoteStorageObject resolves the client and object location entry maps to +// on its mounted remote storage; a nil client means the path is not mounted. +func (f *Filer) remoteStorageObject(ctx context.Context, entry *Entry) (remote_storage.RemoteStorageClient, *remote_pb.RemoteStorageLocation, error) { mountDir, remoteLoc := f.RemoteStorage.FindMountDirectory(entry.FullPath) if remoteLoc == nil { + return nil, nil, nil + } + + remoteConf, found := f.RemoteStorage.GetRemoteStorageConf(remoteLoc.Name) + if !found { + return nil, nil, fmt.Errorf("resolve remote storage client for %s: not found", entry.FullPath) + } + client, clientErr := f.buildRemoteStorageClient(ctx, remoteConf) + if clientErr != nil { + return nil, nil, fmt.Errorf("resolve remote storage client for %s: %w", entry.FullPath, clientErr) + } + if client == nil { + return nil, nil, fmt.Errorf("resolve remote storage client for %s: initialization failed", entry.FullPath) + } + + return client, MapFullPathToRemoteStorageLocation(mountDir, remoteLoc, entry.FullPath), nil +} + +func (f *Filer) maybeDeleteFromRemote(ctx context.Context, entry *Entry) (bool, error) { + if entry == nil || f.RemoteStorage == nil { return false, nil } @@ -216,19 +234,10 @@ func (f *Filer) maybeDeleteFromRemote(ctx context.Context, entry *Entry) (bool, return false, nil } - remoteConf, found := f.RemoteStorage.GetRemoteStorageConf(remoteLoc.Name) - if !found { - return false, fmt.Errorf("resolve remote storage client for %s: not found", entry.FullPath) + client, objectLoc, err := f.remoteStorageObject(ctx, entry) + if err != nil || client == nil { + return false, err } - client, clientErr := f.buildRemoteStorageClient(ctx, remoteConf) - if clientErr != nil { - return false, fmt.Errorf("resolve remote storage client for %s: %w", entry.FullPath, clientErr) - } - if client == nil { - return false, fmt.Errorf("resolve remote storage client for %s: initialization failed", entry.FullPath) - } - - objectLoc := MapFullPathToRemoteStorageLocation(mountDir, remoteLoc, entry.FullPath) if entry.IsDirectory() { if err := client.RemoveDirectory(objectLoc); err != nil { @@ -252,6 +261,35 @@ func (f *Filer) maybeDeleteFromRemote(ctx context.Context, entry *Entry) (bool, return true, nil } +// maybeDeleteFromRemoteIfUnchanged deletes the remote object only when it +// still matches the entry's stamp — the same guard the write-back daemon's +// processDeleteEvent applies, needed wherever a delete can land long after +// another writer replaced the object. +func (f *Filer) maybeDeleteFromRemoteIfUnchanged(ctx context.Context, entry *Entry) error { + if entry == nil || f.RemoteStorage == nil || entry.Remote == nil { + return nil + } + client, objectLoc, err := f.remoteStorageObject(ctx, entry) + if err != nil || client == nil { + return err + } + live, statErr := client.StatFile(objectLoc) + if statErr != nil { + if errors.Is(statErr, remote_storage.ErrRemoteObjectNotFound) { + return nil + } + return fmt.Errorf("stat remote file %s: %w", entry.FullPath, statErr) + } + if !remote_storage.Describes(entry.Remote, live) || entry.Remote.RemoteETag != "" && live.RemoteETag == "" { + glog.V(0).InfofCtx(ctx, "keep %s: the remote object changed since the delete", entry.FullPath) + return nil + } + if err := client.DeleteFile(objectLoc); err != nil && !errors.Is(err, remote_storage.ErrRemoteObjectNotFound) { + return fmt.Errorf("delete remote file %s: %w", entry.FullPath, err) + } + return 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 { diff --git a/weed/filer/meta_aggregator_replay_test.go b/weed/filer/meta_aggregator_replay_test.go index bc4972f182..1bf427b4b9 100644 --- a/weed/filer/meta_aggregator_replay_test.go +++ b/weed/filer/meta_aggregator_replay_test.go @@ -12,6 +12,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/pb/remote_pb" "github.com/seaweedfs/seaweedfs/weed/stats" "github.com/seaweedfs/seaweedfs/weed/util" ) @@ -984,3 +985,228 @@ func TestReplayDirectoryDeleteKeepsSharedHardLinkChunks(t *testing.T) { t.Fatalf("pending deletions = %d, want 0 while a sibling link owns the chunks", got) } } + +// A replayed directory delete must mirror the origin's remote cleanup: entries +// the lagging peer alone still holds lose their remote objects too, and record +// tombstones so lazy remote fetches cannot resurrect them. +func TestReplayDirectoryDeleteSweepsRemoteObjects(t *testing.T) { + const storageType = "stub_replay_remote" + stub := &stubRemoteClient{ + statResult: &filer_pb.RemoteEntry{RemoteMtime: 1700000000, RemoteSize: 10}, + } + defer registerStubMaker(t, storageType, stub)() + + conf := &remote_pb.RemoteConf{Name: "replaystore", Type: storageType} + rs := NewFilerRemoteStorage() + rs.storageNameToConf[conf.Name] = conf + rs.mapDirectoryToRemoteStorage("/buckets/mybucket", &remote_pb.RemoteStorageLocation{ + Name: "replaystore", + Bucket: "mybucket", + Path: "/", + }) + + store := newStubFilerStore() + f := newTestFiler(t, store, rs) + ctx := context.Background() + + put := func(e *Entry) { + if err := store.InsertEntry(ctx, e); err != nil { + t.Fatal(err) + } + } + for _, dir := range []string{"/buckets", "/buckets/mybucket", "/buckets/mybucket/d"} { + put(&Entry{FullPath: util.FullPath(dir), Attr: Attr{Mode: os.ModeDir}}) + } + put(&Entry{ + FullPath: "/buckets/mybucket/d/f.txt", + Attr: Attr{Mtime: time.Now(), Mode: 0644}, + Remote: &filer_pb.RemoteEntry{RemoteMtime: 1700000000, RemoteSize: 10}, + }) + + event := &filer_pb.SubscribeMetadataResponse{ + Directory: "/buckets/mybucket", + EventNotification: &filer_pb.EventNotification{ + OldEntry: &filer_pb.Entry{Name: "d", IsDirectory: true}, + DeleteChunks: false, + }, + TsNs: time.Now().UnixNano(), + } + if err := f.Replay(event); err != nil { + t.Fatalf("replay: %v", err) + } + f.onMetadataChangeEvent(event) + + if _, err := store.FindEntry(ctx, "/buckets/mybucket/d/f.txt"); !errors.Is(err, filer_pb.ErrNotFound) { + t.Fatalf("swept child should be gone, got %v", err) + } + if len(stub.deleteCalls) != 1 || stub.deleteCalls[0].Path != "/d/f.txt" { + t.Fatalf("expected one remote delete of /d/f.txt, got %v", stub.deleteCalls) + } + if got := f.remoteTombstones.dirs["/buckets/mybucket/d"]; got != event.TsNs { + t.Fatalf("dir tombstone stamped %d, want the event ts %d", got, event.TsNs) + } +} + +// A metadata-only delete keeps remote objects: the event's marker means the +// remote tree stays authoritative, so the sweep must not touch it. +func TestReplayDirectoryDeleteKeepRemoteObjects(t *testing.T) { + const storageType = "stub_replay_remote_keep" + stub := &stubRemoteClient{} + defer registerStubMaker(t, storageType, stub)() + + conf := &remote_pb.RemoteConf{Name: "keepstore", Type: storageType} + rs := NewFilerRemoteStorage() + rs.storageNameToConf[conf.Name] = conf + rs.mapDirectoryToRemoteStorage("/buckets/mybucket", &remote_pb.RemoteStorageLocation{ + Name: "keepstore", + Bucket: "mybucket", + Path: "/", + }) + + store := newStubFilerStore() + f := newTestFiler(t, store, rs) + ctx := context.Background() + + put := func(e *Entry) { + if err := store.InsertEntry(ctx, e); err != nil { + t.Fatal(err) + } + } + for _, dir := range []string{"/buckets", "/buckets/mybucket", "/buckets/mybucket/d"} { + put(&Entry{FullPath: util.FullPath(dir), Attr: Attr{Mode: os.ModeDir}}) + } + put(&Entry{ + FullPath: "/buckets/mybucket/d/f.txt", + Attr: Attr{Mtime: time.Now(), Mode: 0644}, + Remote: &filer_pb.RemoteEntry{RemoteMtime: 1700000000, RemoteSize: 10}, + }) + + event := &filer_pb.SubscribeMetadataResponse{ + Directory: "/buckets/mybucket", + EventNotification: &filer_pb.EventNotification{ + OldEntry: &filer_pb.Entry{ + Name: "d", + IsDirectory: true, + Extended: map[string][]byte{ExtKeepRemoteObjectKey: []byte("true")}, + }, + }, + TsNs: time.Now().UnixNano(), + } + if err := f.Replay(event); err != nil { + t.Fatalf("replay: %v", err) + } + f.onMetadataChangeEvent(event) + + if len(stub.deleteCalls) != 0 { + t.Fatalf("metadata-only delete must not touch remote objects, got %v", stub.deleteCalls) + } + if len(f.remoteTombstones.files)+len(f.remoteTombstones.dirs) != 0 { + t.Fatalf("metadata-only delete must not record tombstones, got %v", f.remoteTombstones.dirs) + } +} + +// A delayed sweep must not erase an object another writer replaced since the +// entry's stamp was recorded — the same guard the write-back daemon applies. +func TestReplayDirectoryDeleteKeepsReplacedRemoteObject(t *testing.T) { + const storageType = "stub_replay_remote_replaced" + stub := &stubRemoteClient{ + statResult: &filer_pb.RemoteEntry{RemoteMtime: 1700000099, RemoteSize: 999}, + } + defer registerStubMaker(t, storageType, stub)() + + conf := &remote_pb.RemoteConf{Name: "replacestore", Type: storageType} + rs := NewFilerRemoteStorage() + rs.storageNameToConf[conf.Name] = conf + rs.mapDirectoryToRemoteStorage("/buckets/mybucket", &remote_pb.RemoteStorageLocation{ + Name: "replacestore", + Bucket: "mybucket", + Path: "/", + }) + + store := newStubFilerStore() + f := newTestFiler(t, store, rs) + ctx := context.Background() + + put := func(e *Entry) { + if err := store.InsertEntry(ctx, e); err != nil { + t.Fatal(err) + } + } + for _, dir := range []string{"/buckets", "/buckets/mybucket", "/buckets/mybucket/d"} { + put(&Entry{FullPath: util.FullPath(dir), Attr: Attr{Mode: os.ModeDir}}) + } + put(&Entry{ + FullPath: "/buckets/mybucket/d/f.txt", + Attr: Attr{Mtime: time.Now(), Mode: 0644}, + Remote: &filer_pb.RemoteEntry{RemoteMtime: 1700000000, RemoteSize: 10}, + }) + + event := &filer_pb.SubscribeMetadataResponse{ + Directory: "/buckets/mybucket", + EventNotification: &filer_pb.EventNotification{ + OldEntry: &filer_pb.Entry{Name: "d", IsDirectory: true}, + DeleteChunks: false, + }, + TsNs: time.Now().UnixNano(), + } + if err := f.Replay(event); err != nil { + t.Fatalf("replay: %v", err) + } + + if len(stub.deleteCalls) != 0 { + t.Fatalf("a replaced remote object must be kept, got %v", stub.deleteCalls) + } +} + +// A failed remote delete aborts the sweep so the replication stream retries +// instead of orphaning the object no other event will ever name. +func TestReplayDirectoryDeleteFailsOnRemoteDeleteError(t *testing.T) { + const storageType = "stub_replay_remote_err" + stub := &stubRemoteClient{ + statResult: &filer_pb.RemoteEntry{RemoteMtime: 1700000000, RemoteSize: 10}, + deleteErr: errors.New("remote unavailable"), + } + defer registerStubMaker(t, storageType, stub)() + + conf := &remote_pb.RemoteConf{Name: "errstore", Type: storageType} + rs := NewFilerRemoteStorage() + rs.storageNameToConf[conf.Name] = conf + rs.mapDirectoryToRemoteStorage("/buckets/mybucket", &remote_pb.RemoteStorageLocation{ + Name: "errstore", + Bucket: "mybucket", + Path: "/", + }) + + store := newStubFilerStore() + f := newTestFiler(t, store, rs) + ctx := context.Background() + + put := func(e *Entry) { + if err := store.InsertEntry(ctx, e); err != nil { + t.Fatal(err) + } + } + for _, dir := range []string{"/buckets", "/buckets/mybucket", "/buckets/mybucket/d"} { + put(&Entry{FullPath: util.FullPath(dir), Attr: Attr{Mode: os.ModeDir}}) + } + put(&Entry{ + FullPath: "/buckets/mybucket/d/f.txt", + Attr: Attr{Mtime: time.Now(), Mode: 0644}, + Remote: &filer_pb.RemoteEntry{RemoteMtime: 1700000000, RemoteSize: 10}, + }) + + event := &filer_pb.SubscribeMetadataResponse{ + Directory: "/buckets/mybucket", + EventNotification: &filer_pb.EventNotification{ + OldEntry: &filer_pb.Entry{Name: "d", IsDirectory: true}, + DeleteChunks: false, + }, + TsNs: time.Now().UnixNano(), + } + if err := f.Replay(event); err == nil { + t.Fatal("a failed remote delete must fail the replay for retry") + } + if _, err := store.FindEntry(ctx, "/buckets/mybucket/d/f.txt"); err != nil { + t.Fatalf("the child must survive for the retry, got %v", err) + } +} diff --git a/weed/filer/meta_replay.go b/weed/filer/meta_replay.go index dbfd81fdec..3e359acf10 100644 --- a/weed/filer/meta_replay.go +++ b/weed/filer/meta_replay.go @@ -85,7 +85,7 @@ func (f *Filer) Replay(resp *filer_pb.SubscribeMetadataResponse) error { keepDir := false if message.OldEntry.IsDirectory { var err error - keepDir, err = f.deleteReplayedFolderChildren(ctx, oldPath, message.DeleteChunks, resp.TsNs) + keepDir, err = f.deleteReplayedFolderChildren(ctx, oldPath, message, resp.TsNs) if err != nil { return err } @@ -125,9 +125,9 @@ func (f *Filer) Replay(resp *filer_pb.SubscribeMetadataResponse) error { // the chunks of removed files for deletion. Children this filer committed // after the event stay — a newer local write wins over a delayed delete — // along with the directories that hold them. -func (f *Filer) deleteReplayedFolderChildren(ctx context.Context, dir util.FullPath, deleteChunks bool, tsNs int64) (bool, error) { +func (f *Filer) deleteReplayedFolderChildren(ctx context.Context, dir util.FullPath, message *filer_pb.EventNotification, tsNs int64) (bool, error) { sweep := &folderSweep{chunksByPath: make(map[util.FullPath][]*filer_pb.FileChunk)} - kept, err := f.sweepReplayedFolder(ctx, dir, deleteChunks, tsNs, sweep) + kept, err := f.sweepReplayedFolder(ctx, dir, message, tsNs, sweep) // Flush what was collected even on error: the entries already deleted // cannot be listed again on retry, so their chunks must be queued now. sweep.flush(ctx, f) @@ -164,7 +164,10 @@ func (s *folderSweep) flush(ctx context.Context, f *Filer) { // newer than the event, working subtree-first so nested entries go too. // It reports whether a newer descendant keeps dir alive — the caller then // keeps the dir entry itself. -func (f *Filer) sweepReplayedFolder(ctx context.Context, dir util.FullPath, deleteChunks bool, tsNs int64, sweep *folderSweep) (kept bool, err error) { +func (f *Filer) sweepReplayedFolder(ctx context.Context, dir util.FullPath, message *filer_pb.EventNotification, tsNs int64, sweep *folderSweep) (kept bool, err error) { + deleteChunks := message.DeleteChunks + keepRemote := IsMetadataOnlyDelete(message.OldEntry) + deleteRemote := !keepRemote && !message.GetIsFromOtherCluster() lastFileName := "" for { entries, _, lerr := f.ListDirectoryEntries(ctx, dir, lastFileName, false, PaginationSize, "", "", "") @@ -182,7 +185,7 @@ func (f *Filer) sweepReplayedFolder(ctx context.Context, dir util.FullPath, dele childKept = localTsNs > tsNs } if !childKept && sub.IsDirectory() { - childKept, err = f.sweepReplayedFolder(ctx, sub.FullPath, deleteChunks, tsNs, sweep) + childKept, err = f.sweepReplayedFolder(ctx, sub.FullPath, message, tsNs, sweep) if err != nil { return false, err } @@ -191,6 +194,14 @@ func (f *Filer) sweepReplayedFolder(ctx context.Context, dir util.FullPath, dele kept = true continue } + if !keepRemote { + f.noteRemoteDeletion(sub.FullPath, sub.IsDirectory(), tsNs) + } + if deleteRemote && !sub.IsDirectory() { + if remoteErr := f.maybeDeleteFromRemoteIfUnchanged(ctx, sub); remoteErr != nil { + return false, fmt.Errorf("remote delete %s: %w", sub.FullPath, remoteErr) + } + } if derr := f.Store.DeleteEntry(ctx, sub.FullPath); derr != nil { return false, fmt.Errorf("delete %s: %w", sub.FullPath, derr) } diff --git a/weed/remote_storage/remote_storage.go b/weed/remote_storage/remote_storage.go index 0cefe362a3..705866d57b 100644 --- a/weed/remote_storage/remote_storage.go +++ b/weed/remote_storage/remote_storage.go @@ -26,6 +26,15 @@ func ParseLocationName(remote string) (locationName string) { return } +// Describes reports whether a stored remote entry's stamp still describes the +// live remote object: same size, and equal ETags when both are known. +func Describes(stamp, object *filer_pb.RemoteEntry) bool { + if stamp == nil || object == nil || stamp.RemoteSize != object.RemoteSize { + return false + } + return stamp.RemoteETag == "" || object.RemoteETag == "" || stamp.RemoteETag == object.RemoteETag +} + func parseBucketLocation(remote string) (loc *remote_pb.RemoteStorageLocation) { loc = &remote_pb.RemoteStorageLocation{} remote = strings.TrimSuffix(remote, slash)