Counter/Lock: Redis INCR parity and safe lock release
- Counter#increment now seeds a missing key at the delta (Redis INCR/INCRBY semantics). Couchbase's native increment seeds at 'initial' and ignores the delta on creation, so the first INCR returned 0 — a drop-in replacement that silently undercounts by one is exactly the kind of bug nobody ever finds. Explicit initial: still overrides. Decrement keeps initial: 0 and its unsigned clamp-at-zero divergence from Redis is now documented and tested. - Lock#release requires the CAS handle: passing nil used to delete the lock document unconditionally, so any caller could unlock a lock it never held. force_release! is the explicit administrative override. - Lock#acquire validates a positive ttl — a nil/zero ttl would create a lock that survives its holder's crash forever. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
1 parent
84c76bb093
commit
bde6df57f4
2 files changed
+55
-10
No files matched your search
@@ -7,12 +7,15 @@ module SimplyBased
|
|||||||
# All three use the same cluster/bucket as your models — one datastore.
|
# All three use the same cluster/bucket as your models — one datastore.
|
||||||
|
|
||||||
# Server-side atomic counters (Couchbase binary increment/decrement). A drop-in
|
# Server-side atomic counters (Couchbase binary increment/decrement). A drop-in
|
||||||
# for Redis INCR/DECR.
|
# for Redis INCR/INCRBY.
|
||||||
#
|
#
|
||||||
# counter = SimplyBased::Counter.new
|
# counter = SimplyBased::Counter.new
|
||||||
# counter.increment("jobs:processed") # => 1 (created at 0, +1)
|
# counter.increment("jobs:processed") # => 1 (like Redis INCR on a missing key)
|
||||||
# counter.increment("jobs:processed", 5) # => 6
|
# counter.increment("jobs:processed", 5) # => 6
|
||||||
# counter.get("jobs:processed") # => 6
|
# counter.get("jobs:processed") # => 6
|
||||||
|
#
|
||||||
|
# Divergence from Redis: Couchbase counters are unsigned 64-bit —
|
||||||
|
# decrementing below zero clamps at 0 (Redis DECR would go negative).
|
||||||
class Counter
|
class Counter
|
||||||
def initialize(store: nil)
|
def initialize(store: nil)
|
||||||
@store = store
|
@store = store
|
||||||
@@ -22,10 +25,16 @@ module SimplyBased
|
|||||||
@store ||= SimplyBased.database
|
@store ||= SimplyBased.database
|
||||||
end
|
end
|
||||||
|
|
||||||
def increment(key, by = 1, initial: 0, ttl: nil)
|
# Matches Redis INCRBY semantics: incrementing a missing key seeds it at
|
||||||
store.counter_incr(key, by, initial: initial, ttl: ttl)
|
# `by` (Couchbase's native increment would seed at `initial` and *ignore*
|
||||||
|
# the delta on creation — the first INCR would return 0, silently
|
||||||
|
# undercounting). Pass an explicit `initial:` to override.
|
||||||
|
def increment(key, by = 1, initial: nil, ttl: nil)
|
||||||
|
store.counter_incr(key, by, initial: initial || by, ttl: ttl)
|
||||||
end
|
end
|
||||||
|
|
||||||
|
# A missing key seeds at `initial` (default 0): Couchbase counters cannot
|
||||||
|
# go negative, so Redis DECRBY parity (-by) is impossible here.
|
||||||
def decrement(key, by = 1, initial: 0, ttl: nil)
|
def decrement(key, by = 1, initial: 0, ttl: nil)
|
||||||
store.counter_decr(key, by, initial: initial, ttl: ttl)
|
store.counter_decr(key, by, initial: initial, ttl: ttl)
|
||||||
end
|
end
|
||||||
@@ -101,15 +110,28 @@ module SimplyBased
|
|||||||
end
|
end
|
||||||
|
|
||||||
# Returns a CAS token (the lock handle) or nil if already held.
|
# Returns a CAS token (the lock handle) or nil if already held.
|
||||||
|
# A positive ttl is mandatory: without expiry, a crashed holder would leave
|
||||||
|
# the lock stuck forever.
|
||||||
def acquire(key, ttl: 30)
|
def acquire(key, ttl: 30)
|
||||||
|
raise ArgumentError, "lock ttl must be a positive number of seconds (got #{ttl.inspect})" unless ttl.respond_to?(:positive?) && ttl.positive?
|
||||||
store.kv_add(lock_key(key), { 'locked_at' => Time.now.to_i }, ttl: ttl)
|
store.kv_add(lock_key(key), { 'locked_at' => Time.now.to_i }, ttl: ttl)
|
||||||
end
|
end
|
||||||
|
|
||||||
# Release only if we still hold the same CAS (no stomping a re-acquired lock).
|
# Release only if we still hold the same CAS (no stomping a lock that
|
||||||
def release(key, cas = nil)
|
# expired and was re-acquired by someone else). The handle is required —
|
||||||
|
# a nil CAS used to delete the lock document unconditionally, letting any
|
||||||
|
# caller unlock a lock it never held. Returns false when the handle is
|
||||||
|
# stale (someone else holds the lock now).
|
||||||
|
def release(key, cas)
|
||||||
|
raise ArgumentError, 'release requires the CAS handle returned by acquire (use force_release! to override)' if cas.nil?
|
||||||
store.kv_delete(lock_key(key), cas: cas)
|
store.kv_delete(lock_key(key), cas: cas)
|
||||||
end
|
end
|
||||||
|
|
||||||
|
# Unconditional unlock — administrative override only.
|
||||||
|
def force_release!(key)
|
||||||
|
store.kv_delete(lock_key(key))
|
||||||
|
end
|
||||||
|
|
||||||
def with_lock(key, ttl: 30, retries: 0, wait: 0.1)
|
def with_lock(key, ttl: 30, retries: 0, wait: 0.1)
|
||||||
cas = acquire(key, ttl: ttl)
|
cas = acquire(key, ttl: ttl)
|
||||||
tries = 0
|
tries = 0
|
||||||
|
|||||||
@@ -6,11 +6,16 @@ RSpec.describe 'Redis replacements' do
|
|||||||
describe SimplyBased::Counter do
|
describe SimplyBased::Counter do
|
||||||
subject(:counter) { described_class.new }
|
subject(:counter) { described_class.new }
|
||||||
|
|
||||||
it 'atomically increments, seeding at the initial value' do
|
it 'atomically increments with Redis INCR semantics (first INCR of a missing key returns the delta)' do
|
||||||
expect(counter.increment('c:hits')).to eq(0) # created at initial 0
|
|
||||||
expect(counter.increment('c:hits')).to eq(1)
|
expect(counter.increment('c:hits')).to eq(1)
|
||||||
expect(counter.increment('c:hits', 5)).to eq(6)
|
expect(counter.increment('c:hits')).to eq(2)
|
||||||
expect(counter.get('c:hits')).to eq(6)
|
expect(counter.increment('c:hits', 5)).to eq(7)
|
||||||
|
expect(counter.get('c:hits')).to eq(7)
|
||||||
|
end
|
||||||
|
|
||||||
|
it 'seeds at an explicit initial value when given' do
|
||||||
|
expect(counter.increment('c:seeded', 1, initial: 100)).to eq(100)
|
||||||
|
expect(counter.increment('c:seeded')).to eq(101)
|
||||||
end
|
end
|
||||||
|
|
||||||
it 'decrements and resets' do
|
it 'decrements and resets' do
|
||||||
@@ -19,6 +24,11 @@ RSpec.describe 'Redis replacements' do
|
|||||||
expect(counter.reset('c:n', 0)).to eq(0)
|
expect(counter.reset('c:n', 0)).to eq(0)
|
||||||
expect(counter.get('c:n')).to eq(0)
|
expect(counter.get('c:n')).to eq(0)
|
||||||
end
|
end
|
||||||
|
|
||||||
|
it 'clamps decrement at zero (unsigned Couchbase counters; Redis would go negative)' do
|
||||||
|
counter.reset('c:clamp', 1)
|
||||||
|
expect(counter.decrement('c:clamp', 5)).to eq(0)
|
||||||
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
describe SimplyBased::Cache do
|
describe SimplyBased::Cache do
|
||||||
@@ -45,6 +55,19 @@ RSpec.describe 'Redis replacements' do
|
|||||||
expect(lock.acquire('job:1', ttl: 30)).not_to be_nil # released → free
|
expect(lock.acquire('job:1', ttl: 30)).not_to be_nil # released → free
|
||||||
end
|
end
|
||||||
|
|
||||||
|
it 'refuses to release without the CAS handle' do
|
||||||
|
lock.acquire('job:guard', ttl: 30)
|
||||||
|
expect { lock.release('job:guard', nil) }.to raise_error(ArgumentError)
|
||||||
|
expect(lock.acquire('job:guard', ttl: 30)).to be_nil # still held
|
||||||
|
lock.force_release!('job:guard')
|
||||||
|
expect(lock.acquire('job:guard', ttl: 30)).not_to be_nil
|
||||||
|
end
|
||||||
|
|
||||||
|
it 'rejects a missing/zero ttl (a crashed holder must never leave an immortal lock)' do
|
||||||
|
expect { lock.acquire('job:immortal', ttl: nil) }.to raise_error(ArgumentError)
|
||||||
|
expect { lock.acquire('job:immortal', ttl: 0) }.to raise_error(ArgumentError)
|
||||||
|
end
|
||||||
|
|
||||||
it 'runs a critical section with with_lock' do
|
it 'runs a critical section with with_lock' do
|
||||||
ran = false
|
ran = false
|
||||||
lock.with_lock('job:2') { ran = true }
|
lock.with_lock('job:2') { ran = true }
|
||||||
|
|||||||
Reference in new issue
Block a user