filer: tighten the pending rename source marker

Follow-ups to reading a pending remote rename copy from its source key:

- Only the filer sets the marker. The HTTP upload and tagging handlers
  store Seaweed-* request headers as extended attributes, so they skip
  this one, and a marker that leads out of the entry's own mount is
  ignored, so it cannot point reads at another remote.
- A pending entry renamed again copies from the marker's source, not
  from its own key, which still holds the old target's bytes. A further
  deferred rename keeps naming the original source; an immediate copy
  clears the marker.
- When remote.sync finds the destination already holding the content,
  it stamps the entry (clearing the marker) before deleting the source,
  and deletes the source only once the entry no longer reads it.
- A read that resolved the source just before the copy finished and
  the source went away retries with the entry as it is now.

weed/command/filer_remote_sync_dir.go

@@ -387,7 +387,13 @@ func completeRemoteOnlyRename(filerClient filer_pb.FilerClient, client remote_st
 	if existing, err := client.StatFile(dest); err == nil {
 		if describes(current.RemoteEntry, existing) {
 			glog.V(0).Infof("%s already holds the renamed content", remote_storage.FormatLocation(dest))
-			return deleteRemoteFile(client, oldDest)
+			if len(current.Extended[filer.ExtRemoteRenameSourceKey]) > 0 {
+				// reads still go to the source until the stamp clears the marker
+				if err := updateLocalEntry(filerClient, dir, current, existing); err != nil {
+					return err
+				}
+			}
+			return deleteRenameSource(filerClient, client, dir, current.Name, oldDest)
 		}
 		glog.V(0).Infof("%s holds an object the entry does not describe (size %d, want %d); replacing it from %s", remote_storage.FormatLocation(dest), existing.RemoteSize, current.RemoteEntry.RemoteSize, remote_storage.FormatLocation(oldDest))
 	} else if !errors.Is(err, remote_storage.ErrRemoteObjectNotFound) {
@@ -413,6 +419,20 @@ func completeRemoteOnlyRename(filerClient filer_pb.FilerClient, client remote_st
 	if err := updateLocalEntry(filerClient, dir, current, remoteEntry); err != nil {
 		return err
 	}
+	return deleteRenameSource(filerClient, client, dir, current.Name, oldDest)
+}
+
+// deleteRenameSource deletes a rename's old key once the renamed entry no
+// longer reads it: an entry still carrying the pending-source marker (its
+// stamp lost a race) keeps the source, and the event is retried.
+func deleteRenameSource(filerClient filer_pb.FilerClient, client remote_storage.RemoteStorageClient, dir, name string, oldDest *remote_pb.RemoteStorageLocation) error {
+	current, err := currentEntry(filerClient, dir, name)
+	if err != nil {
+		return err
+	}
+	if current != nil && len(current.Extended[filer.ExtRemoteRenameSourceKey]) > 0 {
+		return fmt.Errorf("%s still reads %s", util.NewFullPath(dir, name), remote_storage.FormatLocation(oldDest))
+	}
 	return deleteRemoteFile(client, oldDest)
 }
 

weed/command/filer_remote_sync_dir_test.go

@@ -1401,3 +1401,48 @@ func TestSupersededRenameUploadsCurrentEntry(t *testing.T) {
 		}
 	})
 }
+
+func TestCompleteRemoteOnlyRenameClearsThePendingSourceFirst(t *testing.T) {
+	oldDest := &remote_pb.RemoteStorageLocation{Name: "b2", Bucket: "bucket", Path: "/b/src/a.jpg"}
+	dest := &remote_pb.RemoteStorageLocation{Name: "b2", Bucket: "bucket", Path: "/b/dst/a.jpg"}
+	pending := func() *filer_pb.Entry {
+		return &filer_pb.Entry{
+			Name:        "a.jpg",
+			Attributes:  &filer_pb.FuseAttributes{FileSize: 4},
+			RemoteEntry: &filer_pb.RemoteEntry{RemoteETag: `"copy"`, RemoteSize: 4},
+			Extended:    map[string][]byte{filer.ExtRemoteRenameSourceKey: []byte("/buckets/b/src/a.jpg")},
+		}
+	}
+	objects := func() map[string][]byte {
+		return map[string][]byte{"/b/src/a.jpg": []byte("data"), "/b/dst/a.jpg": []byte("data")}
+	}
+
+	t.Run("stamped, then the source goes", func(t *testing.T) {
+		// the destination already holds the copy; the entry still reads the
+		// source until its stamp clears the marker
+		current := pending()
+		filerClient := &stubFilerClient{entry: current}
+		remote := &recordingRemote{objects: objects(), statETag: `"copy"`}
+		if err := completeRemoteOnlyRename(filerClient, remote, "/buckets/b/dst", current, oldDest, dest, ""); err != nil {
+			t.Fatalf("err = %v", err)
+		}
+		if filerClient.updates != 1 {
+			t.Errorf("updates = %d, want the stamp that clears the marker", filerClient.updates)
+		}
+		if len(remote.deletes) != 1 || remote.deletes[0].Path != "/b/src/a.jpg" {
+			t.Errorf("deletes = %+v, want the source", remote.deletes)
+		}
+	})
+
+	t.Run("a marker still in place keeps the source", func(t *testing.T) {
+		// the stamp did not land: the filer still has the marker
+		filerClient := &stubFilerClient{entry: pending()}
+		remote := &recordingRemote{objects: objects(), statETag: `"copy"`}
+		if err := completeRemoteOnlyRename(filerClient, remote, "/buckets/b/dst", pending(), oldDest, dest, ""); err == nil {
+			t.Fatal("err = nil, want the event retried")
+		}
+		if len(remote.deletes) != 0 {
+			t.Errorf("deletes = %+v, want none while the entry reads the source", remote.deletes)
+		}
+	})
+}

weed/filer/filer_lazy_remote.go

@@ -238,14 +238,29 @@ func (f *Filer) DeleteFromRemote(ctx context.Context, entry *Entry) error {
 // object remote-only reads must serve until the copy stamps the entry.
 const ExtRemoteRenameSourceKey = "Seaweed-X-Remote-Rename-Source"
 
-// RemoteReadPath returns the dir and name whose object a remote-only read of
-// entry serves: its own, or the renamed-from path a pending deferred copy
-// still reads.
-func RemoteReadPath(entry *Entry, dir, name string) (string, string) {
-	if src := entry.Extended[ExtRemoteRenameSourceKey]; len(src) > 0 {
-		return util.FullPath(src).DirAndName()
+// RemoteSourcePath returns the path whose remote object holds the content of
+// the remote-only entry at p: the renamed-from path a pending deferred copy
+// still reads, or p itself. A marker leading out of p's mount is ignored, so
+// whoever can set extended attributes cannot point reads at another mount.
+func (f *Filer) RemoteSourcePath(entry *Entry, p util.FullPath) util.FullPath {
+	src := util.FullPath(entry.Extended[ExtRemoteRenameSourceKey])
+	if src == "" || f.RemoteStorage == nil {
+		return p
 	}
-	return dir, name
+	mountDir, remoteLoc := f.RemoteStorage.FindMountDirectory(p)
+	if remoteLoc == nil {
+		return p
+	}
+	if srcMountDir, _ := f.RemoteStorage.FindMountDirectory(src); srcMountDir != mountDir {
+		return p
+	}
+	return src
+}
+
+// RemoteReadPath returns the dir and name whose object a remote-only read of
+// the entry at dir/name serves.
+func (f *Filer) RemoteReadPath(entry *Entry, dir, name string) (string, string) {
+	return f.RemoteSourcePath(entry, util.NewFullPath(dir, name)).DirAndName()
 }
 
 // CopyRemoteOnlyEntry copies the remote object of a renamed entry whose
@@ -300,7 +315,7 @@ func (f *Filer) RemoteCopyLocations(ctx context.Context, entry *Entry, oldPath,
 		return nil, nil, nil, fmt.Errorf("%s exists only on the remote and remote storage type %s cannot copy it to the new name; cache it first", oldPath, remoteConf.Type)
 	}
 
-	src = MapFullPathToRemoteStorageLocation(mountDir, remoteLoc, oldPath)
+	src = MapFullPathToRemoteStorageLocation(mountDir, remoteLoc, f.RemoteSourcePath(entry, oldPath))
 	dst = MapFullPathToRemoteStorageLocation(mountDir, remoteLoc, newPath)
 	return src, dst, copier, nil
 }

weed/server/filer_grpc_server_remote.go

@@ -96,7 +96,7 @@ func (fs *FilerServer) doCacheRemoteObjectToLocalCluster(ctx context.Context, re
 
 	glog.V(1).Infof("CacheRemoteObjectToLocalCluster: caching %s/%s (remote size: %d)", req.Directory, req.Name, entry.Remote.RemoteSize)
 
-	readDir, readName := filer.RemoteReadPath(entry, req.Directory, req.Name)
+	readDir, readName := fs.filer.RemoteReadPath(entry, req.Directory, req.Name)
 	storageConf, remoteLocation, err := fs.resolveMountedRemote(ctx, readDir, readName)
 	if err != nil {
 		return nil, err

weed/server/filer_grpc_server_rename.go

@@ -399,11 +399,16 @@ func (fs *FilerServer) moveSelfEntry(ctx context.Context, stream filer_pb.Seawee
 		Quota:           entry.Quota,
 	}
 	if deferredCopy {
+		// a pending entry renamed again still reads its original source
 		newEntry.Extended = maps.Clone(newEntry.Extended)
 		if newEntry.Extended == nil {
 			newEntry.Extended = map[string][]byte{}
 		}
-		newEntry.Extended[filer.ExtRemoteRenameSourceKey] = []byte(oldPath)
+		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
+		newEntry.Extended = maps.Clone(newEntry.Extended)
+		delete(newEntry.Extended, filer.ExtRemoteRenameSourceKey)
 	}
 	if skipTargetLookup {
 		if newEntry.FullPath.IsLongerFileName(fs.filer.MaxFilenameLength) {

weed/server/filer_grpc_server_rename_remote_test.go

@@ -3,6 +3,8 @@ package weed_server
 import (
 	"context"
 	"errors"
+	"net/http"
+	"net/http/httptest"
 	"sync"
 	"testing"
 
@@ -181,7 +183,7 @@ func TestRenameRemoteOnlyEntryCopyFailureKeepsTheOldObject(t *testing.T) {
 	// reads keep resolving the source's key until the copy stamps the entry:
 	// its own key still holds the target's old object
 	assert.Equal(t, "/buckets/b/src/a.jpg", string(dst.Extended[filer.ExtRemoteRenameSourceKey]))
-	readDir, readName := filer.RemoteReadPath(dst, "/buckets/b/dst", "a.jpg")
+	readDir, readName := server.filer.RemoteReadPath(dst, "/buckets/b/dst", "a.jpg")
 	assert.Equal(t, "/buckets/b/src", readDir)
 	assert.Equal(t, "a.jpg", readName)
 }
@@ -353,3 +355,58 @@ func TestRenameRemoteOnlyEntryFinishesAfterTheCommitOnASQLStore(t *testing.T) {
 		})
 	}
 }
+
+func TestRenamePendingEntryAgainCopiesFromItsSource(t *testing.T) {
+	client := copyingRenameRemoteClient{&renameRemoteClient{copyErr: errors.New("copy refused")}}
+	server, store := newRenameRemoteTestServer(t, client)
+	store.entries["/buckets/b/src/a.jpg"] = remoteOnlyRenameEntry("/buckets/b/src/a.jpg", 101)
+	store.entries["/buckets/b/dst/a.jpg"] = remoteOnlyRenameEntry("/buckets/b/dst/a.jpg", 202)
+	store.entries["/buckets/b/dst/d.jpg"] = remoteOnlyRenameEntry("/buckets/b/dst/d.jpg", 303)
+
+	// the deferred copy fails: dst/a.jpg reads src/a.jpg's object
+	require.NoError(t, renameFile(server, "/buckets/b/src/a.jpg", "/buckets/b/dst/a.jpg"))
+
+	// renamed again over another remote-only file, failing again: the marker
+	// keeps naming the original source, not dst/a.jpg's old object
+	require.NoError(t, renameFile(server, "/buckets/b/dst/a.jpg", "/buckets/b/dst/d.jpg"))
+	d, err := store.FindEntry(context.Background(), "/buckets/b/dst/d.jpg")
+	require.NoError(t, err)
+	assert.Equal(t, "/buckets/b/src/a.jpg", string(d.Extended[filer.ExtRemoteRenameSourceKey]))
+
+	// renamed to a new path, the copy comes from the source and clears the marker
+	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")
+	c, err := store.FindEntry(context.Background(), "/buckets/b/dst/c.jpg")
+	require.NoError(t, err)
+	assert.Empty(t, c.Extended[filer.ExtRemoteRenameSourceKey])
+	assert.Equal(t, `"copy"`, c.Remote.RemoteETag)
+}
+
+func TestRemoteSourcePathStaysInTheMount(t *testing.T) {
+	server, _ := newRenameRemoteTestServer(t, &renameRemoteClient{})
+	entry := remoteOnlyRenameEntry("/buckets/b/x.jpg", 101)
+	for marker, want := range map[string]util.FullPath{
+		"/buckets/b/src/y.jpg": "/buckets/b/src/y.jpg", // same mount: the pending source
+		"/buckets/c/y.jpg":     "/buckets/b/x.jpg",     // another mount: ignored
+		"/tmp/y.jpg":           "/buckets/b/x.jpg",     // no mount: ignored
+	} {
+		entry.Extended = map[string][]byte{filer.ExtRemoteRenameSourceKey: []byte(marker)}
+		assert.Equal(t, want, server.filer.RemoteSourcePath(entry, "/buckets/b/x.jpg"), marker)
+	}
+}
+
+func TestTaggingCannotSetThePendingSourceMarker(t *testing.T) {
+	server, store := newRenameRemoteTestServer(t, &renameRemoteClient{})
+	store.entries["/buckets/b/x.jpg"] = remoteOnlyRenameEntry("/buckets/b/x.jpg", 101)
+
+	r := httptest.NewRequest(http.MethodPut, "/buckets/b/x.jpg?tagging", nil)
+	r.Header.Set("Seaweed-X-Remote-Rename-Source", "/buckets/b/other.jpg")
+	r.Header.Set("Seaweed-Album", "2026")
+	server.PutTaggingHandler(httptest.NewRecorder(), r)
+
+	x, err := store.FindEntry(context.Background(), "/buckets/b/x.jpg")
+	require.NoError(t, err)
+	assert.Equal(t, "2026", string(x.Extended["Seaweed-Album"]))
+	assert.Empty(t, x.Extended[filer.ExtRemoteRenameSourceKey], "only the filer sets the pending source")
+}

weed/server/filer_server_handlers_read.go

@@ -206,16 +206,18 @@ func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request)
 
 	// Every part of a multipart Range is prepared on its own, so the origin
 	// preflight below is memoized to one stat per request.
-	statOrigin := sync.OnceValue(func() error {
-		dir, name := entry.FullPath.DirAndName()
-		readDir, readName := filer.RemoteReadPath(entry, dir, name)
-		client, remoteLocation, err := fs.mountedRemoteClient(ctx, readDir, readName)
-		if err != nil {
+	statOriginAt := func(readDir, readName string) func() error {
+		return sync.OnceValue(func() error {
+			client, remoteLocation, err := fs.mountedRemoteClient(ctx, readDir, readName)
+			if err != nil {
+				return err
+			}
+			_, err = client.StatFile(remoteLocation)
 			return err
-		}
-		_, err = client.StatFile(remoteLocation)
-		return err
-	})
+		})
+	}
+	entryDir, entryName := entry.FullPath.DirAndName()
+	statOrigin := statOriginAt(fs.filer.RemoteReadPath(entry, entryDir, entryName))
 
 	ProcessRangeRequest(r, w, totalSize, mimeType, func(offset int64, size int64) (filer.DoStreamContent, error) {
 		if offset+size <= int64(len(entry.Content)) {
@@ -231,14 +233,23 @@ func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request)
 		chunks := entry.GetChunks()
 		if entry.IsInRemoteOnly() {
 			dir, name := entry.FullPath.DirAndName()
-			readDir, readName := filer.RemoteReadPath(entry, dir, name)
+			readDir, readName := fs.filer.RemoteReadPath(entry, dir, name)
 			var mountedLocation *remote_pb.RemoteStorageLocation
 			if fs.filer.RemoteStorage != nil {
 				_, mountedLocation = fs.filer.RemoteStorage.FindMountDirectory(entry.FullPath)
 			}
 			cacheWait := remote_storage.CacheWaitTimeout(entry.Remote.RemoteSize, mountedLocation)
 			if cacheWait <= 0 {
-				return fs.streamFromRemoteOnly(ctx, r, readDir, readName, offset, size, statOrigin)
+				streamFn, err := fs.streamFromRemoteOnly(ctx, r, readDir, readName, offset, size, statOrigin)
+				if errors.Is(err, filer_pb.ErrNotFound) && (readDir != dir || readName != name) {
+					// the pending copy finished and its source went away since
+					// the entry was read; serve what the entry names now
+					if fresh, findErr := fs.filer.FindEntry(ctx, entry.FullPath); findErr == nil {
+						freshDir, freshName := fs.filer.RemoteReadPath(fresh, dir, name)
+						return fs.streamFromRemoteOnly(ctx, r, freshDir, freshName, offset, size, statOriginAt(freshDir, freshName))
+					}
+				}
+				return streamFn, err
 			}
 			// Bounded wait: a large download outlasts any client timeout, so
 			// serve straight from the origin once the wait expires while the

weed/server/filer_server_handlers_tagging.go

@@ -4,6 +4,7 @@ import (
 	"net/http"
 	"strings"
 
+	"github.com/seaweedfs/seaweedfs/weed/filer"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
 	"github.com/seaweedfs/seaweedfs/weed/util"
@@ -35,7 +36,7 @@ func (fs *FilerServer) PutTaggingHandler(w http.ResponseWriter, r *http.Request)
 	}
 
 	for header, values := range r.Header {
-		if strings.HasPrefix(header, needle.PairNamePrefix) {
+		if strings.HasPrefix(header, needle.PairNamePrefix) && header != filer.ExtRemoteRenameSourceKey {
 			for _, value := range values {
 				existingEntry.Extended[header] = []byte(value)
 			}

weed/server/filer_server_handlers_write_autochunk.go

@@ -468,6 +468,9 @@ func (fs *FilerServer) saveMetaData(ctx context.Context, r *http.Request, fileNa
 	}
 	for k, v := range r.Header {
 		if len(v) > 0 && len(v[0]) > 0 {
+			if k == filer.ExtRemoteRenameSourceKey {
+				continue // set by the filer only: it picks the remote key reads serve
+			}
 			if strings.HasPrefix(k, needle.PairNamePrefix) || k == "Cache-Control" || k == "Expires" || k == "Content-Disposition" {
 				entry.Extended[k] = []byte(v[0])
 			}