Support redis lock automatic renewal

README.md

@@ -31,7 +31,7 @@ func main() {
 		Addr: "localhost:6379",
 	}
 	redisClient := redis.NewClient(redisOptions)
-	locker, err := redislock.NewRedisLocker(redisClient, redislock.WithTries(1))
+	locker, err := redislock.NewRedisLocker(redisClient,time.Second * 2, redislock.WithTries(1))
 	if err != nil {
 		// handle the error
 	}
@@ -71,7 +71,7 @@ func main() {
 		Addrs: []string{"localhost:6379"},
 	}
 	redisClient := redis.NewUniversalClient(redisOptions)
-	locker, err := redislock.NewRedisLocker(redisClient, redislock.WithTries(1))
+	locker, err := redislock.NewRedisLocker(redisClient,time.Second * 2, redislock.WithTries(1))
 	if err != nil {
 		// handle the error
 	}

redislock.go

@@ -3,6 +3,7 @@ package redislock
 import (
 	"context"
 	"fmt"
+	"time"
 
 	"github.com/go-co-op/gocron"
 	"github.com/go-redsync/redsync/v4"
@@ -24,30 +25,31 @@ var (
 
 // NewRedisLocker provides an implementation of the Locker interface using
 // redis for storage.
-func NewRedisLocker(r redis.UniversalClient, options ...redsync.Option) (gocron.Locker, error) {
+func NewRedisLocker(r redis.UniversalClient, autoExtendDuration time.Duration, options ...redsync.Option) (gocron.Locker, error) {
 	if err := r.Ping(context.Background()).Err(); err != nil {
 		return nil, fmt.Errorf("%s: %w", gocron.ErrFailedToConnectToRedis, err)
 	}
-	return newLocker(r, options...), nil
+	return newLocker(r, autoExtendDuration, options...), nil
 }
 
 // NewRedisLockerAlways provides an implementation of the Locker interface using
 // redis for storage, even if the connection fails.
-func NewRedisLockerAlways(r redis.UniversalClient, options ...redsync.Option) (gocron.Locker, error) {
-	return newLocker(r, options...), r.Ping(context.Background()).Err()
+func NewRedisLockerAlways(r redis.UniversalClient, autoExtendDuration time.Duration, options ...redsync.Option) (gocron.Locker, error) {
+	return newLocker(r, autoExtendDuration, options...), r.Ping(context.Background()).Err()
 }
 
-func newLocker(r redis.UniversalClient, options ...redsync.Option) gocron.Locker {
+func newLocker(r redis.UniversalClient, autoExtendDuration time.Duration, options ...redsync.Option) gocron.Locker {
 	pool := goredis.NewPool(r)
 	rs := redsync.New(pool)
-	return &redisLocker{rs: rs, options: options}
+	return &redisLocker{rs: rs, autoExtendDuration: autoExtendDuration, options: options}
 }
 
 var _ gocron.Locker = (*redisLocker)(nil)
 
 type redisLocker struct {
-	rs      *redsync.Redsync
-	options []redsync.Option
+	rs                 *redsync.Redsync
+	options            []redsync.Option
+	autoExtendDuration time.Duration
 }
 
 func (r *redisLocker) Lock(ctx context.Context, key string) (gocron.Lock, error) {
@@ -57,18 +59,27 @@ func (r *redisLocker) Lock(ctx context.Context, key string) (gocron.Lock, error)
 		return nil, gocron.ErrFailedToObtainLock
 	}
 	rl := &redisLock{
-		mu: mu,
+		mu:                 mu,
+		autoExtendDuration: r.autoExtendDuration,
+		done:               make(chan struct{}),
+	}
+
+	if r.autoExtendDuration > 0 {
+		go func() { rl.doExtend() }()
 	}
 	return rl, nil
 }
 
 var _ gocron.Lock = (*redisLock)(nil)
 
 type redisLock struct {
-	mu *redsync.Mutex
+	mu                 *redsync.Mutex
+	done               chan struct{}
+	autoExtendDuration time.Duration
 }
 
 func (r *redisLock) Unlock(ctx context.Context) error {
+	close(r.done)
 	unlocked, err := r.mu.UnlockContext(ctx)
 	if err != nil {
 		return gocron.ErrFailedToReleaseLock
@@ -79,3 +90,18 @@ func (r *redisLock) Unlock(ctx context.Context) error {
 
 	return nil
 }
+
+func (r *redisLock) doExtend() {
+	ticker := time.NewTicker(r.autoExtendDuration)
+	for {
+		select {
+		case <-r.done:
+			return
+		case <-ticker.C:
+			_, err := r.mu.Extend()
+			if err != nil {
+				return
+			}
+		}
+	}
+}

redislock_test.go

@@ -32,7 +32,7 @@ func TestEnableDistributedLocking(t *testing.T) {
 	}
 
 	redisClient := redis.NewClient(&redis.Options{Addr: strings.TrimPrefix(uri, "redis://")})
-	l, err := NewRedisLocker(redisClient, WithTries(1))
+	l, err := NewRedisLocker(redisClient, 0, WithTries(1))
 	require.NoError(t, err)
 
 	s1 := gocron.NewScheduler(time.UTC)
@@ -67,3 +67,44 @@ func TestEnableDistributedLocking(t *testing.T) {
 	}
 	assert.Len(t, results, 4)
 }
+
+func TestAutoExtend(t *testing.T) {
+	ctx := context.Background()
+	redisContainer, err := testcontainersredis.RunContainer(ctx)
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		if err := redisContainer.Terminate(ctx); err != nil {
+			t.Fatalf("failed to terminate container: %s", err)
+		}
+	})
+
+	uri, err := redisContainer.ConnectionString(ctx)
+	require.NoError(t, err)
+
+	redisClient := redis.NewClient(&redis.Options{Addr: strings.TrimPrefix(uri, "redis://")})
+	l1, err := NewRedisLocker(redisClient, 0, WithTries(1))
+	_, err = l1.Lock(ctx, "test1")
+	require.NoError(t, err)
+
+	t.Logf("waiting 9 seconds for lock to expire")
+	// wait for the lock to expire
+	time.Sleep(9 * time.Second)
+
+	_, err = l1.Lock(ctx, "test1")
+	require.NoError(t, err)
+
+	// create auto extend lock
+	l2, err := NewRedisLocker(redisClient, time.Second*2, WithTries(1))
+	unlocker, err := l2.Lock(ctx, "test2")
+	require.NoError(t, err)
+
+	t.Log("waiting 9 seconds for lock to expire")
+	// wait for the lock to expire
+	time.Sleep(9 * time.Second)
+
+	_, err = l2.Lock(ctx, "test2")
+	require.Equal(t, gocron.ErrFailedToObtainLock, err)
+
+	err = unlocker.Unlock(ctx)
+	require.NoError(t, err)
+}