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