remote.sync: release a pending rename's source, the filer leaves it

The filer deleted a pending entry's source after copying it on a later
rename, but that rename locks only its own paths: another rename can
copy onto the source key before its entry is visible, and the cleanup
then deleted that copy. And deleting a pending entry never released its
source at all.

The filer now leaves marked sources alone. remote.sync releases one
when the pending entry is deleted, or when a later rename was copied
right away, and only while the live object is still the one the
entry's stamp describes and no remote-only entry holds the path: a
copy that landed there since has a different ETag and is kept.

weed/command/filer_remote_sync_dir.go

@@ -236,6 +236,12 @@ 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 {
+		// a deleted entry whose rename copy was still pending read its
+		// content from the source its marker names, which nothing else
+		// releases
+		if err := releasePendingSource(filerSource, client, mountedDir, remoteStorageMountLocation, message.OldEntry); err != nil {
+			return err
+		}
 		// The object at the key may have been replaced since the delete: a
 		// rename copies another entry's content onto it, a rewrite uploads
 		// over it, and a copy can land before its entry is visible. Delete
@@ -267,6 +273,41 @@ func processDeleteEvent(
 	return deleteRemoteFile(client, dest)
 }
 
+// releasePendingSource deletes the source a pending-rename marker on entry
+// names, once nothing needs it. The filer never deletes a marked source: a
+// concurrent rename can copy onto that key before its entry is visible. So
+// the object goes only while it is still the one entry's stamp describes, and
+// no remote-only entry holds its path.
+func releasePendingSource(filerSource filer_pb.FilerClient, client remote_storage.RemoteStorageClient, mountedDir string, remoteStorageMountLocation *remote_pb.RemoteStorageLocation, entry *filer_pb.Entry) error {
+	src := util.FullPath(entry.GetExtended()[filer.ExtRemoteRenameSourceKey])
+	if src == "" || !strings.HasPrefix(string(src), strings.TrimSuffix(mountedDir, "/")+"/") {
+		return nil
+	}
+	srcDest := toRemoteStorageLocation(util.FullPath(mountedDir), src, remoteStorageMountLocation)
+	srcDir, srcName := src.DirAndName()
+	current, err := currentEntry(filerSource, srcDir, srcName)
+	if err != nil {
+		return err
+	}
+	if isRemoteOnly(current) {
+		glog.V(0).Infof("keep %s: a remote-only entry holds the pending rename's source", remote_storage.FormatLocation(srcDest))
+		return nil
+	}
+	live, err := client.StatFile(srcDest)
+	if errors.Is(err, remote_storage.ErrRemoteObjectNotFound) {
+		return nil
+	}
+	if err != nil {
+		return err
+	}
+	if stamp := entry.GetRemoteEntry(); !describes(stamp, live) || stamp.RemoteETag != "" && live.RemoteETag == "" {
+		glog.V(0).Infof("keep %s: the pending rename's source changed since", remote_storage.FormatLocation(srcDest))
+		return nil
+	}
+	glog.V(0).Infof("delete %s, the source of a pending rename", remote_storage.FormatLocation(srcDest))
+	return deleteRemoteFile(client, srcDest)
+}
+
 // deleteRemoteFile deletes the object and treats an already-absent object as
 // deleted.
 func deleteRemoteFile(client remote_storage.RemoteStorageClient, dest *remote_pb.RemoteStorageLocation) error {
@@ -340,7 +381,16 @@ func processUpdateEvent(
 			return err
 		}
 		if isRemoteOnly(current) {
-			return completeRemoteOnlyRename(filerClient, client, message.NewParentPath, current, sourceOf(current, oldDest), dest, storageClass)
+			source := sourceOf(current, oldDest)
+			if err := completeRemoteOnlyRename(filerClient, client, message.NewParentPath, current, source, dest, storageClass); err != nil {
+				return err
+			}
+			// a pending entry the filer copied on this rename leaves the
+			// source its marker named
+			if pending := sourceOf(message.OldEntry, nil); pending != nil && !proto.Equal(pending, source) {
+				return releasePendingSource(filerSource, client, mountedDir, remoteStorageMountLocation, message.OldEntry)
+			}
+			return nil
 		}
 		if current == nil && len(message.NewEntry.Extended[filer.ExtRemoteRenameSourceKey]) > 0 {
 			// the rename's copy was pending and the entry has moved on: a

weed/command/filer_remote_sync_dir_test.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"slices"
 	"strings"
 	"testing"
 	"time"
@@ -1508,3 +1509,76 @@ func TestRenameOfAPendingEntryFollowsItsSource(t *testing.T) {
 		}
 	})
 }
+
+func TestPendingRenameSourceIsReleasedOnlyWhileUnchanged(t *testing.T) {
+	const mountedDir = "/buckets"
+	mountLoc := &remote_pb.RemoteStorageLocation{Name: "b2", Bucket: "bucket", Path: "/"}
+	// b.jpg's copy failed after it replaced a target: it read a.jpg's object
+	pendingB := func() *filer_pb.Entry {
+		return &filer_pb.Entry{
+			Name:        "b.jpg",
+			RemoteEntry: &filer_pb.RemoteEntry{StorageName: "b2", RemoteETag: `"a"`, RemoteSize: 6},
+			Extended:    map[string][]byte{filer.ExtRemoteRenameSourceKey: []byte("/buckets/b/a.jpg")},
+		}
+	}
+	deleted := func() *filer_pb.SubscribeMetadataResponse {
+		return &filer_pb.SubscribeMetadataResponse{Directory: "/buckets/b", EventNotification: &filer_pb.EventNotification{OldEntry: pendingB()}}
+	}
+	deletedPaths := func(r *recordingRemote) (paths []string) {
+		for _, d := range r.deletes {
+			paths = append(paths, d.Path)
+		}
+		return
+	}
+
+	t.Run("deleting the pending entry releases its source", func(t *testing.T) {
+		remote := &recordingRemote{objects: map[string][]byte{"/b/a.jpg": []byte("data-a")}, statETag: `"a"`}
+		if err := processDeleteEvent(&stubFilerClient{}, remote, mountedDir, mountLoc, deleted()); err != nil {
+			t.Fatal(err)
+		}
+		if got := deletedPaths(remote); !slices.Contains(got, "/b/a.jpg") {
+			t.Errorf("deletes = %v, want the source", got)
+		}
+	})
+
+	t.Run("a source replaced since is kept", func(t *testing.T) {
+		// another rename copied onto a.jpg before its entry became visible
+		remote := &recordingRemote{objects: map[string][]byte{"/b/a.jpg": []byte("data-d")}, statETag: `"d"`}
+		if err := processDeleteEvent(&stubFilerClient{}, remote, mountedDir, mountLoc, deleted()); err != nil {
+			t.Fatal(err)
+		}
+		if got := deletedPaths(remote); slices.Contains(got, "/b/a.jpg") {
+			t.Errorf("deletes = %v, want the replaced source kept", got)
+		}
+	})
+
+	t.Run("a later rename the filer copied releases the source", func(t *testing.T) {
+		// b.jpg was renamed to c.jpg and the filer copied a.jpg there at once
+		copied := &filer_pb.Entry{Name: "c.jpg", Attributes: &filer_pb.FuseAttributes{FileSize: 6}, RemoteEntry: &filer_pb.RemoteEntry{StorageName: "b2", RemoteETag: `"a"`, RemoteSize: 6}}
+		remote := &recordingRemote{objects: map[string][]byte{"/b/a.jpg": []byte("data-a"), "/b/b.jpg": []byte("old-b"), "/b/c.jpg": []byte("data-a")}, statETag: `"a"`}
+		event := &filer_pb.SubscribeMetadataResponse{Directory: "/buckets/b", EventNotification: &filer_pb.EventNotification{
+			OldEntry: pendingB(), NewParentPath: "/buckets/b", NewEntry: copied,
+		}}
+		filerClient := &pathFilerClient{entries: map[string]*filer_pb.Entry{"/buckets/b/c.jpg": copied}}
+		if err := processUpdateEvent(filerClient, filerClient, "", remote, mountedDir, mountLoc, event); err != nil {
+			t.Fatal(err)
+		}
+		if got := deletedPaths(remote); !slices.Contains(got, "/b/a.jpg") {
+			t.Errorf("deletes = %v, want the pending source released", got)
+		}
+	})
+}
+
+// pathFilerClient serves entries by path; a path it lacks is not found.
+type pathFilerClient struct {
+	stubFilerClient
+	entries map[string]*filer_pb.Entry
+}
+
+func (c *pathFilerClient) LookupDirectoryEntry(_ context.Context, req *filer_pb.LookupDirectoryEntryRequest, _ ...grpc.CallOption) (*filer_pb.LookupDirectoryEntryResponse, error) {
+	return &filer_pb.LookupDirectoryEntryResponse{Entry: c.entries[req.Directory+"/"+req.Name]}, nil
+}
+
+func (c *pathFilerClient) WithFilerClient(_ bool, fn func(filer_pb.SeaweedFilerClient) error) error {
+	return fn(c)
+}

weed/server/filer_grpc_server_rename.go

@@ -164,8 +164,6 @@ type metadataEvent struct {
 	// 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
-	// the pending source an immediate copy read, released along with it
-	deleteRemoteSource util.FullPath
 }
 
 // pendingRemoteCopy is a remote-only rename over a remote-only target whose
@@ -237,9 +235,6 @@ func (event metadataEvent) notify(f *filer.Filer, ctx context.Context, signature
 	if event.deleteOldRemote {
 		deleteRenamedRemoteObject(f, ctx, event.oldEntry)
 	}
-	if event.deleteRemoteSource != "" {
-		deleteRenamedRemoteObject(f, ctx, &filer.Entry{FullPath: event.deleteRemoteSource, Remote: event.oldEntry.Remote})
-	}
 	f.NotifyUpdateEvent(ctx, event.oldEntry, event.newEntry, event.deleteChunks, false, signatures)
 }
 
@@ -397,7 +392,6 @@ func (fs *FilerServer) moveSelfEntry(ctx context.Context, stream filer_pb.Seawee
 	}
 
 	// add to new directory
-	var pendingSource util.FullPath
 	newEntry := &filer.Entry{
 		FullPath:        newPath,
 		Attr:            entry.Attr,
@@ -417,10 +411,9 @@ func (fs *FilerServer) moveSelfEntry(ctx context.Context, stream filer_pb.Seawee
 		}
 		newEntry.Extended[filer.ExtRemoteRenameSourceKey] = []byte(fs.filer.RemoteSourcePath(entry, oldPath))
 	} else if copiedRemote != nil && len(newEntry.Extended[filer.ExtRemoteRenameSourceKey]) > 0 {
-		// the copy came from the pending source; the new key holds it now
-		if source := fs.filer.RemoteSourcePath(entry, oldPath); source != oldPath {
-			pendingSource = source
-		}
+		// the copy came from the pending source; the new key holds it now.
+		// The source itself is left to remote.sync, which deletes it only
+		// while it is still the object the rename event recorded
 		newEntry.Extended = maps.Clone(newEntry.Extended)
 		delete(newEntry.Extended, filer.ExtRemoteRenameSourceKey)
 	}
@@ -474,10 +467,9 @@ func (fs *FilerServer) moveSelfEntry(ctx context.Context, stream filer_pb.Seawee
 		})
 	}
 	*metadataEvents = append(*metadataEvents, metadataEvent{
-		oldEntry:           sourceEntry,
-		newEntry:           newEntry,
-		deleteOldRemote:    copiedRemote != nil || deferredCopy,
-		deleteRemoteSource: pendingSource,
+		oldEntry:        sourceEntry,
+		newEntry:        newEntry,
+		deleteOldRemote: copiedRemote != nil || deferredCopy,
 	})
 	selfEvent := len(*metadataEvents) - 1
 	keepOldRemote := copiedRemote != nil || deferredCopy

weed/server/filer_grpc_server_rename_remote_test.go

@@ -377,8 +377,9 @@ func TestRenamePendingEntryAgainCopiesFromItsSource(t *testing.T) {
 	client.copyErr = nil
 	require.NoError(t, renameFile(server, "/buckets/b/dst/d.jpg", "/buckets/b/dst/c.jpg"))
 	assert.Contains(t, client.recorded(), "copy origin/src/a.jpg origin/dst/c.jpg")
-	// the source the copy read goes too, once the rename has committed
-	assert.Contains(t, client.recorded(), "delete origin/src/a.jpg")
+	// the source is left to remote.sync: a concurrent rename may have copied
+	// onto that key before its entry is visible
+	assert.NotContains(t, client.recorded(), "delete origin/src/a.jpg")
 	c, err := store.FindEntry(context.Background(), "/buckets/b/dst/c.jpg")
 	require.NoError(t, err)
 	assert.Empty(t, c.Extended[filer.ExtRemoteRenameSourceKey])