filer: mirror remote-object cleanup in the replayed delete sweep

Children a lagging peer alone still holds under a replayed directory delete were dropped with plain store deletes: their remote objects leaked and lazy remote fetches could resurrect them. The sweep now records tombstones stamped by the parent event and deletes swept non-directory children through maybeDeleteFromRemoteIfUnchanged, which carries the same stat guard the write-back daemon's processDeleteEvent applies — a delayed replay cannot erase an object another writer replaced. A failed remote delete fails the replay so the event is retried instead of orphaning the object. The directory's own tombstone is recorded by the event pipeline at the event timestamp and covers the whole subtree. Metadata-only deletes and cross-cluster events keep the origin's semantics.

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

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 {

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)
+	}
+}

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)
 			}

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)