Skip to content

Commit 7d20990

Browse files
ItalyPaleAleItalyPaleAle
committed
Distributed lock: add Lock (blocking) and RenewLock methods
Includes conformance tests Signed-off-by: ItalyPaleAle <[email protected]>
1 parent 9f2cc5c commit 7d20990

File tree

12 files changed

+397
-186
lines changed

12 files changed

+397
-186
lines changed

‎crypto/feature.go

Lines changed: 2 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -14,13 +14,8 @@ limitations under the License.
1414
package crypto
1515

1616
import (
17-
"golang.org/x/exp/slices"
17+
"github.com/dapr/components-contrib/internal/features"
1818
)
1919

2020
// Feature names a feature that can be implemented by the crypto provider components.
21-
type Feature string
22-
23-
// IsPresent checks if a given feature is present in the list.
24-
func (f Feature) IsPresent(features []Feature) bool {
25-
return slices.Contains(features, f)
26-
}
21+
type Feature = features.Feature[SubtleCrypto]

‎lock/metadata.go renamed to ‎internal/features/feature.go

Lines changed: 11 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
Copyright 2021 The Dapr Authors
2+
Copyright 2023 The Dapr Authors
33
Licensed under the Apache License, Version 2.0 (the "License");
44
you may not use this file except in compliance with the License.
55
You may obtain a copy of the License at
@@ -11,11 +11,16 @@ See the License for the specific language governing permissions and
1111
limitations under the License.
1212
*/
1313

14-
package lock
14+
package features
1515

16-
import "github.com/dapr/components-contrib/metadata"
16+
import (
17+
"golang.org/x/exp/slices"
18+
)
1719

18-
// Metadata contains a lock store specific set of metadata property.
19-
type Metadata struct {
20-
metadata.Base `json:",inline"`
20+
// Feature is a generic type for features supported by components.
21+
type Feature[T any] string
22+
23+
// IsPresent checks if a given feature is present in the list.
24+
func (f Feature[T]) IsPresent(features []Feature[T]) bool {
25+
return slices.Contains(features, f)
2126
}

‎lock/lock.go

Lines changed: 100 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,100 @@
1+
/*
2+
Copyright 2021 The Dapr Authors
3+
Licensed under the Apache License, Version 2.0 (the "License");
4+
you may not use this file except in compliance with the License.
5+
You may obtain a copy of the License at
6+
http://www.apache.org/licenses/LICENSE-2.0
7+
Unless required by applicable law or agreed to in writing, software
8+
distributed under the License is distributed on an "AS IS" BASIS,
9+
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
10+
See the License for the specific language governing permissions and
11+
limitations under the License.
12+
*/
13+
14+
package lock
15+
16+
import (
17+
"context"
18+
19+
"github.com/dapr/components-contrib/internal/features"
20+
"github.com/dapr/components-contrib/metadata"
21+
)
22+
23+
// Store is the interface for lock stores.
24+
type Store interface {
25+
metadata.ComponentWithMetadata
26+
27+
// Init the lock store.
28+
Init(ctx context.Context, metadata Metadata) error
29+
30+
// Features returns the list of supported features.
31+
Features() []Feature
32+
33+
// Lock acquires a lock.
34+
// If the lock is owned by someone else, this method blocks until the lock can be acquired or the context is canceled.
35+
Lock(ctx context.Context, req *LockRequest) (*LockResponse, error)
36+
37+
// TryLock tries to acquire a lock.
38+
// If the lock cannot be acquired, it returns immediately.
39+
TryLock(ctx context.Context, req *LockRequest) (*LockResponse, error)
40+
41+
// RenewLock attempts to renew a lock if the lock is still valid.
42+
RenewLock(ctx context.Context, req *RenewLockRequest) (*RenewLockResponse, error)
43+
44+
// Unlock tries to release a lock if the lock is still valid.
45+
Unlock(ctx context.Context, req *UnlockRequest) (*UnlockResponse, error)
46+
}
47+
48+
// Metadata contains a lock store specific set of metadata property.
49+
type Metadata struct {
50+
metadata.Base `json:",inline"`
51+
}
52+
53+
// Feature names a feature that can be implemented by the lock stores.
54+
type Feature = features.Feature[Store]
55+
56+
// LockRequest is the request to acquire locks, used by Lock and TryLock.
57+
type LockRequest struct {
58+
ResourceID string `json:"resourceId"`
59+
LockOwner string `json:"lockOwner"`
60+
ExpiryInSeconds int32 `json:"expiryInSeconds"`
61+
}
62+
63+
// LockResponse is the response used by Lock and TryLock when the operation is completed.
64+
type LockResponse struct {
65+
Success bool `json:"success"`
66+
}
67+
68+
// RenewLockRequest is a lock renewal request.
69+
type RenewLockRequest struct {
70+
ResourceID string `json:"resourceId"`
71+
LockOwner string `json:"lockOwner"`
72+
ExpiryInSeconds int32 `json:"expiryInSeconds"`
73+
}
74+
75+
// RenewLockResponse is a lock renewal request.
76+
type RenewLockResponse struct {
77+
Status LockStatus `json:"status"`
78+
}
79+
80+
// UnlockRequest is a lock release request.
81+
type UnlockRequest struct {
82+
ResourceID string `json:"resourceId"`
83+
LockOwner string `json:"lockOwner"`
84+
}
85+
86+
// Status when releasing the lock.
87+
type UnlockResponse struct {
88+
Status LockStatus `json:"status"`
89+
}
90+
91+
// LockStatus is the status included in lock responses.
92+
type LockStatus int32
93+
94+
// lock status.
95+
const (
96+
LockStatusInternalError LockStatus = -1
97+
LockStatusSuccess LockStatus = 0
98+
LockStatusNotExist LockStatus = 1
99+
LockStatusOwnerMismatch LockStatus = 2
100+
)

‎lock/redis/standalone.go

Lines changed: 138 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -18,37 +18,47 @@ import (
1818
"errors"
1919
"fmt"
2020
"reflect"
21+
"sync/atomic"
2122
"time"
2223

24+
"github.com/cenkalti/backoff/v4"
2325
rediscomponent "github.com/dapr/components-contrib/internal/component/redis"
2426
"github.com/dapr/components-contrib/lock"
2527
contribMetadata "github.com/dapr/components-contrib/metadata"
2628
"github.com/dapr/kit/logger"
2729
)
2830

29-
const unlockScript = `local v = redis.call("get",KEYS[1]); if v==false then return -1 end; if v~=ARGV[1] then return -2 else return redis.call("del",KEYS[1]) end`
31+
const (
32+
unlockScript = `local v = redis.call("get",KEYS[1]); if v==false then return -1 end; if v~=ARGV[1] then return -2 else return redis.call("del",KEYS[1]) end`
33+
renewLockScript = `local v = redis.call("get",KEYS[1]); if v==false then return -1 end; if v~=ARGV[1] then return -2 else return redis.call("expire",KEYS[1],ARGV[2]) end`
34+
)
35+
36+
var ErrComponentClosed = errors.New("component is closed")
3037

3138
// Standalone Redis lock store.
3239
// Any fail-over related features are not supported, such as Sentinel and Redis Cluster.
3340
type StandaloneRedisLock struct {
3441
client rediscomponent.RedisClient
3542
clientSettings *rediscomponent.Settings
3643

37-
logger logger.Logger
44+
closed atomic.Bool
45+
runnincCh chan struct{}
46+
logger logger.Logger
3847
}
3948

4049
// NewStandaloneRedisLock returns a new standalone redis lock.
41-
// Do not use this lock with a redis cluster, which might lead to unexpected lock loss.
50+
// Do not use this lock with a Redis cluster, which might lead to unexpected lock loss.
4251
func NewStandaloneRedisLock(logger logger.Logger) lock.Store {
4352
s := &StandaloneRedisLock{
44-
logger: logger,
53+
logger: logger,
54+
runnincCh: make(chan struct{}),
4555
}
4656

4757
return s
4858
}
4959

50-
// Init StandaloneRedisLock.
51-
func (r *StandaloneRedisLock) InitLockStore(ctx context.Context, metadata lock.Metadata) (err error) {
60+
// Init the lock store.
61+
func (r *StandaloneRedisLock) Init(ctx context.Context, metadata lock.Metadata) (err error) {
5262
// Create the client
5363
r.client, r.clientSettings, err = rediscomponent.ParseClientFromProperties(metadata.Properties, contribMetadata.LockStoreType)
5464
if err != nil {
@@ -79,51 +89,150 @@ func (r *StandaloneRedisLock) InitLockStore(ctx context.Context, metadata lock.M
7989
return nil
8090
}
8191

92+
// Features returns the list of supported features.
93+
func (r *StandaloneRedisLock) Features() []lock.Feature {
94+
return nil
95+
}
96+
97+
// Lock tries to acquire a lock.
98+
// If the lock is owned by someone else, this method blocks until the lock can be acquired or the context is canceled.
99+
func (r *StandaloneRedisLock) Lock(ctx context.Context, req *lock.LockRequest) (res *lock.LockResponse, err error) {
100+
if r.closed.Load() {
101+
return nil, ErrComponentClosed
102+
}
103+
104+
// We try to acquire a lock through periodic polling
105+
// A potentially more efficient way would be to use keyspace notifications to subscribe to changes in the key we subscribe to
106+
// However, keyspace notifications:
107+
// 1. Are not enabled by default in Redis, and require an explicit configuration change, which adds quite a bit of complexity for the user: https://redis.io/docs/manual/keyspace-notifications/
108+
// 2. When a connection to Redis calls SUBSCRIBE to watch for notifications, it cannot be used for anything else (unless we switch the protocol to RESP3, which must be explicitly chosen and only works with Redis 6+: https://redis.io/commands/hello/)
109+
// So, periodic polling it is
110+
111+
// We use an exponential backoff here because it supports a randomization factor
112+
bo := backoff.NewExponentialBackOff()
113+
bo.MaxElapsedTime = 0
114+
bo.InitialInterval = 50 * time.Millisecond
115+
bo.MaxInterval = 500 * time.Millisecond
116+
bo.RandomizationFactor = 0.5
117+
118+
// Repat until we get the lock, or context is canceled
119+
for {
120+
// Try to acquire the lock
121+
res, err = r.TryLock(ctx, req)
122+
if err != nil {
123+
// If we got an error, return right away
124+
return nil, err
125+
}
126+
127+
// Let's see if we got the lock
128+
if res.Success {
129+
return res, nil
130+
}
131+
132+
// Sleep till the next tick and try again
133+
// Stop when context is done or component is closed
134+
t := time.NewTimer(bo.NextBackOff())
135+
select {
136+
case <-t.C:
137+
// Nop, retry
138+
case <-ctx.Done():
139+
return nil, ctx.Err()
140+
case <-r.runnincCh:
141+
return nil, ErrComponentClosed
142+
}
143+
}
144+
}
145+
82146
// TryLock tries to acquire a lock.
83147
// If the lock cannot be acquired, it returns immediately.
84-
func (r *StandaloneRedisLock) TryLock(ctx context.Context, req *lock.TryLockRequest) (*lock.TryLockResponse, error) {
85-
// Set a key if doesn't exist with an expiration time
148+
func (r *StandaloneRedisLock) TryLock(ctx context.Context, req *lock.LockRequest) (*lock.LockResponse, error) {
149+
if r.closed.Load() {
150+
return nil, ErrComponentClosed
151+
}
152+
153+
// Set a key if doesn't exist, with an expiration time
86154
nxval, err := r.client.SetNX(ctx, req.ResourceID, req.LockOwner, time.Second*time.Duration(req.ExpiryInSeconds))
87155
if nxval == nil {
88-
return &lock.TryLockResponse{}, fmt.Errorf("setNX returned a nil response")
156+
return &lock.LockResponse{}, fmt.Errorf("setNX returned a nil response")
89157
}
90-
91158
if err != nil {
92-
return &lock.TryLockResponse{}, err
159+
return &lock.LockResponse{}, err
93160
}
94161

95-
return &lock.TryLockResponse{
162+
return &lock.LockResponse{
96163
Success: *nxval,
97164
}, nil
98165
}
99166

167+
// RenewLock attempts to renew a lock if the lock is still valid.
168+
func (r *StandaloneRedisLock) RenewLock(ctx context.Context, req *lock.RenewLockRequest) (*lock.RenewLockResponse, error) {
169+
if r.closed.Load() {
170+
return nil, ErrComponentClosed
171+
}
172+
173+
// Delegate to client.eval lua script
174+
evalInt, parseErr, err := r.client.EvalInt(ctx, renewLockScript, []string{req.ResourceID}, req.LockOwner, req.ExpiryInSeconds)
175+
if evalInt == nil {
176+
res := &lock.RenewLockResponse{
177+
Status: lock.LockStatusInternalError,
178+
}
179+
return res, errors.New("eval renew lock script returned a nil response")
180+
}
181+
182+
// Parse result
183+
if parseErr != nil {
184+
return &lock.RenewLockResponse{
185+
Status: lock.LockStatusInternalError,
186+
}, err
187+
}
188+
var status lock.LockStatus
189+
switch {
190+
case *evalInt >= 0:
191+
status = lock.LockStatusSuccess
192+
case *evalInt == -1:
193+
status = lock.LockStatusNotExist
194+
case *evalInt == -2:
195+
status = lock.LockStatusOwnerMismatch
196+
default:
197+
status = lock.LockStatusInternalError
198+
}
199+
200+
return &lock.RenewLockResponse{
201+
Status: status,
202+
}, nil
203+
}
204+
100205
// Unlock tries to release a lock if the lock is still valid.
101206
func (r *StandaloneRedisLock) Unlock(ctx context.Context, req *lock.UnlockRequest) (*lock.UnlockResponse, error) {
207+
if r.closed.Load() {
208+
return nil, ErrComponentClosed
209+
}
210+
102211
// Delegate to client.eval lua script
103212
evalInt, parseErr, err := r.client.EvalInt(ctx, unlockScript, []string{req.ResourceID}, req.LockOwner)
104213
if evalInt == nil {
105214
res := &lock.UnlockResponse{
106-
Status: lock.InternalError,
215+
Status: lock.LockStatusInternalError,
107216
}
108217
return res, errors.New("eval unlock script returned a nil response")
109218
}
110219

111220
// Parse result
112221
if parseErr != nil {
113222
return &lock.UnlockResponse{
114-
Status: lock.InternalError,
223+
Status: lock.LockStatusInternalError,
115224
}, err
116225
}
117-
var status lock.Status
226+
var status lock.LockStatus
118227
switch {
119228
case *evalInt >= 0:
120-
status = lock.Success
229+
status = lock.LockStatusSuccess
121230
case *evalInt == -1:
122-
status = lock.LockDoesNotExist
231+
status = lock.LockStatusNotExist
123232
case *evalInt == -2:
124-
status = lock.LockBelongsToOthers
233+
status = lock.LockStatusOwnerMismatch
125234
default:
126-
status = lock.InternalError
235+
status = lock.LockStatusInternalError
127236
}
128237

129238
return &lock.UnlockResponse{
@@ -133,12 +242,17 @@ func (r *StandaloneRedisLock) Unlock(ctx context.Context, req *lock.UnlockReques
133242

134243
// Close shuts down the client's redis connections.
135244
func (r *StandaloneRedisLock) Close() error {
136-
if r.client != nil {
137-
err := r.client.Close()
138-
r.client = nil
139-
return err
245+
if !r.closed.CompareAndSwap(false, true) {
246+
return nil
140247
}
141-
return nil
248+
249+
close(r.runnincCh)
250+
251+
if r.client == nil {
252+
return nil
253+
}
254+
255+
return r.client.Close()
142256
}
143257

144258
// GetComponentMetadata returns the metadata of the component.

0 commit comments

Comments
 (0)