diff --git a/weed/filer/filer_lazy_remote.go b/weed/filer/filer_lazy_remote.go index 6d8cce3593..ba43bdcc50 100644 --- a/weed/filer/filer_lazy_remote.go +++ b/weed/filer/filer_lazy_remote.go @@ -438,3 +438,17 @@ func (f *Filer) RemoteCopyLocations(ctx context.Context, entry *Entry, oldPath, dst = MapFullPathToRemoteStorageLocation(mountDir, remoteLoc, newPath) return src, dst, copier, nil } + +// PrefetchRemoteEntries runs the lazy remote fetch for the paths a store +// transaction is about to read. Inside the transaction the fetch is off, +// so an object the filer has not seen yet would read as absent there. +func (f *Filer) PrefetchRemoteEntries(ctx context.Context, paths ...util.FullPath) { + if ctx.Value("tx") != nil { + return + } + for _, p := range paths { + if _, err := f.FindEntry(ctx, p); err != nil && !errors.Is(err, filer_pb.ErrNotFound) { + glog.V(1).InfofCtx(ctx, "prefetch remote entry %s: %v", p, err) + } + } +} diff --git a/weed/filer/filer_lazy_remote_test.go b/weed/filer/filer_lazy_remote_test.go index b1d4c6c436..d366c9cae7 100644 --- a/weed/filer/filer_lazy_remote_test.go +++ b/weed/filer/filer_lazy_remote_test.go @@ -756,6 +756,39 @@ func TestMaybeLazyFetchFromRemote_SkipsInsideStoreTransaction(t *testing.T) { assert.NotZero(t, countingStub.statCalls, "the same lookup outside a transaction consults the remote") } +func TestPrefetchRemoteEntries_MaterializesBeforeATransaction(t *testing.T) { + const storageType = "stub_lazy_prefetch" + countingStub := &countingRemoteClient{ + stubRemoteClient: stubRemoteClient{ + statResult: &filer_pb.RemoteEntry{RemoteMtime: 1, RemoteSize: 1}, + }, + } + f, store := newMountedTestFiler(t, storageType, countingStub, 0) + putDirEntry(store, "/buckets/mybucket/dir") + target := util.FullPath("/buckets/mybucket/dir/target.txt") + txCtx := context.WithValue(context.Background(), "tx", struct{}{}) + + // unseen on the filer, present on the remote: inside a transaction it + // reads as absent + _, err := f.FindEntry(txCtx, target) + require.ErrorIs(t, err, filer_pb.ErrNotFound) + assert.Equal(t, 0, countingStub.statCalls) + + // the prefetch outside the transaction materializes it, and the + // transaction then finds it in the store + f.PrefetchRemoteEntries(context.Background(), target, "/buckets/mybucket/dir/absent.txt") + assert.NotZero(t, countingStub.statCalls) + entry, err := f.FindEntry(txCtx, target) + require.NoError(t, err) + require.NotNil(t, entry) + assert.True(t, entry.IsInRemoteOnly()) + + // a prefetch that itself runs inside a transaction does nothing + before := countingStub.statCalls + f.PrefetchRemoteEntries(txCtx, "/buckets/mybucket/dir/other.txt") + assert.Equal(t, before, countingStub.statCalls) +} + func TestMaybeLazyFetchFromRemote_GuardedClientRejectsEndpoint(t *testing.T) { const storageType = "stub_lazy_guarded" countingStub := &countingRemoteClient{ diff --git a/weed/server/filer_grpc_server_rename.go b/weed/server/filer_grpc_server_rename.go index 62097e478d..68ed0c6fe7 100644 --- a/weed/server/filer_grpc_server_rename.go +++ b/weed/server/filer_grpc_server_rename.go @@ -47,6 +47,10 @@ func (fs *FilerServer) AtomicRenameEntry(ctx context.Context, req *filer_pb.Atom defer fs.acquireRenamePathLocks("AtomicRenameEntry", oldParent.Child(req.OldName), newParent.Child(req.NewName))() + // the lazy remote fetch is off inside the transaction: a source or + // target the filer only holds on the remote is materialized first + fs.filer.PrefetchRemoteEntries(ctx, oldParent.Child(req.OldName), newParent.Child(req.NewName)) + // a SQL store keeps the transaction in the context and refuses calls // through it once committed; the work after the commit uses this one afterCommitCtx := ctx @@ -118,6 +122,7 @@ func (fs *FilerServer) StreamRenameEntry(req *filer_pb.StreamRenameEntryRequest, defer fs.acquireRenamePathLocks("StreamRenameEntry", oldParent.Child(req.OldName), newParent.Child(req.NewName))() ctx := context.Background() + fs.filer.PrefetchRemoteEntries(ctx, oldParent.Child(req.OldName), newParent.Child(req.NewName)) afterCommitCtx := ctx ctx, err = fs.filer.BeginTransaction(ctx)