filer, remote.sync: follow a pending rename's source across renames

When a remote-only entry is renamed over another remote-only file and
the deferred copy fails, it reads the source key through its marker.
Renamed again the same way, the marker keeps naming that source, but:

- remote.sync completed each rename from the event's old key. For the
  second one that is the first target's old object, which it copied
  over the destination before clearing the marker. It now copies from
  the source the entry's marker names.
- For the first rename it found no entry at the new path and deleted
  the old key, which the later entry still reads. When the event's new
  entry carried a marker and the path is gone, it now leaves the key to
  the entry that moved on.
- When the filer copies a pending entry immediately on a later rename,
  it now also releases the source the copy read, after the commit and
  with the same checks as the old key.

weed/command/filer_remote_sync_dir.go

@@ -295,6 +295,15 @@ func processUpdateEvent(
 	}
 	oldDest := toRemoteStorageLocation(util.FullPath(mountedDir), util.NewFullPath(resp.Directory, message.OldEntry.Name), remoteStorageMountLocation)
 	dest := toRemoteStorageLocation(util.FullPath(mountedDir), util.NewFullPath(message.NewParentPath, message.NewEntry.Name), remoteStorageMountLocation)
+	// sourceOf is where a remote-only entry's content is: the source its
+	// pending-rename marker names, which may be an earlier rename's old key,
+	// or else the event's old key
+	sourceOf := func(entry *filer_pb.Entry, eventSource *remote_pb.RemoteStorageLocation) *remote_pb.RemoteStorageLocation {
+		if src := string(entry.Extended[filer.ExtRemoteRenameSourceKey]); strings.HasPrefix(src, strings.TrimSuffix(mountedDir, "/")+"/") {
+			return toRemoteStorageLocation(util.FullPath(mountedDir), util.FullPath(src), remoteStorageMountLocation)
+		}
+		return eventSource
+	}
 	if proto.Equal(oldDest, dest) && !shouldSendToRemote(message.NewEntry) {
 		glog.V(2).Infof("skipping updating: %+v", resp)
 		return nil
@@ -331,7 +340,14 @@ func processUpdateEvent(
 			return err
 		}
 		if isRemoteOnly(current) {
-			return completeRemoteOnlyRename(filerClient, client, message.NewParentPath, current, oldDest, dest, storageClass)
+			return completeRemoteOnlyRename(filerClient, client, message.NewParentPath, current, sourceOf(current, oldDest), dest, storageClass)
+		}
+		if current == nil && len(message.NewEntry.Extended[filer.ExtRemoteRenameSourceKey]) > 0 {
+			// the rename's copy was pending and the entry has moved on: a
+			// later rename carries the source in its own marker and finishes
+			// or releases it, so this one leaves it
+			glog.V(0).Infof("keep %s: the pending rename moved on", remote_storage.FormatLocation(oldDest))
+			return nil
 		}
 		glog.V(0).Infof("delete %s", remote_storage.FormatLocation(oldDest))
 		if err := client.DeleteFile(oldDest); err != nil {
@@ -347,7 +363,7 @@ func processUpdateEvent(
 	if errors.Is(writeErr, errSuperseded) {
 		glog.Errorf("skipping %s: %v", remote_storage.FormatLocation(dest), writeErr)
 		if !proto.Equal(oldDest, dest) {
-			return uploadCurrentEntry(filerClient, filerSource, client, message.NewParentPath, message.NewEntry.Name, oldDest, dest, storageClass)
+			return uploadCurrentEntry(filerClient, filerSource, client, message.NewParentPath, message.NewEntry.Name, oldDest, dest, storageClass, sourceOf)
 		}
 		return nil
 	}
@@ -469,7 +485,7 @@ func openRemoteObject(client remote_storage.RemoteStorageClient, loc *remote_pb.
 // up once. A remote-only entry is finished the way completeRemoteOnlyRename
 // finishes a rename: its stamp can already describe the destination (a sync
 // plus remote.uncache in the meantime), and only then is it complete.
-func uploadCurrentEntry(filerClient filer_pb.FilerClient, filerSource filer_pb.FilerClient, client remote_storage.RemoteStorageClient, dir, name string, oldDest, dest *remote_pb.RemoteStorageLocation, storageClass string) error {
+func uploadCurrentEntry(filerClient filer_pb.FilerClient, filerSource filer_pb.FilerClient, client remote_storage.RemoteStorageClient, dir, name string, oldDest, dest *remote_pb.RemoteStorageLocation, storageClass string, sourceOf func(*filer_pb.Entry, *remote_pb.RemoteStorageLocation) *remote_pb.RemoteStorageLocation) error {
 	current, _, _, err := filer_pb.GetEntry(context.Background(), filerSource, util.NewFullPath(dir, name))
 	if errors.Is(err, filer_pb.ErrNotFound) {
 		return nil
@@ -481,7 +497,7 @@ func uploadCurrentEntry(filerClient filer_pb.FilerClient, filerSource filer_pb.F
 		return nil
 	}
 	if isRemoteOnly(current) {
-		return completeRemoteOnlyRename(filerClient, client, dir, current, oldDest, dest, storageClass)
+		return completeRemoteOnlyRename(filerClient, client, dir, current, sourceOf(current, oldDest), dest, storageClass)
 	}
 	if shouldSendToRemote(current) {
 		glog.V(0).Infof("leaving %s to the rewrite that superseded the rename", remote_storage.FormatLocation(dest))

weed/command/filer_remote_sync_dir_test.go

@@ -1377,7 +1377,7 @@ func TestSupersededRenameUploadsCurrentEntry(t *testing.T) {
 		current := entryWith("b.txt", &filer_pb.RemoteEntry{StorageName: "gcs", RemoteMtime: 1786096669, RemoteSize: 1024})
 		remote := &recordingRemote{}
 		filerClient := &stubFilerClient{entry: current}
-		err := uploadCurrentEntry(filerClient, filerClient, remote, "/buckets/b/dir", "b.txt", wantDelete, wantWrite, "")
+		err := uploadCurrentEntry(filerClient, filerClient, remote, "/buckets/b/dir", "b.txt", wantDelete, wantWrite, "", eventSource)
 		if err == nil || !strings.Contains(err.Error(), "neither") {
 			t.Errorf("err = %v, want the lost-content failure", err)
 		}
@@ -1390,7 +1390,7 @@ func TestSupersededRenameUploadsCurrentEntry(t *testing.T) {
 		current := entryWith("b.txt", &filer_pb.RemoteEntry{StorageName: "gcs", RemoteMtime: 1786096669, RemoteSize: 7})
 		remote := &recordingRemote{objects: map[string][]byte{"/b/dir/b.txt": []byte("payload")}}
 		filerClient := &stubFilerClient{entry: current}
-		if err := uploadCurrentEntry(filerClient, filerClient, remote, "/buckets/b/dir", "b.txt", wantDelete, wantWrite, ""); err != nil {
+		if err := uploadCurrentEntry(filerClient, filerClient, remote, "/buckets/b/dir", "b.txt", wantDelete, wantWrite, "", eventSource); err != nil {
 			t.Fatalf("err = %v, want nil: the destination object matches the entry's stamp", err)
 		}
 		if len(remote.writes) != 0 {
@@ -1446,3 +1446,65 @@ func TestCompleteRemoteOnlyRenameClearsThePendingSourceFirst(t *testing.T) {
 		}
 	})
 }
+
+// eventSource resolves a rename's source to the event's old key.
+func eventSource(_ *filer_pb.Entry, oldDest *remote_pb.RemoteStorageLocation) *remote_pb.RemoteStorageLocation {
+	return oldDest
+}
+
+func TestRenameOfAPendingEntryFollowsItsSource(t *testing.T) {
+	// A was renamed over B and B over C, both copies failing: C reads A's
+	// object through the marker, and B's key holds the old target's bytes
+	const mountedDir = "/buckets"
+	mountLoc := &remote_pb.RemoteStorageLocation{Name: "b2", Bucket: "bucket", Path: "/"}
+	remoteOnly := &filer_pb.RemoteEntry{StorageName: "b2", RemoteETag: `"a"`, RemoteSize: 6}
+	pendingC := func() *filer_pb.Entry {
+		return &filer_pb.Entry{
+			Name:        "c.jpg",
+			Attributes:  &filer_pb.FuseAttributes{FileSize: 6},
+			RemoteEntry: remoteOnly,
+			Extended:    map[string][]byte{filer.ExtRemoteRenameSourceKey: []byte("/buckets/b/a.jpg")},
+		}
+	}
+	event := func(oldName, newName string) *filer_pb.SubscribeMetadataResponse {
+		newEntry := pendingC()
+		newEntry.Name = newName
+		return &filer_pb.SubscribeMetadataResponse{
+			Directory: "/buckets/b",
+			EventNotification: &filer_pb.EventNotification{
+				OldEntry:      &filer_pb.Entry{Name: oldName, RemoteEntry: remoteOnly},
+				NewParentPath: "/buckets/b",
+				NewEntry:      newEntry,
+			},
+		}
+	}
+	objects := func() map[string][]byte {
+		return map[string][]byte{"/b/a.jpg": []byte("data-a"), "/b/b.jpg": []byte("old-b"), "/b/c.jpg": []byte("old-c")}
+	}
+
+	t.Run("the first rename leaves the source to the entry that moved on", func(t *testing.T) {
+		remote := &recordingRemote{objects: objects()}
+		filerClient := &stubFilerClient{} // nothing at b.jpg any more
+		if err := processUpdateEvent(filerClient, filerClient, "", remote, mountedDir, mountLoc, event("a.jpg", "b.jpg")); err != nil {
+			t.Fatal(err)
+		}
+		if len(remote.deletes) != 0 || len(remote.writes) != 0 {
+			t.Errorf("deletes = %+v, writes = %+v, want neither: c.jpg still reads a.jpg", remote.deletes, remote.writes)
+		}
+	})
+
+	t.Run("the second rename copies from the marker's source", func(t *testing.T) {
+		current := pendingC()
+		remote := &recordingRemote{objects: objects()}
+		filerClient := &stubFilerClient{entry: current}
+		if err := processUpdateEvent(filerClient, filerClient, "", remote, mountedDir, mountLoc, event("b.jpg", "c.jpg")); err != nil {
+			t.Fatal(err)
+		}
+		if len(remote.written) != 1 || string(remote.written[0]) != "data-a" {
+			t.Errorf("written = %q, want a.jpg's content", remote.written)
+		}
+		if len(remote.deletes) != 1 || remote.deletes[0].Path != "/b/a.jpg" {
+			t.Errorf("deletes = %+v, want the source once the stamp cleared the marker", remote.deletes)
+		}
+	})
+}

weed/server/filer_grpc_server_rename.go

@@ -164,6 +164,8 @@ 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
@@ -233,25 +235,33 @@ func applyRemoteCopies(events []metadataEvent, pendings []*pendingRemoteCopy) {
 
 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)
-			}
-		}
+		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)
 }
 
+// deleteRenamedRemoteObject deletes the remote object a committed rename no
+// longer needs. A child path is not locked by its directory's rename, so
+// another rename may have put a remote-only entry at the key meanwhile; the
+// store is asked directly, a lazy fetch would revive the key. When in doubt
+// the object stays: an extra object loses nothing.
+func deleteRenamedRemoteObject(f *filer.Filer, ctx context.Context, entry *filer.Entry) {
+	current, findErr := f.Store.FindEntry(ctx, entry.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", entry.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", entry.FullPath)
+	default:
+		if err := f.DeleteFromRemote(ctx, entry); err != nil {
+			glog.WarningfCtx(ctx, "delete remote object of renamed %s: %v", entry.FullPath, err)
+		}
+	}
+}
+
 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 {
@@ -387,6 +397,7 @@ 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,
@@ -407,6 +418,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
+		}
 		newEntry.Extended = maps.Clone(newEntry.Extended)
 		delete(newEntry.Extended, filer.ExtRemoteRenameSourceKey)
 	}
@@ -460,9 +474,10 @@ func (fs *FilerServer) moveSelfEntry(ctx context.Context, stream filer_pb.Seawee
 		})
 	}
 	*metadataEvents = append(*metadataEvents, metadataEvent{
-		oldEntry:        sourceEntry,
-		newEntry:        newEntry,
-		deleteOldRemote: copiedRemote != nil || deferredCopy,
+		oldEntry:           sourceEntry,
+		newEntry:           newEntry,
+		deleteOldRemote:    copiedRemote != nil || deferredCopy,
+		deleteRemoteSource: pendingSource,
 	})
 	selfEvent := len(*metadataEvents) - 1
 	keepOldRemote := copiedRemote != nil || deferredCopy

weed/server/filer_grpc_server_rename_remote_test.go

@@ -377,6 +377,8 @@ 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")
 	c, err := store.FindEntry(context.Background(), "/buckets/b/dst/c.jpg")
 	require.NoError(t, err)
 	assert.Empty(t, c.Extended[filer.ExtRemoteRenameSourceKey])