Class: Wurk::Leader

Inherits:
Object
  • Object
show all
Defined in:
lib/wurk/leader.rb

Overview

Cluster leader election via Redis SET NX EX. Single-leader-per-cluster is best-effort (not Raft): a partitioned ex-leader can briefly co-exist with a new one until the TTL expires. Callers that need strict mutual exclusion must idempotency-guard their writes (see #token).

Wire-compat: the cluster lock lives at dear-leader (STRING, EX≈30s) holding the <hostname>:<pid>:<process_nonce> identity. Each fresh gain also pulls a monotonic fencing token via INCR leader-token, exposed on #token. Wurk goes a small step beyond Sidekiq Enterprise (which deliberately does not expose fencing) so downstream code that can benefit from a guard has one available; the token is best-effort too — it is never re-read on subsequent acquires, only on transitions.

Cadence per spec: renew every 15s while leader, recheck every 60s as follower, lock TTL 30s. Opt out a process from campaigning entirely with WURK_LEADER=false (or its Sidekiq alias SIDEKIQ_LEADER=false), useful for hot-standby pools.

Spec: docs/target/sidekiq-ent.md §6.

Constant Summary collapse

DEFAULT_KEY = 
'dear-leader'
TOKEN_KEY = 
'leader-token'
DEFAULT_TTL = 
30
DEFAULT_RENEW_INTERVAL = 
15
DEFAULT_FOLLOWER_INTERVAL = 
60
OPT_OUT_ENV =

native opt-out env

'WURK_LEADER'
SIDEKIQ_OPT_OUT_ENV =

Sidekiq Ent drop-in alias (§6.2/§7.2)

'SIDEKIQ_LEADER'
THREAD_NAME = 
'wurk-leader'

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(config: nil, key: DEFAULT_KEY, ttl: DEFAULT_TTL, renew_interval: DEFAULT_RENEW_INTERVAL, follower_interval: DEFAULT_FOLLOWER_INTERVAL, pool: nil, owner: nil) ⇒ Leader

rubocop:disable Metrics/ParameterLists



48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
 
# File 'lib/wurk/leader.rb', line 48

def initialize(config: nil, key: DEFAULT_KEY, ttl: DEFAULT_TTL, # rubocop:disable Metrics/ParameterLists
               renew_interval: DEFAULT_RENEW_INTERVAL,
               follower_interval: DEFAULT_FOLLOWER_INTERVAL,
               pool: nil, owner: nil)
  @config = config
  @key = key
  @ttl = ttl
  @renew_interval = renew_interval
  @follower_interval = follower_interval
  @pool = pool
  @owner = owner || cluster_identity
  @held = false
  @token = nil
  @thread = nil
  @done = false
  @mutex = ::Mutex.new
  @sleeper = ::ConditionVariable.new
end

Instance Attribute Details

#config ⇒ Object (readonly)

Returns the value of attribute config.



46
47
48
 
# File 'lib/wurk/leader.rb', line 46

def config
  @config
end

#key ⇒ Object (readonly)

Returns the value of attribute key.



46
47
48
 
# File 'lib/wurk/leader.rb', line 46

def key
  @key
end

#owner ⇒ Object (readonly)

Returns the value of attribute owner.



46
47
48
 
# File 'lib/wurk/leader.rb', line 46

def owner
  @owner
end

#token ⇒ Object (readonly)

Returns the value of attribute token.



46
47
48
 
# File 'lib/wurk/leader.rb', line 46

def token
  @token
end

#ttl ⇒ Object (readonly)

Returns the value of attribute ttl.



46
47
48
 
# File 'lib/wurk/leader.rb', line 46

def ttl
  @ttl
end

Class Method Details

.opted_out? ⇒ Boolean

True when this process has opted out of campaigning via WURK_LEADER=false or its Sidekiq alias SIDEKIQ_LEADER=false (hot-standby pools that must never lead). Either env name works.

Returns:

  • (Boolean) 


42
43
44
 
# File 'lib/wurk/leader.rb', line 42

def self.opted_out?
  [OPT_OUT_ENV, SIDEKIQ_OPT_OUT_ENV].any? { |k| ENV[k].to_s.downcase == 'false' }
end

Instance Method Details

#acquire ⇒ Object

SET NX EX. If the key already holds our owner string (rare — same process re-entering after a hiccup), refresh via EXPIRE so leadership doesn't lapse. On any follower → leader transition, INCR the global leader-token so the new token is strictly greater than every prior leader's, then dispatch the :leader lifecycle event.



79
80
81
82
83
84
85
86
87
88
89
90
 
# File 'lib/wurk/leader.rb', line 79

def acquire # rubocop:disable Naming/PredicateMethod
  return false if disabled?

  transition = run_set_or_refresh
  return false unless transition

  if transition == :gained
    @token = redis_call { |c| c.call('INCR', TOKEN_KEY) }
    dispatch_leader_event
  end
  true
end

#disabled? ⇒ Boolean

WURK_LEADER=false (or SIDEKIQ_LEADER=false) makes acquire a no-op and leader? permanently false; the renewal thread also refuses to start. Useful for hot-standby pools that must never campaign.

Returns:

  • (Boolean) 


70
71
72
 
# File 'lib/wurk/leader.rb', line 70

def disabled?
  self.class.opted_out?
end

#leader? ⇒ Boolean

Returns:

  • (Boolean) 


113
114
115
 
# File 'lib/wurk/leader.rb', line 113

def leader?
  @held
end

#release ⇒ Object

CAS DEL — only drop the key if we still own it, otherwise a stale release would yank leadership from whichever follower took over. GET-then-DEL is not that CAS: our key can lapse between the two commands and a follower can win the election in the gap, whereupon the bare DEL deletes its lock and leaves the cluster leaderless (or, once the ex-leader re-campaigns, doubly led) for up to one renew interval — double cron fires, double rollups. The compare and the delete happen in one server-side step instead, via the script Unique.release_if_owner already shares.

idempotent: — a replay after a lost reply finds the key either already gone or no longer ours, so it can only be a no-op.



104
105
106
107
108
109
110
111
 
# File 'lib/wurk/leader.rb', line 104

def release
  redis_call(idempotent: true) do |c|
    Wurk::Lua::Loader.eval_cached(c, :release_if_owner, keys: [@key], argv: [@owner])
  end
  @held = false
  @token = nil
  nil
end

#running? ⇒ Boolean

Returns:

  • (Boolean) 


150
151
152
 
# File 'lib/wurk/leader.rb', line 150

def running?
  !@thread.nil? && @thread.alive?
end

#start ⇒ Object

Spawns the periodic re-election thread. Idempotent. While leader, the loop re-acquires (refreshing TTL) every renew_interval; while follower, it polls every follower_interval. Caller must invoke stop for orderly shutdown — the thread also releases its lock on exit.

The spawn happens inside the lock: with it outside, two callers racing into start both clear the nil check and both spawn, and the loser's loop is never recorded — an unjoinable second campaigner that outlives stop and keeps renewing the cluster lock.



127
128
129
130
131
132
133
134
135
136
137
 
# File 'lib/wurk/leader.rb', line 127

def start
  return nil if disabled?

  @mutex.synchronize do
    return @thread if @thread

    @done = false
    @thread = spawn_loop_thread
  end
  @thread
end

#stop ⇒ Object 



139
140
141
142
143
144
145
146
147
148
 
# File 'lib/wurk/leader.rb', line 139

def stop
  thread = @mutex.synchronize do
    @done = true
    @sleeper.signal
    @thread
  end
  thread&.join
  @mutex.synchronize { @thread = nil }
  release
end