mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-14 01:58:33 +02:00
filer: read a pending remote rename copy from the source key
A remote-only rename over a remote-only target commits before its copy lands, so reads on the new path served the target's old object until the copy stamped the entry — and indefinitely when the copy failed without remote.sync running. The committed entry now records the renamed-from path, and remote-only reads resolve that object's key while the stamp is pending; the copy's stamp and the sync's completion stamp both clear the marker. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
1 parent
bc6f528d97
commit
68109213a4
6 files changed
+38
-4
No files matched your search
@@ -658,6 +658,7 @@ func updateLocalEntry(filerClient filer_pb.FilerClient, dir string, entry *filer
|
||||
remoteEntry.LastLocalSyncTsNs = time.Now().UnixNano()
|
||||
expected := proto.Clone(entry).(*filer_pb.Entry)
|
||||
entry.RemoteEntry = remoteEntry
|
||||
delete(entry.Extended, filer.ExtRemoteRenameSourceKey)
|
||||
err := filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
|
||||
_, err := client.UpdateEntry(context.Background(), &filer_pb.UpdateEntryRequest{
|
||||
Directory: dir,
|
||||
|
||||
@@ -233,6 +233,21 @@ func (f *Filer) DeleteFromRemote(ctx context.Context, entry *Entry) error {
|
||||
return err
|
||||
}
|
||||
|
||||
// ExtRemoteRenameSourceKey marks a remote-only entry whose rename committed
|
||||
// while its copy to the new key was still pending: it holds the path whose
|
||||
// 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()
|
||||
}
|
||||
return dir, name
|
||||
}
|
||||
|
||||
// CopyRemoteOnlyEntry copies the remote object of a renamed entry whose
|
||||
// content exists only on the remote to the key of newPath, and returns the
|
||||
// RemoteEntry of the copy. The rename deletes the old key along with the old
|
||||
|
||||
@@ -96,7 +96,8 @@ 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)
|
||||
|
||||
storageConf, remoteLocation, err := fs.resolveMountedRemote(ctx, req.Directory, req.Name)
|
||||
readDir, readName := filer.RemoteReadPath(entry, req.Directory, req.Name)
|
||||
storageConf, remoteLocation, err := fs.resolveMountedRemote(ctx, readDir, readName)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
@@ -196,6 +196,7 @@ func (fs *FilerServer) finishRemoteCopies(ctx context.Context, pendings []*pendi
|
||||
continue
|
||||
}
|
||||
entry.Remote = remoteEntry
|
||||
delete(entry.Extended, filer.ExtRemoteRenameSourceKey)
|
||||
if updateErr := fs.filer.UpdateEntry(ctx, entry, entry, false); updateErr != nil {
|
||||
glog.WarningfCtx(ctx, "stamp remote rename copy of %s: %v", pending.newPath, updateErr)
|
||||
pending.failed = true
|
||||
@@ -394,6 +395,13 @@ func (fs *FilerServer) moveSelfEntry(ctx context.Context, stream filer_pb.Seawee
|
||||
Remote: remoteEntry,
|
||||
Quota: entry.Quota,
|
||||
}
|
||||
if deferredCopy {
|
||||
newEntry.Extended = maps.Clone(newEntry.Extended)
|
||||
if newEntry.Extended == nil {
|
||||
newEntry.Extended = map[string][]byte{}
|
||||
}
|
||||
newEntry.Extended[filer.ExtRemoteRenameSourceKey] = []byte(oldPath)
|
||||
}
|
||||
if skipTargetLookup {
|
||||
if newEntry.FullPath.IsLongerFileName(fs.filer.MaxFilenameLength) {
|
||||
return filer_pb.ErrEntryNameTooLong
|
||||
|
||||
@@ -153,6 +153,7 @@ func TestRenameRemoteOnlyEntryOverTargetKeepsTheCopy(t *testing.T) {
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, uint64(101), dst.Attr.Inode)
|
||||
assert.Equal(t, `"copy"`, dst.Remote.RemoteETag)
|
||||
assert.Empty(t, dst.Extended[filer.ExtRemoteRenameSourceKey])
|
||||
}
|
||||
|
||||
func TestRenameRemoteOnlyEntryCopyFailureKeepsTheOldObject(t *testing.T) {
|
||||
@@ -172,6 +173,12 @@ func TestRenameRemoteOnlyEntryCopyFailureKeepsTheOldObject(t *testing.T) {
|
||||
dst, err := store.FindEntry(context.Background(), "/buckets/b/dst/a.jpg")
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, `"orig"`, dst.Remote.RemoteETag)
|
||||
// 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")
|
||||
assert.Equal(t, "/buckets/b/src", readDir)
|
||||
assert.Equal(t, "a.jpg", readName)
|
||||
}
|
||||
|
||||
func TestRenameRemoteOnlyEntryWithoutCopierIsRefused(t *testing.T) {
|
||||
|
||||
@@ -208,7 +208,8 @@ func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request)
|
||||
// preflight below is memoized to one stat per request.
|
||||
statOrigin := sync.OnceValue(func() error {
|
||||
dir, name := entry.FullPath.DirAndName()
|
||||
client, remoteLocation, err := fs.mountedRemoteClient(ctx, dir, name)
|
||||
readDir, readName := filer.RemoteReadPath(entry, dir, name)
|
||||
client, remoteLocation, err := fs.mountedRemoteClient(ctx, readDir, readName)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -230,13 +231,14 @@ 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)
|
||||
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, dir, name, offset, size, statOrigin)
|
||||
return fs.streamFromRemoteOnly(ctx, r, readDir, readName, offset, size, statOrigin)
|
||||
}
|
||||
// Bounded wait: a large download outlasts any client timeout, so
|
||||
// serve straight from the origin once the wait expires while the
|
||||
@@ -264,7 +266,7 @@ func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request)
|
||||
// when a later prepare fails; keep those on the 503 retry path.
|
||||
if !strings.Contains(r.Header.Get("Range"), ",") {
|
||||
glog.V(1).InfofCtx(ctx, "stream %s from remote while caching: %v", entry.FullPath, err)
|
||||
if streamFn, remoteErr := fs.streamFromRemote(ctx, dir, name, offset, size); remoteErr == nil {
|
||||
if streamFn, remoteErr := fs.streamFromRemote(ctx, readDir, readName, offset, size); remoteErr == nil {
|
||||
return streamFn, nil
|
||||
} else {
|
||||
glog.WarningfCtx(ctx, "stream %s from remote: %v", entry.FullPath, remoteErr)
|
||||
|
||||
Reference in new issue
Block a user