fix(lock): stop renewing a lease that cannot be extended

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()
 		}
 	}()

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))
 

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())
+			})
+		})
+	}
+}

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() }

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 {