From e549d732e65d9372853f94fc9c1c0198824b8740 Mon Sep 17 00:00:00 2001 From: MontaGhanmy Date: Thu, 10 Sep 2026 12:24:37 +0100 Subject: [PATCH] fix(lock): stop renewing a lease that cannot be extended Return renewal failures from Extend and stop the renewer after logging its first failure. Keep the ownership-loss error internal. Cover ownership loss and Redis errors with deterministic virtual-time tests. Shutdown on Unlock is handled in a separate change. --- pkg/lock/lock.go | 13 +++++-- pkg/lock/lock_test.go | 12 +++++-- pkg/lock/long_operation_test.go | 62 +++++++++++++++++++++++++++++++++ pkg/lock/simple_mem.go | 10 +++--- pkg/lock/simple_redis.go | 11 ++++-- 5 files changed, 97 insertions(+), 11 deletions(-) create mode 100644 pkg/lock/long_operation_test.go diff --git a/pkg/lock/lock.go b/pkg/lock/lock.go index f525f781000..7545059136a 100644 --- a/pkg/lock/lock.go +++ b/pkg/lock/lock.go @@ -1,9 +1,11 @@ package lock import ( + "errors" "sync" "time" + "github.com/cozy/cozy-stack/pkg/logger" "github.com/cozy/cozy-stack/pkg/prefixer" "github.com/redis/go-redis/v9" ) @@ -43,9 +45,12 @@ type ErrorRWLocker interface { type longOperationLocker interface { ErrorLocker - Extend() + Extend() error } +// errLockLost means that an operation no longer owns its distributed lock. +var errLockLost = errors.New("lock ownership lost") + type longOperation struct { lock longOperationLocker mu sync.Mutex @@ -72,7 +77,11 @@ func (l *longOperation) Lock() error { if l.tick == nil { return } - l.lock.Extend() + if err := l.lock.Extend(); err != nil { + logger.WithNamespace("lock"). + Warnf("cannot extend a long operation lease: %s", err) + return + } l.mu.Unlock() } }() diff --git a/pkg/lock/lock_test.go b/pkg/lock/lock_test.go index 3807b8fc1a0..7b27a0a32ff 100644 --- a/pkg/lock/lock_test.go +++ b/pkg/lock/lock_test.go @@ -9,6 +9,7 @@ package lock import ( "flag" "fmt" + "os" "runtime" "sync/atomic" "testing" @@ -24,6 +25,13 @@ import ( // slow for CI, and the lock package has very few commits in the last years. var nb = 100 +func redisTestURL() string { + if raw := os.Getenv("COZY_TEST_REDIS_URL"); raw != "" { + return raw + } + return "redis://localhost:6379/0" +} + func TestLock(t *testing.T) { if testing.Short() { t.Skip("an instance is required for this test: test skipped due to the use of --short flag") @@ -43,7 +51,7 @@ func TestLock(t *testing.T) { }) t.Run("RedisLock", func(t *testing.T) { - opt, err := redis.ParseURL("redis://localhost:6379/0") + opt, err := redis.ParseURL(redisTestURL()) require.NoError(t, err) client := NewRedisLockGetter(redis.NewClient(opt)) @@ -82,7 +90,7 @@ func TestLock(t *testing.T) { return } - opt, err := redis.ParseURL("redis://localhost:6379/0") + opt, err := redis.ParseURL(redisTestURL()) require.NoError(t, err) client := NewRedisLockGetter(redis.NewClient(opt)) diff --git a/pkg/lock/long_operation_test.go b/pkg/lock/long_operation_test.go new file mode 100644 index 00000000000..a96a7d733cb --- /dev/null +++ b/pkg/lock/long_operation_test.go @@ -0,0 +1,62 @@ +package lock + +import ( + "errors" + "sync" + "testing" + "testing/synctest" + "time" + + "github.com/stretchr/testify/require" +) + +type leaseStub struct { + mu sync.Mutex + err error + renewals int +} + +func (s *leaseStub) Lock() error { return nil } +func (s *leaseStub) Unlock() {} +func (s *leaseStub) Extend() error { + s.mu.Lock() + defer s.mu.Unlock() + s.renewals++ + return s.err +} + +func (s *leaseStub) count() int { + s.mu.Lock() + defer s.mu.Unlock() + return s.renewals +} + +func TestLongOperationRenewsUntilItCannot(t *testing.T) { + for name, err := range map[string]error{ + "lost ownership": errLockLost, + "redis error": errors.New("redis unavailable"), + } { + t.Run(name, func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + s := &leaseStub{} + l := &longOperation{lock: s, timeout: 3 * time.Second} + require.NoError(t, l.Lock()) + defer l.Unlock() + synctest.Wait() + time.Sleep(time.Second) + synctest.Wait() + require.Equal(t, 1, s.count()) + + s.mu.Lock() + s.err = err + s.mu.Unlock() + time.Sleep(time.Second) + synctest.Wait() + require.Equal(t, 2, s.count()) + time.Sleep(2 * time.Second) + synctest.Wait() + require.Equal(t, 2, s.count()) + }) + }) + } +} diff --git a/pkg/lock/simple_mem.go b/pkg/lock/simple_mem.go index 9d0366d6b04..3328f8040e1 100644 --- a/pkg/lock/simple_mem.go +++ b/pkg/lock/simple_mem.go @@ -33,8 +33,8 @@ type memLock struct { sync.RWMutex } -func (ml *memLock) Lock() error { ml.RWMutex.Lock(); return nil } -func (ml *memLock) RLock() error { ml.RWMutex.RLock(); return nil } -func (ml *memLock) Extend() {} -func (ml *memLock) Unlock() { ml.RWMutex.Unlock() } -func (ml *memLock) RUnlock() { ml.RWMutex.RUnlock() } +func (ml *memLock) Lock() error { ml.RWMutex.Lock(); return nil } +func (ml *memLock) RLock() error { ml.RWMutex.RLock(); return nil } +func (ml *memLock) Extend() error { return nil } +func (ml *memLock) Unlock() { ml.RWMutex.Unlock() } +func (ml *memLock) RUnlock() { ml.RWMutex.RUnlock() } diff --git a/pkg/lock/simple_redis.go b/pkg/lock/simple_redis.go index 0d5b2897bfa..1427215741d 100644 --- a/pkg/lock/simple_redis.go +++ b/pkg/lock/simple_redis.go @@ -120,10 +120,17 @@ func (rl *redisLock) Lock() error { } } -func (rl *redisLock) Extend() { +func (rl *redisLock) Extend() error { rl.mu.Lock() defer rl.mu.Unlock() - _, _ = rl.extends() + ok, err := rl.extends() + if err != nil { + return err + } + if !ok { + return errLockLost + } + return nil } func (rl *redisLock) RLock() error {