From 5882fb4af01beb5a13030a6e03339f4c26dee3cd Mon Sep 17 00:00:00 2001 From: Alex Russell Date: Mon, 12 Oct 2026 05:11:50 +0200 Subject: [PATCH] filer: finish a rename's remote work outside the committed transaction The SQL stores keep the transaction in the context and refuse every call through it once it has committed. The work that runs after CommitTransaction used that context, so on postgres2 the deferred old-key delete could not look up the old path and kept the object, and finishRemoteCopies could not read or stamp the entry. Both failed with "sql: transaction has already been committed or rolled back" on every copied rename; remote.sync cleaned up after them. Keep the context from before the transaction for the work after the commit. The rename test store gets a strict mode that fails calls through a committed transaction the way database/sql does. --- weed/server/filer_grpc_server_rename.go | 12 +++-- .../filer_grpc_server_rename_remote_test.go | 24 +++++++++ weed/server/filer_grpc_server_rename_test.go | 51 ++++++++++++++++--- 3 files changed, 77 insertions(+), 10 deletions(-) diff --git a/weed/server/filer_grpc_server_rename.go b/weed/server/filer_grpc_server_rename.go index bdafe6ce5..af5123d14 100644 --- a/weed/server/filer_grpc_server_rename.go +++ b/weed/server/filer_grpc_server_rename.go @@ -47,6 +47,9 @@ func (fs *FilerServer) AtomicRenameEntry(ctx context.Context, req *filer_pb.Atom defer fs.acquireRenamePathLocks("AtomicRenameEntry", 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 ctx, err := fs.filer.BeginTransaction(ctx) if err != nil { return nil, err @@ -78,7 +81,7 @@ func (fs *FilerServer) AtomicRenameEntry(ctx context.Context, req *filer_pb.Atom if len(pendingChunkDeletes) > 0 { fs.filer.DeleteChunksNotRecursive(pendingChunkDeletes) } - fs.finishRemoteCopies(ctx, pendingRemoteCopies) + fs.finishRemoteCopies(afterCommitCtx, pendingRemoteCopies) for i := range metadataEvents { event := &metadataEvents[i] for _, pending := range pendingRemoteCopies { @@ -86,7 +89,7 @@ func (fs *FilerServer) AtomicRenameEntry(ctx context.Context, req *filer_pb.Atom event.deleteOldRemote = false } } - event.notify(fs.filer, ctx, req.Signatures) + event.notify(fs.filer, afterCommitCtx, req.Signatures) } return &filer_pb.AtomicRenameEntryResponse{}, nil @@ -107,6 +110,7 @@ func (fs *FilerServer) StreamRenameEntry(req *filer_pb.StreamRenameEntryRequest, ctx := context.Background() + afterCommitCtx := ctx ctx, err = fs.filer.BeginTransaction(ctx) if err != nil { return err @@ -149,7 +153,7 @@ func (fs *FilerServer) StreamRenameEntry(req *filer_pb.StreamRenameEntryRequest, if len(pendingChunkDeletes) > 0 { fs.filer.DeleteChunksNotRecursive(pendingChunkDeletes) } - fs.finishRemoteCopies(ctx, pendingRemoteCopies) + fs.finishRemoteCopies(afterCommitCtx, pendingRemoteCopies) for i := range metadataEvents { event := &metadataEvents[i] for _, pending := range pendingRemoteCopies { @@ -157,7 +161,7 @@ func (fs *FilerServer) StreamRenameEntry(req *filer_pb.StreamRenameEntryRequest, event.deleteOldRemote = false } } - event.notify(fs.filer, ctx, req.Signatures) + event.notify(fs.filer, afterCommitCtx, req.Signatures) } return nil diff --git a/weed/server/filer_grpc_server_rename_remote_test.go b/weed/server/filer_grpc_server_rename_remote_test.go index dd448633a..e25d33610 100644 --- a/weed/server/filer_grpc_server_rename_remote_test.go +++ b/weed/server/filer_grpc_server_rename_remote_test.go @@ -315,3 +315,27 @@ func TestDeferredOldKeyDeleteKeepsAKeyRetakenSince(t *testing.T) { event.notify(server.filer, context.Background(), nil) assert.Equal(t, []string{"delete origin/src/a.jpg"}, client.recorded()) } + +func TestRenameRemoteOnlyEntryFinishesAfterTheCommitOnASQLStore(t *testing.T) { + // a SQL store keeps the transaction in the context and refuses calls + // through it once committed, so the work after the commit must not use it + for _, overTarget := range []bool{false, true} { + t.Run(map[bool]string{false: "new path", true: "over a remote-only target"}[overTarget], func(t *testing.T) { + client := copyingRenameRemoteClient{&renameRemoteClient{}} + server, store := newRenameRemoteTestServer(t, client) + store.strictTx = true + store.entries["/buckets/b/src/a.jpg"] = remoteOnlyRenameEntry("/buckets/b/src/a.jpg", 101) + if overTarget { + store.entries["/buckets/b/dst/a.jpg"] = remoteOnlyRenameEntry("/buckets/b/dst/a.jpg", 202) + } + + require.NoError(t, renameFile(server, "/buckets/b/src/a.jpg", "/buckets/b/dst/a.jpg")) + + // the copy is stamped and the old key deleted by the filer itself + assert.Equal(t, []string{"copy origin/src/a.jpg origin/dst/a.jpg", "delete origin/src/a.jpg"}, client.recorded()) + dst, err := store.FindEntry(context.Background(), "/buckets/b/dst/a.jpg") + require.NoError(t, err) + assert.Equal(t, `"copy"`, dst.Remote.RemoteETag) + }) + } +} diff --git a/weed/server/filer_grpc_server_rename_test.go b/weed/server/filer_grpc_server_rename_test.go index 71fc92e01..1a36bc45d 100644 --- a/weed/server/filer_grpc_server_rename_test.go +++ b/weed/server/filer_grpc_server_rename_test.go @@ -32,6 +32,20 @@ type renameTestStore struct { listErr error // simulates a transient store/RPC failure from a directory listing findDelay time.Duration // optional: widen check-then-act windows in tests findErr error // simulates a failing entry lookup + strictTx bool // like a SQL store: calls through a committed transaction fail +} + +type renameTestTxKey struct{} + +type renameTestTx struct{ done bool } + +// txDone reports a call through a transaction that has already committed or +// rolled back, which a SQL store refuses. +func (s *renameTestStore) txDone(ctx context.Context) error { + if tx, ok := ctx.Value(renameTestTxKey{}).(*renameTestTx); ok && tx.done { + return errors.New("sql: transaction has already been committed or rolled back") + } + return nil } func newRenameTestStore() *renameTestStore { @@ -45,10 +59,23 @@ func (s *renameTestStore) GetName() string { return func (s *renameTestStore) Initialize(util.Configuration, string) error { return nil } func (s *renameTestStore) Shutdown() {} func (s *renameTestStore) BeginTransaction(ctx context.Context) (context.Context, error) { + if s.strictTx { + return context.WithValue(ctx, renameTestTxKey{}, &renameTestTx{}), nil + } return ctx, nil } -func (s *renameTestStore) CommitTransaction(context.Context) error { return s.commitErr } -func (s *renameTestStore) RollbackTransaction(context.Context) error { return nil } +func (s *renameTestStore) CommitTransaction(ctx context.Context) error { + if tx, ok := ctx.Value(renameTestTxKey{}).(*renameTestTx); ok { + tx.done = true + } + return s.commitErr +} +func (s *renameTestStore) RollbackTransaction(ctx context.Context) error { + if tx, ok := ctx.Value(renameTestTxKey{}).(*renameTestTx); ok { + tx.done = true + } + return nil +} func (s *renameTestStore) KvPut(context.Context, []byte, []byte) error { return nil } @@ -57,21 +84,30 @@ func (s *renameTestStore) KvGet(context.Context, []byte) ([]byte, error) { } func (s *renameTestStore) KvDelete(context.Context, []byte) error { return nil } -func (s *renameTestStore) InsertEntry(_ context.Context, entry *filer.Entry) error { +func (s *renameTestStore) InsertEntry(ctx context.Context, entry *filer.Entry) error { + if err := s.txDone(ctx); err != nil { + return err + } s.mu.Lock() defer s.mu.Unlock() s.entries[string(entry.FullPath)] = entry.ShallowClone() return nil } -func (s *renameTestStore) UpdateEntry(_ context.Context, entry *filer.Entry) error { +func (s *renameTestStore) UpdateEntry(ctx context.Context, entry *filer.Entry) error { + if err := s.txDone(ctx); err != nil { + return err + } s.mu.Lock() defer s.mu.Unlock() s.entries[string(entry.FullPath)] = entry.ShallowClone() return nil } -func (s *renameTestStore) FindEntry(_ context.Context, p util.FullPath) (*filer.Entry, error) { +func (s *renameTestStore) FindEntry(ctx context.Context, p util.FullPath) (*filer.Entry, error) { + if err := s.txDone(ctx); err != nil { + return nil, err + } if s.findDelay > 0 { time.Sleep(s.findDelay) } @@ -88,7 +124,10 @@ func (s *renameTestStore) FindEntry(_ context.Context, p util.FullPath) (*filer. return entry.ShallowClone(), nil } -func (s *renameTestStore) DeleteEntry(_ context.Context, p util.FullPath) error { +func (s *renameTestStore) DeleteEntry(ctx context.Context, p util.FullPath) error { + if err := s.txDone(ctx); err != nil { + return err + } if s.deleteErr != nil { return s.deleteErr }