fix(redis,scanner): atomic IncrWindow Lua, lock rand/ctx fixes, ScanLock renewal, single-flight, error logging (B1,B2,B3,B9,B10,B11)

2026-09-14 19:40:53 +08:00
parent 2c2f6d3116
commit 8fbc58eaaa
3 changed files with 223 additions and 21 deletions
@@ -44,18 +44,25 @@ func (r *R) Set(ctx context.Context, key, val string, ttl time.Duration) {
 	}
 }

+// incrWindowScript atomically increments and sets TTL on first value,
+// preventing the INCR+EXPIRE race that could leave keys without TTL (B1).
+var incrWindowScript = redis.NewScript(`
+local n = redis.call('INCR', KEYS[1])
+if n == 1 then
+	redis.call('EXPIRE', KEYS[1], ARGV[1])
+end
+return n
+`)
+
 func (r *R) IncrWindow(ctx context.Context, key string, ttl time.Duration) int {
 	if r.c == nil {
 		return 1
 	}
-	n, err := r.c.Incr(ctx, key).Result()
+	n, err := incrWindowScript.Run(ctx, r.c, []string{key}, int(ttl.Seconds())).Int()
 	if err != nil {
-		return 1
+		return 1 // fail-open
 	}
-	if n == 1 {
-		r.c.Expire(ctx, key, ttl)
-	}
-	return int(n)
+	return n
 }

 func (r *R) Lock(ctx context.Context, key string, ttl time.Duration) (func(), bool) {
@@ -64,7 +71,11 @@ func (r *R) Lock(ctx context.Context, key string, ttl time.Duration) (func(), bo
 		return noop, true
 	}
 	b := make([]byte, 8)
-	rand.Read(b)
+	if _, err := rand.Read(b); err != nil {
+		// B2: rand failure → degrade to no-lock instead of using a zero token.
+		log.Printf("rand.Read failed: %v (proceeding without lock)", err)
+		return noop, true
+	}
 	tok := hex.EncodeToString(b)
 	ok, err := r.c.SetNX(ctx, key, tok, ttl).Result()
 	if err != nil { // spec §9: Redis 故障降级放行,锁只做尽力去重
@@ -75,8 +86,68 @@ func (r *R) Lock(ctx context.Context, key string, ttl time.Duration) (func(), bo
 		return noop, false // 锁被持有,别的副本在扫
 	}
 	return func() {
-		r.c.Eval(ctx,
+		// B3: use WithoutCancel so unlock survives caller cancellation.
+		if err := r.c.Eval(context.WithoutCancel(ctx),
 			"if redis.call('get',KEYS[1])==ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end",
-			[]string{key}, tok)
+			[]string{key}, tok).Err(); err != nil {
+			log.Printf("redis unlock %s: %v", key, err)
+		}
 	}, true
 }
+
+// ScanLock acquires a distributed lock with automatic renewal.
+// The lock is renewed every ttl/2 until unlock is called.
+// Returns (unlock, true) on success, (noop, true) on redis failure (degrade),
+// or (noop, false) if the lock is already held.
+func (r *R) ScanLock(ctx context.Context, key string, ttl time.Duration) (func(), bool) {
+	noop := func() {}
+	if r.c == nil {
+		return noop, true
+	}
+	b := make([]byte, 8)
+	if _, err := rand.Read(b); err != nil {
+		log.Printf("rand.Read failed: %v (proceeding without lock)", err)
+		return noop, true
+	}
+	tok := hex.EncodeToString(b)
+	ok, err := r.c.SetNX(ctx, key, tok, ttl).Result()
+	if err != nil {
+		log.Printf("redis scanlock %s: %v (proceeding without lock)", key, err)
+		return noop, true
+	}
+	if !ok {
+		return noop, false
+	}
+
+	// Start renewal goroutine.
+	done := make(chan struct{})
+	go func() {
+		ticker := time.NewTicker(ttl / 2)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-done:
+				return
+			case <-ticker.C:
+				// Renew only if we still own the lock.
+				if err := r.c.Eval(context.Background(),
+					`if redis.call('get',KEYS[1])==ARGV[1] then
+						return redis.call('expire',KEYS[1],ARGV[2])
+					else return 0 end`,
+					[]string{key}, tok, int(ttl.Seconds())).Err(); err != nil {
+					log.Printf("redis scanlock renew %s: %v", key, err)
+				}
+			}
+		}
+	}()
+
+	unlock := func() {
+		close(done) // stop renewal
+		if err := r.c.Eval(context.WithoutCancel(ctx),
+			"if redis.call('get',KEYS[1])==ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end",
+			[]string{key}, tok).Err(); err != nil {
+			log.Printf("redis scanlock unlock %s: %v", key, err)
+		}
+	}
+	return unlock, true
+}
@@ -2,6 +2,7 @@ package redispkg

 import (
 	"context"
+	"os"
 	"testing"
 	"time"
 )
@@ -21,6 +22,12 @@ func TestDisabledIsSafe(t *testing.T) {
 		t.Fatal("disabled Lock must always acquire")
 	}
 	un()
+	// ScanLock disabled mode.
+	un2, ok2 := r.ScanLock(ctx, "slk", time.Second)
+	if !ok2 {
+		t.Fatal("disabled ScanLock must always acquire")
+	}
+	un2()
 }

 func TestDeadRedisLockFailsOpen(t *testing.T) {
@@ -31,3 +38,96 @@ func TestDeadRedisLockFailsOpen(t *testing.T) {
 	}
 	un()
 }
+
+func TestDeadRedisScanLockFailsOpen(t *testing.T) {
+	r := New("redis://127.0.0.1:16399")
+	un, ok := r.ScanLock(context.Background(), "slk", time.Second)
+	if !ok {
+		t.Fatal("ScanLock on redis error must fail open")
+	}
+	un()
+}
+
+func newLiveRedis(t *testing.T) *R {
+	t.Helper()
+	url := os.Getenv("REDIS_URL")
+	if url == "" {
+		t.Skip("REDIS_URL not set")
+	}
+	return New(url)
+}
+
+func TestIncrWindowSetsTTL(t *testing.T) {
+	r := newLiveRedis(t)
+	ctx := context.Background()
+	key := "test:incrwindow:ttl:" + t.Name()
+	r.c.Del(ctx, key)
+
+	n := r.IncrWindow(ctx, key, 5*time.Second)
+	if n != 1 {
+		t.Fatalf("first call = %d, want 1", n)
+	}
+	ttl, err := r.c.TTL(ctx, key).Result()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if ttl <= 0 {
+		t.Fatalf("TTL should be positive after first increment, got %v", ttl)
+	}
+
+	// Second call: n=2, TTL should still be positive.
+	n = r.IncrWindow(ctx, key, 5*time.Second)
+	if n != 2 {
+		t.Fatalf("second call = %d, want 2", n)
+	}
+	r.c.Del(ctx, key)
+}
+
+func TestScanLockRenewal(t *testing.T) {
+	r := newLiveRedis(t)
+	ctx := context.Background()
+	key := "test:scanlock:" + t.Name()
+	r.c.Del(ctx, key)
+
+	unlock, ok := r.ScanLock(ctx, key, 2*time.Second)
+	if !ok {
+		t.Fatal("should acquire")
+	}
+	// Wait 3 seconds — without renewal, the lock would expire at 2s.
+	time.Sleep(3 * time.Second)
+
+	// A second attempt should fail (lock still held by first, renewed).
+	_, ok2 := r.ScanLock(ctx, key, 2*time.Second)
+	if ok2 {
+		t.Fatal("second acquire should fail — lock should have been renewed")
+	}
+	unlock()
+	r.c.Del(ctx, key)
+}
+
+func TestScanLockMutualExclusion(t *testing.T) {
+	r := newLiveRedis(t)
+	ctx := context.Background()
+	key := "test:scanlock:mutex:" + t.Name()
+	r.c.Del(ctx, key)
+
+	unlock1, ok1 := r.ScanLock(ctx, key, 10*time.Second)
+	if !ok1 {
+		t.Fatal("first should acquire")
+	}
+
+	_, ok2 := r.ScanLock(ctx, key, 10*time.Second)
+	if ok2 {
+		t.Fatal("second should not acquire while first holds")
+	}
+
+	unlock1()
+	// After unlock, a new acquire should succeed.
+	time.Sleep(50 * time.Millisecond)
+	unlock3, ok3 := r.ScanLock(ctx, key, 10*time.Second)
+	if !ok3 {
+		t.Fatal("should acquire after unlock")
+	}
+	unlock3()
+	r.c.Del(ctx, key)
+}