[feat][evaluation] experiment retry mode

backend/go.sum

@@ -473,6 +473,7 @@ github.com/google/pprof v0.0.0-20240827171923-fa2c70bbbfe5/go.mod h1:vavhavw2zAx
 github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI=
 github.com/google/s2a-go v0.1.8 h1:zZDs9gcbt9ZPLV0ndSyQk6Kacx2g/X+SKYovpnz3SMM=
 github.com/google/s2a-go v0.1.8/go.mod h1:6iNWHTpQ+nfNRN5E00MSdfDwVesa8hhS32PhPO8deJA=
+github.com/google/subcommands v1.2.0 h1:vWQspBTo2nEqTUFita5/KeEWlUL8kQObDFbub/EN9oE=
 github.com/google/subcommands v1.2.0/go.mod h1:ZjhPrFU+Olkh9WazFPsl27BQ4UPiG37m3yTrtFlrHVk=
 github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=

backend/infra/lock/lock.go

@@ -30,6 +30,13 @@ type ILocker interface {
 	// 后退出续期。调用方做写操作前应检查 ctx.Done 以确认仍持有锁，发生错误时应调用 cancel 以主动释放锁。
 	LockBackoffWithRenew(parent context.Context, key string, ttl time.Duration, maxHold time.Duration) (locked bool, ctx context.Context, cancel func(), err error)
 	LockWithRenew(parent context.Context, key string, ttl time.Duration, maxHold time.Duration) (locked bool, ctx context.Context, cancel func(), err error)
+
+	BackoffLockWithValue(ctx context.Context, key, val string, expiresIn time.Duration, backoff time.Duration) (bool, string, error)
+	UnlockWithValue(ctx context.Context, key, val string) (bool, error)
+	// UnlockForce deletes the key without comparing its value.
+	UnlockForce(ctx context.Context, key string) (bool, error)
+	// Exists returns true if the key exists.
+	Exists(ctx context.Context, key string) (bool, error)
 }
 
 func NewRedisLocker(c redis.Cmdable) ILocker {
@@ -143,6 +150,35 @@ func (r *redisLocker) Unlock(key string) (bool, error) {
 	return rt == 1, nil
 }
 
+func (r *redisLocker) UnlockWithValue(ctx context.Context, key, val string) (bool, error) {
+	const unlockWithValueScript = `if redis.call('GET', KEYS[1]) == ARGV[1] then redis.call('DEL', KEYS[1]); return 1; end; return 0;`
+	result, err := r.c.Eval(ctx, unlockWithValueScript, []string{key}, val).Result()
+	if err != nil {
+		return false, errors.WithMessage(err, "unlock with lua script")
+	}
+	rt, ok := result.(int64)
+	if !ok {
+		return false, errors.Errorf("unknown result type %T", result)
+	}
+	return rt == 1, nil
+}
+
+func (r *redisLocker) UnlockForce(ctx context.Context, key string) (bool, error) {
+	n, err := r.c.Del(ctx, key).Result()
+	if err != nil {
+		return false, errors.WithMessage(err, "unlock force del")
+	}
+	return n > 0, nil
+}
+
+func (r *redisLocker) Exists(ctx context.Context, key string) (bool, error) {
+	n, err := r.c.Exists(ctx, key).Result()
+	if err != nil {
+		return false, errors.WithMessage(err, "exists")
+	}
+	return n > 0, nil
+}
+
 func (r *redisLocker) renewLock(ctx context.Context, key string, ttl time.Duration, maxHold time.Duration) {
 	t1 := time.After(maxHold)
 	t2 := time.NewTicker(gvalue.Max(time.Second, ttl>>2))
@@ -203,3 +239,59 @@ func (r *redisLocker) ExpireLockIn(key string, expiresIn time.Duration) (bool, e
 	}
 	return rt == 1, nil
 }
+
+const setNXWithGetScript = `
+local ok = redis.call('SET', KEYS[1], ARGV[1], 'NX', 'PX', ARGV[2])
+if ok then
+  return {1, ARGV[1]}
+else
+  local cur = redis.call('GET', KEYS[1])
+  return {0, cur or ''}
+end
+`
+
+func (r *redisLocker) BackoffLockWithValue(ctx context.Context, key, val string, expiresIn time.Duration, maxWait time.Duration) (bool, string, error) {
+	if expiresIn < time.Second {
+		return false, "", fmt.Errorf("lock ttl too short")
+	}
+
+	var ok bool
+	var lastHolder string
+	bf := backoff.NewExponentialBackOff()
+	bf.InitialInterval = 50 * time.Millisecond
+	bf.MaxInterval = 300 * time.Millisecond
+	bf.MaxElapsedTime = maxWait
+
+	errNotLocked := errors.New("lock hold by others")
+	err := backoff.Retry(func() error {
+		result, err := r.c.Eval(ctx, setNXWithGetScript, []string{key}, val, int64(expiresIn/time.Millisecond)).Result()
+		if err != nil {
+			return errors.WithMessage(err, fmt.Sprintf("redis setnx with get fail, key: %v", key))
+		}
+		sl, okType := result.([]interface{})
+		if !okType || len(sl) != 2 {
+			return errors.Errorf("unexpected script result type %T or length", result)
+		}
+		locked, _ := sl[0].(int64)
+		if locked == 1 {
+			ok = true
+			return nil
+		}
+		switch v := sl[1].(type) {
+		case string:
+			lastHolder = v
+		case []byte:
+			lastHolder = string(v)
+		default:
+			return errors.Errorf("unexpected lua script result type %T or length, key: %v", sl[1], key)
+		}
+		return errNotLocked
+	}, bf)
+	if err != nil {
+		if errors.Is(err, errNotLocked) {
+			return false, lastHolder, nil
+		}
+		return false, "", err
+	}
+	return ok, val, nil
+}

backend/infra/lock/lock_test.go

@@ -25,6 +25,157 @@ func newIntCmdResult(val int64, err error) *redis.Cmd {
 	return cmd
 }
 
+// helper to build a redis.Cmd for Eval script result (val is typically []interface{}{locked, holder})
+func newEvalCmdResult(val interface{}, err error) *redis.Cmd {
+	cmd := redis.NewCmd(context.Background())
+	if err != nil {
+		cmd.SetErr(err)
+		return cmd
+	}
+	cmd.SetVal(val)
+	return cmd
+}
+
+func TestRedisLocker_BackoffLockWithValue(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	cases := []struct {
+		name      string
+		expiresIn time.Duration
+		maxWait   time.Duration
+		setupMock func(m *redisMocks.MockPersistentCmdable)
+		wantOk    bool
+		wantVal   string
+		wantErr   bool
+	}{
+		{
+			name:      "ttl_too_short_returns_error",
+			expiresIn: 100 * time.Millisecond,
+			maxWait:   time.Second,
+			setupMock: func(m *redisMocks.MockPersistentCmdable) {},
+			wantOk:    false,
+			wantVal:   "",
+			wantErr:   true,
+		},
+		{
+			name:      "success_locked_on_first_try",
+			expiresIn: 2 * time.Second,
+			maxWait:   time.Second,
+			setupMock: func(m *redisMocks.MockPersistentCmdable) {
+				m.EXPECT().
+					Eval(gomock.Any(), setNXWithGetScript, []string{"key1"}, "val1", int64(2000)).
+					Return(newEvalCmdResult([]interface{}{int64(1), "val1"}, nil)).
+					Times(1)
+			},
+			wantOk:  true,
+			wantVal: "val1",
+			wantErr: false,
+		},
+		{
+			name:      "locked_by_others_returns_holder_string",
+			expiresIn: 2 * time.Second,
+			maxWait:   100 * time.Millisecond,
+			setupMock: func(m *redisMocks.MockPersistentCmdable) {
+				m.EXPECT().
+					Eval(gomock.Any(), setNXWithGetScript, gomock.Any(), gomock.Any(), gomock.Any()).
+					Return(newEvalCmdResult([]interface{}{int64(0), "other_holder"}, nil)).
+					AnyTimes()
+			},
+			wantOk:  false,
+			wantVal: "other_holder",
+			wantErr: false,
+		},
+		{
+			name:      "locked_by_others_returns_holder_bytes",
+			expiresIn: 2 * time.Second,
+			maxWait:   100 * time.Millisecond,
+			setupMock: func(m *redisMocks.MockPersistentCmdable) {
+				m.EXPECT().
+					Eval(gomock.Any(), setNXWithGetScript, gomock.Any(), gomock.Any(), gomock.Any()).
+					Return(newEvalCmdResult([]interface{}{int64(0), []byte("byte_holder")}, nil)).
+					AnyTimes()
+			},
+			wantOk:  false,
+			wantVal: "byte_holder",
+			wantErr: false,
+		},
+		{
+			name:      "redis_error_returns_error",
+			expiresIn: 2 * time.Second,
+			maxWait:   100 * time.Millisecond,
+			setupMock: func(m *redisMocks.MockPersistentCmdable) {
+				m.EXPECT().
+					Eval(gomock.Any(), setNXWithGetScript, gomock.Any(), gomock.Any(), gomock.Any()).
+					Return(newEvalCmdResult(nil, context.DeadlineExceeded)).
+					AnyTimes()
+			},
+			wantOk:  false,
+			wantVal: "",
+			wantErr: true,
+		},
+		{
+			name:      "unexpected_script_result_type_returns_error",
+			expiresIn: 2 * time.Second,
+			maxWait:   100 * time.Millisecond,
+			setupMock: func(m *redisMocks.MockPersistentCmdable) {
+				m.EXPECT().
+					Eval(gomock.Any(), setNXWithGetScript, gomock.Any(), gomock.Any(), gomock.Any()).
+					Return(newEvalCmdResult(interface{}("not_a_slice"), nil)).
+					AnyTimes()
+			},
+			wantOk:  false,
+			wantVal: "",
+			wantErr: true,
+		},
+		{
+			name:      "unexpected_script_result_length_returns_error",
+			expiresIn: 2 * time.Second,
+			maxWait:   100 * time.Millisecond,
+			setupMock: func(m *redisMocks.MockPersistentCmdable) {
+				m.EXPECT().
+					Eval(gomock.Any(), setNXWithGetScript, gomock.Any(), gomock.Any(), gomock.Any()).
+					Return(newEvalCmdResult([]interface{}{int64(0)}, nil)).
+					AnyTimes()
+			},
+			wantOk:  false,
+			wantVal: "",
+			wantErr: true,
+		},
+		{
+			name:      "unexpected_holder_type_returns_error",
+			expiresIn: 2 * time.Second,
+			maxWait:   100 * time.Millisecond,
+			setupMock: func(m *redisMocks.MockPersistentCmdable) {
+				m.EXPECT().
+					Eval(gomock.Any(), setNXWithGetScript, gomock.Any(), gomock.Any(), gomock.Any()).
+					Return(newEvalCmdResult([]interface{}{int64(0), 12345}, nil)).
+					AnyTimes()
+			},
+			wantOk:  false,
+			wantVal: "",
+			wantErr: true,
+		},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			mockRedis := redisMocks.NewMockPersistentCmdable(ctrl)
+			c.setupMock(mockRedis)
+			locker := &redisLocker{c: mockRedis, holder: "holder"}
+			ok, val, err := locker.BackoffLockWithValue(context.Background(), "key1", "val1", c.expiresIn, c.maxWait)
+			if c.wantErr != (err != nil) {
+				t.Fatalf("err: got %v, wantErr %v", err, c.wantErr)
+			}
+			if ok != c.wantOk {
+				t.Errorf("ok: got %v, want %v", ok, c.wantOk)
+			}
+			if val != c.wantVal {
+				t.Errorf("val: got %q, want %q", val, c.wantVal)
+			}
+		})
+	}
+}
+
 func TestRedisLocker_renewLock_ContextDoneUnlocksAndReturns(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()