mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-13 17:48:29 +02:00
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.
This commit is contained in:
1 parent
29f9d0b805
commit
5882fb4af0
3 files changed
+77
-10
No files matched your search
@@ -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
|
||||
|
||||
@@ -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)
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user