Module: Wurk::Component

Included in:
CLI, Wurk::Cron::Poller, Embedded, Fetcher::Reaper, Fetcher::Reliable, Heartbeat, History, JobRetry, Launcher, Manager, Metrics::Flusher, Metrics::QueueRollup, Metrics::Rollup, Processor, Scheduled::Enq, Scheduled::Poller, Scheduled::ReliableEnq, Swarm, Swarm::ChildBoot
Defined in:
lib/wurk/component.rb

Overview

Shared mixin for runtime components (Launcher, Manager, Processor, Fetcher, Scheduler, Cron). Wraps clock readings, identity, thread spawning, lifecycle event dispatch, and exception forwarding so each component stays single-purpose.

Host class must expose #config returning either a Wurk::Configuration or a Wurk::Capsule — both duck-type the methods we delegate to.

Spec: docs/target/sidekiq-free.md §11 (Sidekiq::Component).

Constant Summary collapse

DEFAULT_THREAD_PRIORITY = 
-1
PROCESS_NONCE =

Stable for the life of the process — survives fork (children inherit the same nonce). Identity differs across forks because Process.pid does.

SecureRandom.hex(6)
LEADER_CACHE_TTL_MS =

leader? cache TTL — see the method doc below.

5_000

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Instance Attribute Details

#config ⇒ Object (readonly)

Returns the value of attribute config.



23
24
25
 
# File 'lib/wurk/component.rb', line 23

def config
  @config
end

Class Method Details

.tid ⇒ Object

Base36 thread.object_id ^ pid — the id in every log line and the key each Processor publishes its in-flight job under. Constant for the life of a thread inside one process and read several times per job, so it is memoized per thread (frozen: it is used as a Hash key, and an unfrozen String key is duped on every store).

The pid is memoized alongside it because the thread that calls fork keeps its thread-locals in the child, where the pid — and therefore the tid — has changed. Without the guard a forked child would report the parent's tid and collide with it in <identity>:work.

Thread-local, not Thread#[]: the latter is fiber-local, so a job that runs inside a Fiber (or any Enumerator) would miss the memo and allocate a fresh String on every read — the identity the memo exists to cache is the thread's, and it does not change when a fiber does.



55
56
57
58
59
60
61
62
63
64
 
# File 'lib/wurk/component.rb', line 55

def self.tid
  thread = Thread.current
  memo = thread.thread_variable_get(:wurk_tid)
  pid = ::Process.pid
  return memo[1] if memo && memo[0] == pid

  id = (thread.object_id ^ pid).to_s(36).freeze
  thread.thread_variable_set(:wurk_tid, [pid, id].freeze)
  id
end

Instance Method Details

#default_tag(dir = Dir.pwd) ⇒ Object 



82
83
84
 
# File 'lib/wurk/component.rb', line 82

def default_tag(dir = Dir.pwd)
  File.basename(dir)
end

#fire_event(event, oneshot: true, reverse: false, reraise: false) ⇒ Object

Invokes lifecycle hooks for event. Hooks run in registration order (or LIFO when reverse: true, used for teardown). A raise in one hook is reported via handle_exception and does NOT stop the next hook unless reraise: true (used in tests / fail-fast boot). oneshot: true clears the bucket after dispatch so the event can't fire twice.



159
160
161
162
163
164
165
166
 
# File 'lib/wurk/component.rb', line 159

def fire_event(event, oneshot: true, reverse: false, reraise: false)
  bucket = config[:lifecycle_events][event]
  return if bucket.nil? || bucket.empty?

  iter = reverse ? bucket.reverse : bucket
  iter.each { |hook| run_lifecycle_hook(hook, event, reraise) }
  bucket.clear if oneshot
end

#handle_exception(ex, ctx = {}) ⇒ Object 



96
97
98
 
# File 'lib/wurk/component.rb', line 96

def handle_exception(ex, ctx = {})
  config.handle_exception(ex, ctx)
end

#hostname ⇒ Object 



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

def hostname
  ENV['DYNO'] || Socket.gethostname
end

#identity ⇒ Object 



78
79
80
 
# File 'lib/wurk/component.rb', line 78

def identity
  "#{hostname}:#{::Process.pid}:#{process_nonce}"
end

#leader? ⇒ Boolean

True iff this process currently holds the cluster dear-leader lock. Cached per Component instance for LEADER_CACHE_TTL_MS (~5s): cron and the metrics rollups call this every tick, and an uncached GET would double their Redis traffic at short intervals for no benefit — the lock's own renewal cadence (60s+, spec §6.1) easily tolerates a few-second-stale read. Returns false unconditionally when WURK_LEADER=false (or SIDEKIQ_LEADER=false) is set on the process (opt-out hot-standby). Any Redis error is swallowed → false, so a transient partition can't propagate as an exception into user code.

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

Returns:

  • (Boolean) 


113
114
115
116
117
118
119
120
121
122
 
# File 'lib/wurk/component.rb', line 113

def leader?
  return false if Wurk::Leader.opted_out?

  now = mono_ms
  if @leader_checked_at.nil? || (now - @leader_checked_at) >= LEADER_CACHE_TTL_MS
    @leader_checked_at = now
    @leader_cached = fetch_leader?
  end
  @leader_cached
end

#logger ⇒ Object

--- delegated to config -------------------------------------------



88
89
90
 
# File 'lib/wurk/component.rb', line 88

def logger
  config.logger
end

#mono_ms ⇒ Object 



34
35
36
 
# File 'lib/wurk/component.rb', line 34

def mono_ms
  ::Process.clock_gettime(::Process::CLOCK_MONOTONIC, :millisecond)
end

#process_nonce ⇒ Object 



74
75
76
 
# File 'lib/wurk/component.rb', line 74

def process_nonce
  PROCESS_NONCE
end

#real_ms ⇒ Object

--- clocks ---------------------------------------------------------



30
31
32
 
# File 'lib/wurk/component.rb', line 30

def real_ms
  ::Process.clock_gettime(::Process::CLOCK_REALTIME, :millisecond)
end

#redis(idempotent: false) ⇒ Object 



92
93
94
 
# File 'lib/wurk/component.rb', line 92

def redis(idempotent: false, &)
  config.redis(idempotent:, &)
end

#safe_thread(name, priority: nil, &block) ⇒ Object

Spawns a named thread that runs block under watchdog(name). The parent must retain the returned Thread; otherwise GC may not, but report_on_exception is disabled so we don't double-log on death.

Priority resolution matches Sidekiq (component.rb:44-48): explicit argument, then config.thread_priority, then -1. Ruby's default of 0 buys a 100ms timeslice; each negative step halves it, so -1 keeps a CPU-heavy capsule from starving its siblings for a whole tick.



144
145
146
147
148
149
150
151
152
 
# File 'lib/wurk/component.rb', line 144

def safe_thread(name, priority: nil, &block)
  resolved = priority || config.thread_priority || DEFAULT_THREAD_PRIORITY
  Thread.new do
    Thread.current.name = name
    Thread.current.priority = resolved
    Thread.current.report_on_exception = false
    watchdog(name, &block)
  end
end

#tid ⇒ Object 



66
67
68
 
# File 'lib/wurk/component.rb', line 66

def tid
  Component.tid
end

#watchdog(last_words) ⇒ Object

Wraps a block at a thread boundary: any unhandled exception is reported via handle_exception (so it lands in error_handlers / the log) and then re-raised. last_words is the component label included in the context.



129
130
131
132
133
134
 
# File 'lib/wurk/component.rb', line 129

def watchdog(last_words)
  yield
rescue StandardError => e
  handle_exception(e, { context: last_words })
  raise
end