From 4e6339430261f6e3ee63171f35085b8c25161198 Mon Sep 17 00:00:00 2001 From: Alex Russell Date: Mon, 12 Oct 2026 06:07:52 +0200 Subject: [PATCH] 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 | 22 ++++++- weed/command/filer_remote_sync_dir_test.go | 45 ++++++++++++++ weed/filer/filer_lazy_remote.go | 31 +++++++--- weed/server/filer_grpc_server_remote.go | 2 +- weed/server/filer_grpc_server_rename.go | 7 ++- .../filer_grpc_server_rename_remote_test.go | 59 ++++++++++++++++++- weed/server/filer_server_handlers_read.go | 33 +++++++---- weed/server/filer_server_handlers_tagging.go | 3 +- .../filer_server_handlers_write_autochunk.go | 3 + 9 files changed, 181 insertions(+), 24 deletions(-) diff --git a/weed/command/filer_remote_sync_dir.go b/weed/command/filer_remote_sync_dir.go index 3b0e604783..2c30cfcf50 100644 --- a/weed/command/filer_remote_sync_dir.go +++ b/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) } diff --git a/weed/command/filer_remote_sync_dir_test.go b/weed/command/filer_remote_sync_dir_test.go index 374963c373..69a96e551f 100644 --- a/weed/command/filer_remote_sync_dir_test.go +++ b/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) + } + }) +} diff --git a/weed/filer/filer_lazy_remote.go b/weed/filer/filer_lazy_remote.go index a9727f7a61..09110adfae 100644 --- a/weed/filer/filer_lazy_remote.go +++ b/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 } diff --git a/weed/server/filer_grpc_server_remote.go b/weed/server/filer_grpc_server_remote.go index c732b47388..892789fb7f 100644 --- a/weed/server/filer_grpc_server_remote.go +++ b/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 diff --git a/weed/server/filer_grpc_server_rename.go b/weed/server/filer_grpc_server_rename.go index 63e8cdd406..93d02a0963 100644 --- a/weed/server/filer_grpc_server_rename.go +++ b/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) { diff --git a/weed/server/filer_grpc_server_rename_remote_test.go b/weed/server/filer_grpc_server_rename_remote_test.go index 8750b3a479..28ec86f154 100644 --- a/weed/server/filer_grpc_server_rename_remote_test.go +++ b/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") +} diff --git a/weed/server/filer_server_handlers_read.go b/weed/server/filer_server_handlers_read.go index f9a3662bd7..2fa32fa2ff 100644 --- a/weed/server/filer_server_handlers_read.go +++ b/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 diff --git a/weed/server/filer_server_handlers_tagging.go b/weed/server/filer_server_handlers_tagging.go index 06cd1295e9..0d809365ea 100644 --- a/weed/server/filer_server_handlers_tagging.go +++ b/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) } diff --git a/weed/server/filer_server_handlers_write_autochunk.go b/weed/server/filer_server_handlers_write_autochunk.go index 2c86affe25..d0abfd7b04 100644 --- a/weed/server/filer_server_handlers_write_autochunk.go +++ b/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]) }