diff --git a/weed/filer/filer.go b/weed/filer/filer.go index bae29d10bf..11d5427b4c 100644 --- a/weed/filer/filer.go +++ b/weed/filer/filer.go @@ -516,7 +516,16 @@ func (f *Filer) CreateEntry(ctx context.Context, entry *Entry, existing *Entry, oldEntry := existing if oldEntry == nil { var findErr error - oldEntry, findErr = f.FindEntry(ctx, entry.FullPath) + // Inside a store transaction the lookup must not run the lazy remote + // fetch: the fetch persists on a second connection, which waits on + // the row lock this transaction holds when it deleted the path (a + // rename onto an existing target). The create writes the path either + // way; reads inside the transaction keep the fetch. + findCtx := ctx + if ctx.Value("tx") != nil { + findCtx = context.WithValue(ctx, lazyFetchContextKey{}, true) + } + oldEntry, findErr = f.FindEntry(findCtx, entry.FullPath) if findErr != nil && !errors.Is(findErr, filer_pb.ErrNotFound) { // A create cannot decide whether the path exists when the lookup // itself failed; proceeding would upsert over it. diff --git a/weed/filer/filer_lazy_remote.go b/weed/filer/filer_lazy_remote.go index ba43bdcc50..81d4e3996a 100644 --- a/weed/filer/filer_lazy_remote.go +++ b/weed/filer/filer_lazy_remote.go @@ -41,12 +41,6 @@ func (f *Filer) maybeLazyFetchFromRemote(ctx context.Context, p util.FullPath) ( return nil, nil } - // A lookup inside a store transaction persists on a second connection, - // which waits on the row locks the transaction holds. - if ctx.Value("tx") != nil { - return nil, nil - } - if f.RemoteStorage == nil { return nil, nil } @@ -438,17 +432,3 @@ 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 d366c9cae7..7c59b8384a 100644 --- a/weed/filer/filer_lazy_remote_test.go +++ b/weed/filer/filer_lazy_remote_test.go @@ -734,59 +734,34 @@ func TestMaybeLazyFetchFromRemote_ContextGuardPreventsRecursion(t *testing.T) { assert.Equal(t, 0, countingStub.statCalls, "guard should prevent StatFile from being called") } -func TestMaybeLazyFetchFromRemote_SkipsInsideStoreTransaction(t *testing.T) { +func TestCreateEntry_SkipsTheLazyFetchInsideAStoreTransaction(t *testing.T) { const storageType = "stub_lazy_tx" countingStub := &countingRemoteClient{ stubRemoteClient: stubRemoteClient{ statResult: &filer_pb.RemoteEntry{RemoteMtime: 1, RemoteSize: 1}, }, } - f, _ := newMountedTestFiler(t, storageType, countingStub, 0) - - // the key a store transaction travels under (abstract_sql getTxOrDB) - txCtx := context.WithValue(context.Background(), "tx", struct{}{}) - entry, err := f.maybeLazyFetchFromRemote(txCtx, "/buckets/mybucket/dir/a.txt") - require.NoError(t, err) - assert.Nil(t, entry) - assert.Equal(t, 0, countingStub.statCalls, "a lookup inside a transaction must not consult the remote") - - entry, err = f.maybeLazyFetchFromRemote(context.Background(), "/buckets/mybucket/dir/a.txt") - require.NoError(t, err) - require.NotNil(t, entry) - 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") + // the key a store transaction travels under (abstract_sql getTxOrDB) 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) + // a create inside the transaction: its lookup does not consult the + // remote, the entry is inserted as given + created := &Entry{FullPath: "/buckets/mybucket/dir/a.txt", Attr: Attr{Mode: 0644, Mtime: time.Now(), Crtime: time.Now()}} + require.NoError(t, f.CreateEntry(txCtx, created, nil, false, false, nil, false, f.MaxFilenameLength)) + assert.Equal(t, 0, countingStub.statCalls, "a create inside a transaction must not consult the remote") + got, err := f.FindEntry(txCtx, "/buckets/mybucket/dir/a.txt") + require.NoError(t, err) + assert.False(t, got.IsInRemoteOnly()) - // 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) + // a read inside the transaction keeps the fetch: a target the filer only + // holds on the remote is still seen by a rename's target check + entry, err := f.FindEntry(txCtx, "/buckets/mybucket/dir/b.txt") 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) + assert.NotZero(t, countingStub.statCalls) } func TestMaybeLazyFetchFromRemote_GuardedClientRejectsEndpoint(t *testing.T) { diff --git a/weed/server/filer_grpc_server_rename.go b/weed/server/filer_grpc_server_rename.go index 68ed0c6fe7..62097e478d 100644 --- a/weed/server/filer_grpc_server_rename.go +++ b/weed/server/filer_grpc_server_rename.go @@ -47,10 +47,6 @@ 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 @@ -122,7 +118,6 @@ 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)