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 }