diff --git a/README.md b/README.md index f37e3b8..c2cc2c7 100644 --- a/README.md +++ b/README.md @@ -75,6 +75,56 @@ func main() { } ``` +### Fencing tokens + +Set `Options.FenceKey` to mint a **fencing token** with the lock: a strictly +increasing value, incremented atomically on each new acquisition and returned by +`Lock.FenceToken`. Stamp every write to the protected resource with the token and +have the resource reject any write carrying an older token. A lock holder that +pauses (GC, scheduling) long enough to lose the lock without noticing is then +fenced out — its writes carry a stale token and are refused. This is the +mitigation described in Martin Kleppmann's +[How to do distributed locking](https://martin.kleppmann.com/2016/02/08/how-to-do-distributed-locking.html). + +```go +func fence() { + client := redis.NewClient(&redis.Options{Network: "tcp", Addr: "127.0.0.1:6379"}) + defer client.Close() + + locker := redislock.New(client) + + ctx := context.Background() + + // Obtain a lock with a fencing token. + lock, err := locker.Obtain(ctx, "my-key", time.Second, &redislock.Options{FenceKey: "my-key:fence"}) + if err != nil { + log.Fatalln(err) + } + defer lock.Release(ctx) + + // FenceToken is 0 without a FenceKey. Stamp writes with the token; reject older ones. + if token := lock.FenceToken(); token != 0 { + fmt.Printf("fenced write with token %d\n", token) + } +} +``` + +A few notes: + +- Enforcement is the resource's job: `redislock` mints and returns the token, but + the check (`reject if incoming < highest applied`) must happen atomically at the + resource, which is the only place the comparison and the write can be made one + operation. +- You supply `FenceKey`, so you control its placement. On Redis Cluster it must + hash to the same slot as the lock key(s) — share a `{hashtag}`, e.g. lock + `{job}:lock` with fence `{job}:fence` — or the script fails with `CROSSSLOT`. +- The counter persists across release so it keeps increasing; it is never reset + by `Release`. +- The token is only as monotonic as the underlying Redis. On a single instance it + is strict; on a Sentinel/Cluster failover that loses the `INCR`, it can regress. + For strict cross-failover monotonicity, source the token from a linearizable + store instead. + ### External watchdog `redislock` deliberately does not bundle a built-in watchdog goroutine: refresh diff --git a/README.md.tpl b/README.md.tpl index cd541d0..21394cd 100644 --- a/README.md.tpl +++ b/README.md.tpl @@ -27,6 +27,37 @@ import ( func main() {{ "Example" | code }} ``` +### Fencing tokens + +Set `Options.FenceKey` to mint a **fencing token** with the lock: a strictly +increasing value, incremented atomically on each new acquisition and returned by +`Lock.FenceToken`. Stamp every write to the protected resource with the token and +have the resource reject any write carrying an older token. A lock holder that +pauses (GC, scheduling) long enough to lose the lock without noticing is then +fenced out — its writes carry a stale token and are refused. This is the +mitigation described in Martin Kleppmann's +[How to do distributed locking](https://martin.kleppmann.com/2016/02/08/how-to-do-distributed-locking.html). + +```go +func fence() {{ "ExampleClient_Obtain_fence" | code }} +``` + +A few notes: + +- Enforcement is the resource's job: `redislock` mints and returns the token, but + the check (`reject if incoming < highest applied`) must happen atomically at the + resource, which is the only place the comparison and the write can be made one + operation. +- You supply `FenceKey`, so you control its placement. On Redis Cluster it must + hash to the same slot as the lock key(s) — share a `{hashtag}`, e.g. lock + `{job}:lock` with fence `{job}:fence` — or the script fails with `CROSSSLOT`. +- The counter persists across release so it keeps increasing; it is never reset + by `Release`. +- The token is only as monotonic as the underlying Redis. On a single instance it + is strict; on a Sentinel/Cluster failover that loses the `INCR`, it can regress. + For strict cross-failover monotonicity, source the token from a linearizable + store instead. + ### External watchdog `redislock` deliberately does not bundle a built-in watchdog goroutine: refresh diff --git a/example_test.go b/example_test.go index d8e9023..ef5ab0c 100644 --- a/example_test.go +++ b/example_test.go @@ -89,6 +89,27 @@ func ExampleClient_Obtain_retry() { fmt.Println("I have a lock!") } +func ExampleClient_Obtain_fence() { + client := redis.NewClient(&redis.Options{Network: "tcp", Addr: "127.0.0.1:6379"}) + defer client.Close() + + locker := redislock.New(client) + + ctx := context.Background() + + // Obtain a lock with a fencing token. + lock, err := locker.Obtain(ctx, "my-key", time.Second, &redislock.Options{FenceKey: "my-key:fence"}) + if err != nil { + log.Fatalln(err) + } + defer lock.Release(ctx) + + // FenceToken is 0 without a FenceKey. Stamp writes with the token; reject older ones. + if token := lock.FenceToken(); token != 0 { + fmt.Printf("fenced write with token %d\n", token) + } +} + func ExampleLock_Refresh_watchdog() { client := redis.NewClient(&redis.Options{Network: "tcp", Addr: "127.0.0.1:6379"}) defer client.Close() diff --git a/obtain.lua b/obtain.lua index 11e55d6..4bafa46 100644 --- a/obtain.lua +++ b/obtain.lua @@ -1,31 +1,54 @@ --- obtain.lua: arguments => [value, tokenLen, ttl] +-- obtain.lua: arguments => [value, tokenLen, ttl, fenced] -- Obtain.lua try to set provided keys's with value and ttl if they do not exists. --- Keys can be overriden if they already exists and the correct value+tokenLen is provided. +-- Keys can be overriden if they already exists and the correct value+tokenLen is provided. +-- When fenced is "1", the last KEYS entry is a fence key and obtain returns a +-- fencing token instead of "OK". + +local fenced = tonumber(ARGV[4]) == 1 + +-- Lock keys are KEYS[1..lockCount]; the fence key, if any, is the last entry. +local lockCount = #KEYS +if fenced then + lockCount = lockCount - 1 +end local function pexpire(ttl) -- Update keys ttls. - for _, key in ipairs(KEYS) do - redis.call("pexpire", key, ttl) + for i = 1, lockCount do + redis.call("pexpire", KEYS[i], ttl) end end -- canOverrideLock check either or not the provided token match -- previously set lock's tokens. -local function canOverrideKeys() +local function canOverrideKeys() local offset = tonumber(ARGV[2]) - for _, key in ipairs(KEYS) do - if redis.call("getrange", key, 0, offset-1) ~= string.sub(ARGV[1], 1, offset) then + for i = 1, lockCount do + if redis.call("getrange", KEYS[i], 0, offset-1) ~= string.sub(ARGV[1], 1, offset) then return false end end return true end +-- reply returns the fencing token, advancing it only on a fresh acquisition, +-- or "OK" when fencing is disabled. +local function reply(fresh) + if not fenced then + return redis.status_reply("OK") + end + local fenceKey = KEYS[#KEYS] + if fresh then + return redis.call("incr", fenceKey) + end + return tonumber(redis.call("get", fenceKey) or "0") +end + -- Prepare mset arguments. local setArgs = {} -for _, key in ipairs(KEYS) do - table.insert(setArgs, key) +for i = 1, lockCount do + table.insert(setArgs, KEYS[i]) table.insert(setArgs, ARGV[1]) end @@ -34,7 +57,9 @@ if redis.call("msetnx", unpack(setArgs)) ~= 1 then return false end redis.call("mset", unpack(setArgs)) + pexpire(ARGV[3]) + return reply(false) end pexpire(ARGV[3]) -return redis.status_reply("OK") \ No newline at end of file +return reply(true) diff --git a/redislock.go b/redislock.go index 5219420..aad76c1 100644 --- a/redislock.go +++ b/redislock.go @@ -81,8 +81,12 @@ func (c *Client) ObtainMulti(ctx context.Context, keys []string, ttl time.Durati value := token + opt.getMetadata() ttlVal := strconv.FormatInt(int64(ttl/time.Millisecond), 10) + fenceKey := opt.getFenceKey() + + var fenceToken int64 + if err := withRetry(ctx, ttl, opt.getRetryStrategy(), func(ctx context.Context) (bool, error) { - ok, err := c.obtain(ctx, keys, value, len(token), ttlVal) + ok, ft, err := c.obtain(ctx, keys, value, len(token), ttlVal, fenceKey) if err != nil { // any non-nil error from obtain is terminal (transient redis // errors are unlikely to clear within a lock TTL and retrying a @@ -90,6 +94,7 @@ func (c *Client) ObtainMulti(ctx context.Context, keys []string, ttl time.Durati return true, err } if ok { + fenceToken = ft return true, nil } // lock is held by someone else; retryable. @@ -97,7 +102,7 @@ func (c *Client) ObtainMulti(ctx context.Context, keys []string, ttl time.Durati }); err != nil { return nil, err } - return &Lock{Client: c, keys: keys, value: value, tokenLen: len(token)}, nil + return &Lock{Client: c, keys: keys, value: value, tokenLen: len(token), fenceToken: fenceToken}, nil } // withRetry runs attempt repeatedly until it signals it is done, the retry @@ -160,15 +165,33 @@ func withRetry(ctx context.Context, ttl time.Duration, retry RetryStrategy, atte } } -func (c *Client) obtain(ctx context.Context, keys []string, value string, tokenLen int, ttlVal string) (bool, error) { - _, err := luaObtain.Run(ctx, c.client, keys, value, tokenLen, ttlVal).Result() +func (c *Client) obtain(ctx context.Context, keys []string, value string, tokenLen int, ttlVal, fenceKey string) (ok bool, fenceToken int64, err error) { + // The fence key, if any, goes in KEYS (so Cluster checks its slot) as the + // last entry; ARGV[4] flags its presence. + runKeys := keys + fence := 0 + + if fenceKey != "" { + runKeys = make([]string, 0, len(keys)+1) + runKeys = append(runKeys, keys...) + runKeys = append(runKeys, fenceKey) + fence = 1 + } + + res, err := luaObtain.Run(ctx, c.client, runKeys, value, tokenLen, ttlVal, fence).Result() if err != nil { if errors.Is(err, redis.Nil) { - return false, nil + return false, 0, nil } - return false, err + return false, 0, err + } + + // With fencing the reply is the integer token; otherwise it is the "OK" status. + if token, isInt := res.(int64); isInt { + return true, token, nil } - return true, nil + + return true, 0, nil } func (c *Client) randomToken() (string, error) { @@ -190,9 +213,10 @@ func (c *Client) randomToken() (string, error) { // Lock represents an obtained, distributed lock. type Lock struct { *Client - keys []string - value string - tokenLen int + keys []string + value string + tokenLen int + fenceToken int64 } // Obtain is a short-cut for New(...).Obtain(...). @@ -221,6 +245,12 @@ func (l *Lock) Token() string { return l.value[:l.tokenLen] } +// FenceToken returns the lock's fencing token, or 0 if it was obtained without +// Options.FenceKey. Tokens start at 1, so 0 always means unfenced. +func (l *Lock) FenceToken() int64 { + return l.fenceToken +} + // Metadata returns the metadata of the lock. func (l *Lock) Metadata() string { return l.value[l.tokenLen:] @@ -297,6 +327,12 @@ type Options struct { // Token is a unique value that is used to identify the lock. By default, a random tokens are generated. Use this // option to provide a custom token instead. Token string + + // FenceKey enables a fencing token, minted at this key on each new + // acquisition and returned by Lock.FenceToken. On Redis Cluster it must hash + // to the same slot as the lock key(s). + // Default: empty, no fencing. + FenceKey string } func (o *Options) getMetadata() string { @@ -313,6 +349,13 @@ func (o *Options) getToken() string { return "" } +func (o *Options) getFenceKey() string { + if o != nil { + return o.FenceKey + } + return "" +} + func (o *Options) getRetryStrategy() RetryStrategy { if o != nil && o.RetryStrategy != nil { return o.RetryStrategy diff --git a/redislock_test.go b/redislock_test.go index 5f0b687..d6bbd67 100644 --- a/redislock_test.go +++ b/redislock_test.go @@ -113,6 +113,75 @@ func TestObtain_custom_token(t *testing.T) { } } +func TestObtain_fence(t *testing.T) { + rc := redisConnect(t) + lockKey := rc.lockKey() + fenceKey := lockKey + ":fence" + rc.keys = append(rc.keys, fenceKey) + + // no token without a FenceKey + plain := quickObtain(t, time.Hour) + defer plain.Release(t.Context()) + if got := plain.FenceToken(); got != 0 { + t.Fatalf("expected no fence token without Options.FenceKey, got %v", got) + } + if err := plain.Release(t.Context()); err != nil { + t.Fatal(err) + } + + // first fenced obtain + lock1, err := Obtain(t.Context(), rc, lockKey, time.Hour, &Options{FenceKey: fenceKey}) + if err != nil { + t.Fatal(err) + } + tok1 := lock1.FenceToken() + if exp, got := int64(1), tok1; exp != got { + t.Fatalf("expected first fence token %v, got %v", exp, got) + } + + if err := lock1.Release(t.Context()); err != nil { + t.Fatal(err) + } + + // next obtain must mint a greater token + lock2, err := Obtain(t.Context(), rc, lockKey, time.Hour, &Options{FenceKey: fenceKey}) + if err != nil { + t.Fatal(err) + } + defer lock2.Release(t.Context()) + + if tok2 := lock2.FenceToken(); tok2 <= tok1 { + t.Fatalf("expected fence token > %v, got %v", tok1, tok2) + } +} + +func TestObtain_fence_reentrant(t *testing.T) { + rc := redisConnect(t) + lockKey := rc.lockKey() + fenceKey := lockKey + ":fence" + rc.keys = append(rc.keys, fenceKey) + + // obtain with a known token + lock1, err := Obtain(t.Context(), rc, lockKey, time.Hour, &Options{Token: "foo", FenceKey: fenceKey}) + if err != nil { + t.Fatal(err) + } + defer lock1.Release(t.Context()) + + tok1 := lock1.FenceToken() + + // re-obtain (override) must not advance the token + lock2, err := Obtain(t.Context(), rc, lockKey, time.Hour, &Options{Token: "foo", FenceKey: fenceKey}) + if err != nil { + t.Fatal(err) + } + defer lock2.Release(t.Context()) + + if exp, got := tok1, lock2.FenceToken(); exp != got { + t.Fatalf("re-entrant override must not mint a new token: expected %v, got %v", exp, got) + } +} + func TestObtain_retry_success(t *testing.T) { rc := redisConnect(t) lockKey := rc.lockKey()