diff --git a/example/go.mod b/example/go.mod index 2020e306c..6cffcbbd6 100644 --- a/example/go.mod +++ b/example/go.mod @@ -109,7 +109,7 @@ require ( github.com/prometheus/common v0.48.0 // indirect github.com/prometheus/procfs v0.12.0 // indirect github.com/rcrowley/go-metrics v0.0.0-20250401214520-65e299d6c5c9 // indirect - github.com/redis/go-redis/v9 v9.12.1 // indirect + github.com/redis/go-redis/v9 v9.18.0 // indirect github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect github.com/ryanuber/go-glob v1.0.0 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect @@ -121,6 +121,7 @@ require ( go.opentelemetry.io/otel/sdk v1.40.0 // indirect go.opentelemetry.io/otel/trace v1.40.0 // indirect go.opentelemetry.io/proto/otlp v1.9.0 // indirect + go.uber.org/atomic v1.11.0 // indirect go.uber.org/multierr v1.11.0 // indirect go.uber.org/zap v1.27.0 // indirect golang.org/x/crypto v0.48.0 // indirect diff --git a/example/go.sum b/example/go.sum index ae145cbac..2716a76cb 100644 --- a/example/go.sum +++ b/example/go.sum @@ -214,6 +214,8 @@ github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnr github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo= github.com/klauspost/compress v1.18.1 h1:bcSGx7UbpBqMChDtsF28Lw6v/G94LPrrbMbdC3JH2co= github.com/klauspost/compress v1.18.1/go.mod h1:ZQFFVG+MdnR0P+l6wpXgIL4NTtwiKIdBnrBd8Nrxr+0= +github.com/klauspost/cpuid/v2 v2.0.9 h1:lgaqFMSdTdQYdZ04uHyN2d/eKdOMyi2YLSvlQIBFYa4= +github.com/klauspost/cpuid/v2 v2.0.9/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg= github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= @@ -274,8 +276,8 @@ github.com/prometheus/procfs v0.12.0 h1:jluTpSng7V9hY0O2R9DzzJHYb2xULk9VTR1V1R/k github.com/prometheus/procfs v0.12.0/go.mod h1:pcuDEFsWDnvcgNzo4EEweacyhjeA9Zk3cnaOZAZEfOo= github.com/rcrowley/go-metrics v0.0.0-20250401214520-65e299d6c5c9 h1:bsUq1dX0N8AOIL7EB/X911+m4EHsnWEHeJ0c+3TTBrg= github.com/rcrowley/go-metrics v0.0.0-20250401214520-65e299d6c5c9/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4= -github.com/redis/go-redis/v9 v9.12.1 h1:k5iquqv27aBtnTm2tIkROUDp8JBXhXZIVu1InSgvovg= -github.com/redis/go-redis/v9 v9.12.1/go.mod h1:huWgSWd8mW6+m0VPhJjSSQ+d6Nh1VICQ6Q5lHuCH/Iw= +github.com/redis/go-redis/v9 v9.18.0 h1:pMkxYPkEbMPwRdenAzUNyFNrDgHx9U+DrBabWNfSRQs= +github.com/redis/go-redis/v9 v9.18.0/go.mod h1:k3ufPphLU5YXwNTUcCRXGxUoF1fqxnhFQmscfkCoDA0= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs= @@ -307,6 +309,8 @@ github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6Kllzaw github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc= github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k= github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY= +github.com/zeebo/xxh3 v1.0.2 h1:xZmwmqxHZA8AI603jOQ0tMqmBr9lPeFwGg6d+xy9DC0= +github.com/zeebo/xxh3 v1.0.2/go.mod h1:5NWz9Sef7zIDm2JHfFlcQvNekmcEl9ekUZQQKCYaDcA= go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.65.0 h1:7iP2uCb7sGddAr30RRS6xjKy7AZ2JtTOPA3oolgVSw8= @@ -327,6 +331,8 @@ go.opentelemetry.io/otel/trace v1.40.0 h1:WA4etStDttCSYuhwvEa8OP8I5EWu24lkOzp+ZY go.opentelemetry.io/otel/trace v1.40.0/go.mod h1:zeAhriXecNGP/s2SEG3+Y8X9ujcJOTqQ5RgdEJcawiA= go.opentelemetry.io/proto/otlp v1.9.0 h1:l706jCMITVouPOqEnii2fIAuO3IVGBRPV5ICjceRb/A= go.opentelemetry.io/proto/otlp v1.9.0/go.mod h1:xE+Cx5E/eEHw+ISFkwPLwCZefwVjY+pqKg1qcK03+/4= +go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE= +go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0= go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= go.uber.org/mock v0.6.0 h1:hyF9dfmbgIX5EfOdasqLsWD6xqpNZlXblLB/Dbnwv3Y= diff --git a/go.mod b/go.mod index d8b5fc516..b0a93240d 100644 --- a/go.mod +++ b/go.mod @@ -12,6 +12,7 @@ require ( github.com/GoCodeAlone/go-plugin v0.0.0-20260220090904-b4c35f0e4271 github.com/GoCodeAlone/yaegi v0.17.0 github.com/IBM/sarama v1.46.3 + github.com/alicebob/miniredis/v2 v2.36.1 github.com/aws/aws-sdk-go-v2 v1.41.1 github.com/aws/aws-sdk-go-v2/config v1.32.7 github.com/aws/aws-sdk-go-v2/credentials v1.19.7 @@ -36,6 +37,7 @@ require ( github.com/launchdarkly/go-server-sdk/v7 v7.14.5 github.com/nats-io/nats.go v1.46.1 github.com/prometheus/client_golang v1.19.1 + github.com/redis/go-redis/v9 v9.18.0 github.com/stripe/stripe-go/v82 v82.5.1 go.opentelemetry.io/otel v1.40.0 go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.40.0 @@ -152,15 +154,16 @@ require ( github.com/prometheus/common v0.48.0 // indirect github.com/prometheus/procfs v0.12.0 // indirect github.com/rcrowley/go-metrics v0.0.0-20250401214520-65e299d6c5c9 // indirect - github.com/redis/go-redis/v9 v9.12.1 // indirect github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect github.com/robfig/cron/v3 v3.0.1 // indirect github.com/ryanuber/go-glob v1.0.0 // indirect + github.com/yuin/gopher-lua v1.1.1 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.65.0 // indirect go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.40.0 // indirect go.opentelemetry.io/otel/metric v1.40.0 // indirect go.opentelemetry.io/proto/otlp v1.9.0 // indirect + go.uber.org/atomic v1.11.0 // indirect go.uber.org/multierr v1.11.0 // indirect go.uber.org/zap v1.27.0 // indirect golang.org/x/exp v0.0.0-20251023183803-a4bb9ffd2546 // indirect diff --git a/go.sum b/go.sum index e4b96f60f..c904e6c0d 100644 --- a/go.sum +++ b/go.sum @@ -24,8 +24,8 @@ github.com/IBM/sarama v1.46.3 h1:njRsX6jNlnR+ClJ8XmkO+CM4unbrNr/2vB5KK6UA+IE= github.com/IBM/sarama v1.46.3/go.mod h1:GTUYiF9DMOZVe3FwyGT+dtSPceGFIgA+sPc5u6CBwko= github.com/Microsoft/go-winio v0.5.0 h1:Elr9Wn+sGKPlkaBvwu4mTrxtmOp3F3yV9qhaHbXGjwU= github.com/Microsoft/go-winio v0.5.0/go.mod h1:JPGBdM1cNvN/6ISo+n8V5iA4v8pBzdOpzfwIujj1a84= -github.com/alicebob/miniredis/v2 v2.35.0 h1:QwLphYqCEAo1eu1TqPRN2jgVMPBweeQcR21jeqDCONI= -github.com/alicebob/miniredis/v2 v2.35.0/go.mod h1:TcL7YfarKPGDAthEtl5NBeHZfeUQj6OXMm/+iu5cLMM= +github.com/alicebob/miniredis/v2 v2.36.1 h1:Dvc5oAnNOr7BIfPn7tF269U8DvRW1dBG2D5n0WrfYMI= +github.com/alicebob/miniredis/v2 v2.36.1/go.mod h1:TcL7YfarKPGDAthEtl5NBeHZfeUQj6OXMm/+iu5cLMM= github.com/aws/aws-sdk-go-v2 v1.41.1 h1:ABlyEARCDLN034NhxlRUSZr4l71mh+T5KAeGh6cerhU= github.com/aws/aws-sdk-go-v2 v1.41.1/go.mod h1:MayyLB8y+buD9hZqkCW3kX1AKq07Y5pXxtgB+rRFhz0= github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.4 h1:489krEF9xIGkOaaX3CE/Be2uWjiXrkCH6gUX+bZA/BU= @@ -253,6 +253,8 @@ github.com/karlseguin/expect v1.0.2-0.20190806010014-778a5f0c6003 h1:vJ0Snvo+SLM github.com/karlseguin/expect v1.0.2-0.20190806010014-778a5f0c6003/go.mod h1:zNBxMY8P21owkeogJELCLeHIt+voOSduHYTFUbwRAV8= github.com/klauspost/compress v1.18.1 h1:bcSGx7UbpBqMChDtsF28Lw6v/G94LPrrbMbdC3JH2co= github.com/klauspost/compress v1.18.1/go.mod h1:ZQFFVG+MdnR0P+l6wpXgIL4NTtwiKIdBnrBd8Nrxr+0= +github.com/klauspost/cpuid/v2 v2.0.9 h1:lgaqFMSdTdQYdZ04uHyN2d/eKdOMyi2YLSvlQIBFYa4= +github.com/klauspost/cpuid/v2 v2.0.9/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg= github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= @@ -343,8 +345,8 @@ github.com/prometheus/procfs v0.12.0 h1:jluTpSng7V9hY0O2R9DzzJHYb2xULk9VTR1V1R/k github.com/prometheus/procfs v0.12.0/go.mod h1:pcuDEFsWDnvcgNzo4EEweacyhjeA9Zk3cnaOZAZEfOo= github.com/rcrowley/go-metrics v0.0.0-20250401214520-65e299d6c5c9 h1:bsUq1dX0N8AOIL7EB/X911+m4EHsnWEHeJ0c+3TTBrg= github.com/rcrowley/go-metrics v0.0.0-20250401214520-65e299d6c5c9/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4= -github.com/redis/go-redis/v9 v9.12.1 h1:k5iquqv27aBtnTm2tIkROUDp8JBXhXZIVu1InSgvovg= -github.com/redis/go-redis/v9 v9.12.1/go.mod h1:huWgSWd8mW6+m0VPhJjSSQ+d6Nh1VICQ6Q5lHuCH/Iw= +github.com/redis/go-redis/v9 v9.18.0 h1:pMkxYPkEbMPwRdenAzUNyFNrDgHx9U+DrBabWNfSRQs= +github.com/redis/go-redis/v9 v9.18.0/go.mod h1:k3ufPphLU5YXwNTUcCRXGxUoF1fqxnhFQmscfkCoDA0= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs= @@ -385,6 +387,8 @@ github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1 github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY= github.com/yuin/gopher-lua v1.1.1 h1:kYKnWBjvbNP4XLT3+bPEwAXJx262OhaHDWDVOPjL46M= github.com/yuin/gopher-lua v1.1.1/go.mod h1:GBR0iDaNXjAgGg9zfCvksxSRnQx76gclCIb7kdAd1Pw= +github.com/zeebo/xxh3 v1.0.2 h1:xZmwmqxHZA8AI603jOQ0tMqmBr9lPeFwGg6d+xy9DC0= +github.com/zeebo/xxh3 v1.0.2/go.mod h1:5NWz9Sef7zIDm2JHfFlcQvNekmcEl9ekUZQQKCYaDcA= go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.65.0 h1:7iP2uCb7sGddAr30RRS6xjKy7AZ2JtTOPA3oolgVSw8= @@ -405,6 +409,8 @@ go.opentelemetry.io/otel/trace v1.40.0 h1:WA4etStDttCSYuhwvEa8OP8I5EWu24lkOzp+ZY go.opentelemetry.io/otel/trace v1.40.0/go.mod h1:zeAhriXecNGP/s2SEG3+Y8X9ujcJOTqQ5RgdEJcawiA= go.opentelemetry.io/proto/otlp v1.9.0 h1:l706jCMITVouPOqEnii2fIAuO3IVGBRPV5ICjceRb/A= go.opentelemetry.io/proto/otlp v1.9.0/go.mod h1:xE+Cx5E/eEHw+ISFkwPLwCZefwVjY+pqKg1qcK03+/4= +go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE= +go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0= go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= go.uber.org/mock v0.6.0 h1:hyF9dfmbgIX5EfOdasqLsWD6xqpNZlXblLB/Dbnwv3Y= diff --git a/scale/distributed_lock.go b/scale/distributed_lock.go index 4aec8076f..ac4688525 100644 --- a/scale/distributed_lock.go +++ b/scale/distributed_lock.go @@ -2,11 +2,15 @@ package scale import ( "context" + "crypto/rand" "database/sql" + "encoding/hex" "fmt" "hash/fnv" "sync" "time" + + "github.com/redis/go-redis/v9" ) // DistributedLock provides distributed locking for state machine transitions @@ -241,29 +245,130 @@ func hashToInt64(key string) int64 { // --- RedisLock --- +// redisReleaseScript atomically releases a Redis lock only if the caller +// holds it (i.e., the stored value matches the token). +var redisReleaseScript = redis.NewScript(` +if redis.call("get", KEYS[1]) == ARGV[1] then + return redis.call("del", KEYS[1]) +else + return 0 +end +`) + // RedisLock implements DistributedLock using Redis SET NX with TTL. -// Requires a Redis client connection. This is a stub implementation; -// the full implementation will be provided when the Redis client is -// integrated as a direct dependency. +// Uses a unique token per acquisition to ensure only the holder can release. type RedisLock struct { - // addr is the Redis server address. - addr string + addr string + password string + db int + client *redis.Client + initOnce sync.Once } -// NewRedisLock creates a new Redis distributed lock stub. -// Full implementation requires a Redis client (e.g., go-redis). +// NewRedisLock creates a new Redis distributed lock using the given address. +// The Redis client is created lazily on first use. func NewRedisLock(addr string) *RedisLock { - return &RedisLock{addr: addr} + return NewRedisLockWithOptions(addr, "", 0) +} + +// NewRedisLockWithOptions creates a new Redis distributed lock with full +// connection options. The Redis client is created lazily on first use. +func NewRedisLockWithOptions(addr, password string, db int) *RedisLock { + return &RedisLock{addr: addr, password: password, db: db} +} + +// connect initialises the Redis client exactly once. +func (l *RedisLock) connect() { + l.initOnce.Do(func() { + l.client = redis.NewClient(&redis.Options{ + Addr: l.addr, + Password: l.password, + DB: l.db, + }) + }) } -// Acquire obtains a lock using Redis SET NX with TTL. -// This is a stub that returns an error indicating Redis is not yet configured. -func (l *RedisLock) Acquire(_ context.Context, key string, _ time.Duration) (func(), error) { - return nil, fmt.Errorf("redis lock not implemented: configure Redis client for key %s at %s", key, l.addr) +// Close releases the underlying Redis client connection. +func (l *RedisLock) Close() error { + l.connect() + return l.client.Close() +} + +// randomToken generates a cryptographically random hex string used as the +// lock token, ensuring only the holder can release. +func randomToken() (string, error) { + b := make([]byte, 16) + if _, err := rand.Read(b); err != nil { + return "", fmt.Errorf("generate lock token: %w", err) + } + return hex.EncodeToString(b), nil +} + +// buildRelease returns a release function that atomically deletes the lock +// only when the stored token matches, making it safe to call multiple times. +func (l *RedisLock) buildRelease(key, token string) func() { + var once sync.Once + return func() { + once.Do(func() { + ctx := context.Background() + _ = redisReleaseScript.Run(ctx, l.client, []string{key}, token).Err() + }) + } +} + +// Acquire obtains a Redis lock for the given key using SET NX PX. +// Retries with exponential backoff until the lock is acquired or ctx is +// cancelled. Returns a release function that atomically deletes the lock. +func (l *RedisLock) Acquire(ctx context.Context, key string, ttl time.Duration) (func(), error) { + l.connect() + + token, err := randomToken() + if err != nil { + return nil, err + } + + backoff := 16 * time.Millisecond + const maxBackoff = 512 * time.Millisecond + + for { + cmd := l.client.SetArgs(ctx, key, token, redis.SetArgs{Mode: "NX", TTL: ttl}) + if err := cmd.Err(); err != nil && err != redis.Nil { + return nil, fmt.Errorf("acquire redis lock for %s: %w", key, err) + } + if cmd.Val() == "OK" { + return l.buildRelease(key, token), nil + } + + select { + case <-ctx.Done(): + return nil, fmt.Errorf("acquire redis lock for %s: %w", key, ctx.Err()) + case <-time.After(backoff): + } + + backoff *= 2 + if backoff > maxBackoff { + backoff = maxBackoff + } + } } -// TryAcquire attempts to acquire a Redis lock without blocking. -// This is a stub that returns an error indicating Redis is not yet configured. -func (l *RedisLock) TryAcquire(_ context.Context, key string, _ time.Duration) (func(), bool, error) { - return nil, false, fmt.Errorf("redis lock not implemented: configure Redis client for key %s at %s", key, l.addr) +// TryAcquire attempts to acquire a Redis lock for the given key without +// blocking. Returns (release, true, nil) if acquired, (nil, false, nil) if +// the lock is already held. +func (l *RedisLock) TryAcquire(ctx context.Context, key string, ttl time.Duration) (func(), bool, error) { + l.connect() + + token, err := randomToken() + if err != nil { + return nil, false, err + } + + cmd := l.client.SetArgs(ctx, key, token, redis.SetArgs{Mode: "NX", TTL: ttl}) + if err := cmd.Err(); err != nil && err != redis.Nil { + return nil, false, fmt.Errorf("try acquire redis lock for %s: %w", key, err) + } + if cmd.Val() != "OK" { + return nil, false, nil + } + return l.buildRelease(key, token), true, nil } diff --git a/scale/distributed_lock_test.go b/scale/distributed_lock_test.go index 488b2df48..94f3068ba 100644 --- a/scale/distributed_lock_test.go +++ b/scale/distributed_lock_test.go @@ -6,6 +6,8 @@ import ( "sync/atomic" "testing" "time" + + "github.com/alicebob/miniredis/v2" ) func TestInMemoryLockAcquireRelease(t *testing.T) { @@ -244,19 +246,197 @@ func TestPGAdvisoryLockHashConsistency(t *testing.T) { } } -func TestRedisLockStub(t *testing.T) { - lock := NewRedisLock("localhost:6379") +func TestRedisLockAcquireRelease(t *testing.T) { + mr := miniredis.RunT(t) + lock := NewRedisLock(mr.Addr()) + defer lock.Close() //nolint:errcheck + ctx := context.Background() - _, err := lock.Acquire(ctx, "test", time.Second) - if err == nil { - t.Error("expected error from Redis stub") + release, err := lock.Acquire(ctx, "test-key", time.Second) + if err != nil { + t.Fatalf("Acquire failed: %v", err) + } + if release == nil { + t.Fatal("release function should not be nil") + } + + release() + + // Lock should be available again after release. + release2, err := lock.Acquire(ctx, "test-key", time.Second) + if err != nil { + t.Fatalf("second Acquire failed: %v", err) + } + defer release2() +} + +func TestRedisLockTryAcquireFree(t *testing.T) { + mr := miniredis.RunT(t) + lock := NewRedisLock(mr.Addr()) + defer lock.Close() //nolint:errcheck + + ctx := context.Background() + + release, acquired, err := lock.TryAcquire(ctx, "try-key", time.Second) + if err != nil { + t.Fatalf("TryAcquire failed: %v", err) + } + if !acquired { + t.Fatal("expected to acquire the free lock") + } + if release == nil { + t.Fatal("release function should not be nil") + } + defer release() +} + +func TestRedisLockTryAcquireHeld(t *testing.T) { + mr := miniredis.RunT(t) + lock := NewRedisLock(mr.Addr()) + defer lock.Close() //nolint:errcheck + + ctx := context.Background() + + release, acquired, err := lock.TryAcquire(ctx, "held-key", time.Second) + if err != nil || !acquired { + t.Fatalf("first TryAcquire failed: acquired=%v err=%v", acquired, err) + } + defer release() + + _, acquired2, err := lock.TryAcquire(ctx, "held-key", time.Second) + if err != nil { + t.Fatalf("second TryAcquire returned unexpected error: %v", err) + } + if acquired2 { + t.Error("expected TryAcquire to fail when lock is held") + } +} + +func TestRedisLockReleaseIdempotent(t *testing.T) { + mr := miniredis.RunT(t) + lock := NewRedisLock(mr.Addr()) + defer lock.Close() //nolint:errcheck + + ctx := context.Background() + + release, err := lock.Acquire(ctx, "idempotent-key", time.Second) + if err != nil { + t.Fatalf("Acquire failed: %v", err) + } + + // Calling release twice must not panic or error. + release() + release() +} + +func TestRedisLockDifferentHolderCannotRelease(t *testing.T) { + mr := miniredis.RunT(t) + lock1 := NewRedisLock(mr.Addr()) + lock2 := NewRedisLock(mr.Addr()) + defer lock1.Close() //nolint:errcheck + defer lock2.Close() //nolint:errcheck + + ctx := context.Background() + + // lock1 acquires the lock. + release1, err := lock1.Acquire(ctx, "safety-key", time.Second) + if err != nil { + t.Fatalf("Acquire failed: %v", err) + } + defer release1() + + // lock2 tries to acquire — should not succeed immediately. + _, acquired, err := lock2.TryAcquire(ctx, "safety-key", time.Second) + if err != nil { + t.Fatalf("TryAcquire on lock2 returned error: %v", err) + } + if acquired { + t.Fatal("lock2 should not have acquired a lock held by lock1") + } + + // Simulate lock2 calling the release function it should not have — + // build a release with a wrong token and ensure lock1's key is intact. + fakeRelease := lock2.buildRelease("safety-key", "wrong-token") + fakeRelease() // must not delete lock1's key + + // lock1's lock should still be held. + _, stillHeld, err := lock2.TryAcquire(ctx, "safety-key", time.Second) + if err != nil { + t.Fatalf("TryAcquire after fake release returned error: %v", err) + } + if stillHeld { + t.Error("lock1's lock should still be held after lock2 attempted a fake release") + } +} + +func TestRedisLockTTLExpiry(t *testing.T) { + mr := miniredis.RunT(t) + lock := NewRedisLock(mr.Addr()) + defer lock.Close() //nolint:errcheck + + ctx := context.Background() + + _, err := lock.Acquire(ctx, "ttl-key", 500*time.Millisecond) + if err != nil { + t.Fatalf("Acquire failed: %v", err) + } + + // Advance miniredis clock past the TTL. + mr.FastForward(time.Second) + + // Lock should be available after TTL expiry. + release, acquired, err := lock.TryAcquire(ctx, "ttl-key", time.Second) + if err != nil { + t.Fatalf("TryAcquire after TTL failed: %v", err) + } + if !acquired { + t.Error("expected lock to be available after TTL expired") + } + if release != nil { + release() + } +} + +func TestRedisLockAcquireContextCancel(t *testing.T) { + mr := miniredis.RunT(t) + lock := NewRedisLock(mr.Addr()) + defer lock.Close() //nolint:errcheck + + ctx := context.Background() + + // Hold the lock with the same client (use a different key path via options). + release, err := lock.Acquire(ctx, "cancel-redis-key", time.Minute) + if err != nil { + t.Fatalf("Acquire failed: %v", err) } + defer release() + + // Second acquire should block until context is cancelled. + cancelCtx, cancel := context.WithTimeout(ctx, 100*time.Millisecond) + defer cancel() - _, _, err = lock.TryAcquire(ctx, "test", time.Second) + _, err = lock.Acquire(cancelCtx, "cancel-redis-key", time.Minute) if err == nil { - t.Error("expected error from Redis stub TryAcquire") + t.Fatal("expected error when context is cancelled") + } +} + +func TestRedisLockNewRedisLockWithOptions(t *testing.T) { + mr := miniredis.RunT(t) + lock := NewRedisLockWithOptions(mr.Addr(), "", 0) + defer lock.Close() //nolint:errcheck + + ctx := context.Background() + + release, acquired, err := lock.TryAcquire(ctx, "opts-key", time.Second) + if err != nil { + t.Fatalf("TryAcquire failed: %v", err) + } + if !acquired { + t.Fatal("expected to acquire lock with options constructor") } + defer release() } func TestDistributedLockInterface(t *testing.T) {