feat: shorten rueidislock validity if there is a shorter deadline in …

…the context

Signed-off-by: Rueian <[email protected]>

rueidislock/lock.go

@@ -151,18 +151,18 @@ func keyname(prefix, name string, i int32) string {
 	return sb.String()
 }
 
-func (m *locker) acquire(ctx context.Context, key, val string, deadline time.Time, force bool) (err error) {
+func (m *locker) acquire(ctx context.Context, key, val string, duration time.Duration, deadline time.Time, force bool) (err error) {
 	ctx, cancel := context.WithTimeout(ctx, m.timeout)
 	var resp rueidis.RedisResult
 	if force {
 		if m.setpx {
-			resp = fcqms.Exec(ctx, m.client, []string{key}, []string{val, strconv.FormatInt(m.validity.Milliseconds(), 10)})
+			resp = fcqms.Exec(ctx, m.client, []string{key}, []string{val, strconv.FormatInt(duration.Milliseconds(), 10)})
 		} else {
 			resp = fcqat.Exec(ctx, m.client, []string{key}, []string{val, strconv.FormatInt(deadline.UnixMilli(), 10)})
 		}
 	} else {
 		if m.setpx {
-			resp = acqms.Exec(ctx, m.client, []string{key}, []string{val, strconv.FormatInt(m.validity.Milliseconds(), 10)})
+			resp = acqms.Exec(ctx, m.client, []string{key}, []string{val, strconv.FormatInt(duration.Milliseconds(), 10)})
 		} else {
 			resp = acqat.Exec(ctx, m.client, []string{key}, []string{val, strconv.FormatInt(deadline.UnixMilli(), 10)})
 		}
@@ -249,8 +249,15 @@ func (m *locker) try(ctx context.Context, cancel context.CancelFunc, name string
 	var err error
 
 	val := random()
-	deadline := time.Now().Add(m.validity)
-	cacneltm := time.AfterFunc(m.validity, cancel)
+	now := time.Now()
+	duration := m.validity
+	if dl, ok := ctx.Deadline(); ok {
+		if dur := dl.Sub(now); dur < duration {
+			duration = dur
+		}
+	}
+	deadline := now.Add(duration)
+	cacneltm := time.AfterFunc(duration, cancel)
 	released := int32(0)
 	acquired := int32(0)
 	failures := int32(0)
@@ -266,19 +273,12 @@ func (m *locker) try(ctx context.Context, cancel context.CancelFunc, name string
 					deadline = deadline.Add(m.interval)
 					if err = m.script(ctx, extend, key, val, deadline); err == nil {
 						timer.Reset(m.interval)
-						if !m.noloop {
-							<-csc
-						}
 					}
 				case _, ok := <-csc:
 					if !ok {
 						err = ErrLockerClosed
 					} else {
-						if err = m.script(ctx, extend, key, val, deadline); err == nil {
-							if !m.noloop {
-								<-csc
-							}
-						}
+						err = m.script(ctx, extend, key, val, deadline)
 					}
 				}
 			}
@@ -314,7 +314,7 @@ func (m *locker) try(ctx context.Context, cancel context.CancelFunc, name string
 		default:
 		}
 		if !errors.Is(err, ErrNotLocked) {
-			if err = m.acquire(ctx, key, val, deadline, force); force && err == nil {
+			if err = m.acquire(ctx, key, val, duration, deadline, force); force && err == nil {
 				m.mu.RLock()
 				if m.gates != nil {
 					select {

rueidislock/lock_test.go

@@ -346,6 +346,55 @@ func TestLocker_WithContext_ExtendByClientSideCaching(t *testing.T) {
 	})
 }
 
+func TestLocker_WithContext_AutoExtendConcurrent(t *testing.T) {
+	test := func(t *testing.T, noLoop, setpx, nocsc bool) {
+		locker := newLocker(t, noLoop, setpx, nocsc)
+		locker.validity = time.Second
+		locker.interval = time.Second / 2
+		defer locker.Close()
+
+		key := strconv.Itoa(rand.Int())
+
+		ctx1, cancel1, err1 := locker.WithContext(context.Background(), key)
+		if err1 != nil {
+			t.Fatal(err1)
+		}
+		go func() {
+			for i := 0; i < 4; i++ {
+				select {
+				case <-ctx1.Done():
+					t.Fatalf("unexpected context canceled %v", ctx1.Err())
+				default:
+					time.Sleep(locker.validity)
+				}
+			}
+			cancel1()
+		}()
+		ctx2, cancel2, err2 := locker.WithContext(context.Background(), key)
+		if err2 != nil {
+			t.Fatal(err2)
+		}
+		if !errors.Is(ctx1.Err(), context.Canceled) {
+			t.Fatalf("unexpected context canceled %v", ctx1.Err())
+		}
+		if ctx2.Err() != nil {
+			t.Fatalf("unexpected context canceled %v", ctx2.Err())
+		}
+		cancel2()
+	}
+	for _, nocsc := range []bool{false, true} {
+		t.Run("Tracking Loop", func(t *testing.T) {
+			test(t, false, false, nocsc)
+		})
+		t.Run("Tracking NoLoop", func(t *testing.T) {
+			test(t, true, false, nocsc)
+		})
+		t.Run("SET PX", func(t *testing.T) {
+			test(t, true, true, nocsc)
+		})
+	}
+}
+
 func TestLocker_WithContext_AutoExtend(t *testing.T) {
 	test := func(t *testing.T, noLoop, setpx, nocsc bool) {
 		locker := newLocker(t, noLoop, setpx, nocsc)
@@ -446,6 +495,39 @@ func TestLocker_WithContext_CancelContext(t *testing.T) {
 	}
 }
 
+func TestLocker_WithContext_ShorterTimeoutContext(t *testing.T) {
+	test := func(t *testing.T, noLoop, setpx, nocsc bool) {
+		locker := newLocker(t, noLoop, setpx, nocsc)
+		locker.validity = time.Second * 5
+		locker.interval = time.Second * 3
+		defer locker.Close()
+
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+		defer cancel()
+
+		ctx, cancel, err := locker.WithContext(ctx, strconv.Itoa(rand.Int()))
+		if err != nil {
+			t.Fatal(err)
+		}
+		time.Sleep(time.Second * 2)
+		if !errors.Is(ctx.Err(), context.DeadlineExceeded) {
+			t.Fatalf("unexpected context canceled %v", ctx.Err())
+		}
+		cancel()
+	}
+	for _, nocsc := range []bool{false, true} {
+		t.Run("Tracking Loop", func(t *testing.T) {
+			test(t, false, false, nocsc)
+		})
+		t.Run("Tracking NoLoop", func(t *testing.T) {
+			test(t, true, false, nocsc)
+		})
+		t.Run("SET PX", func(t *testing.T) {
+			test(t, true, true, nocsc)
+		})
+	}
+}
+
 func TestLocker_TryWithContext(t *testing.T) {
 	test := func(t *testing.T, noLoop, setpx, nocsc bool) {
 		locker := newLocker(t, noLoop, setpx, nocsc)