Class: Wurk::Heartbeat

Inherits:
Object
  • Object
show all
Includes:
Component
Defined in:
lib/wurk/heartbeat.rb

Overview

Single-purpose owner of the Redis-side process heartbeat. Lifted out of Wurk::Launcher so the launcher can stay focused on lifecycle and so the heartbeat schema lives in one place readers can grep for.

Each beat is one pipelined round-trip:

SADD   processes <identity>
HSET   <identity>          info concurrency busy beat quiet rss rtt_us
EXPIRE <identity>          60
UNLINK <identity>:work
HSET   <identity>:work     <tid> <json> ...    (only if WORK_STATE non-empty)
EXPIRE <identity>:work     60                  (only if WORK_STATE non-empty)
LPOP   <identity>-signals  × BEAT_PAUSE

The work hash is UNLINK-then-rewritten on every beat — a dropped beat momentarily empties it, and ProcessSet#cleanup compensates by SREM-ing identities whose info field has expired.

Heartbeat owns only the wire. Signals queued by the dashboard are pulled out of the pipelined results and returned to the caller for dispatch — TSTP/TERM semantics live in Launcher.

:heartbeat fires once on the first successful beat and again after any network partition (rearmed when beat! rescues). :beat fires every tick.

Spec: docs/target/sidekiq-free.md §12 (Launcher#❤️).

Constant Summary collapse

BEAT_PAUSE =

Cadence in seconds. Key TTL is 60s — a process is dead after ~6 misses.

10
TTL_SECONDS = 
60

Constants included from Component

Component::DEFAULT_THREAD_PRIORITY, Component::PROCESS_NONCE

Instance Attribute Summary collapse

Attributes included from Component

#config

Instance Method Summary collapse

Methods included from Component

#default_tag, #fire_event, #handle_exception, #hostname, #leader?, #logger, #mono_ms, #process_nonce, #real_ms, #redis, #safe_thread, #tid, #watchdog

Constructor Details

#initialize(identity:, config:, started_at: nil, embedded: false, quiet: nil) ⇒ Heartbeat

quiet: is a callable so Launcher can keep ownership of its @done flag without an awkward setter contract. info_overrides: lets the caller (Launcher#embedded, tests) inject fields without forcing Heartbeat to know about every flag the host process tracks.



48
49
50
51
52
53
54
55
56
 
# File 'lib/wurk/heartbeat.rb', line 48

def initialize(identity:, config:, started_at: nil, embedded: false, quiet: nil)
  @identity = identity
  @config = config
  @started_at = started_at || Time.now.to_f
  @embedded = embedded
  @quiet_proc = quiet || -> { false }
  @first_heartbeat = true
  @rtt_us = 0
end

Instance Attribute Details

#identity ⇒ Object (readonly)

Returns the value of attribute identity.



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

def identity
  @identity
end

#last_beat_at ⇒ Object (readonly)

Returns the value of attribute last_beat_at.



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

def last_beat_at
  @last_beat_at
end

#rtt_us ⇒ Object (readonly)

Returns the value of attribute rtt_us.



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

def rtt_us
  @rtt_us
end

Instance Method Details

#beat! ⇒ Object

Pipelined beat. Returns the drained signals as Array on success, or nil if Redis raised — the next successful beat refires :heartbeat so partition recovery is observable.



61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
 
# File 'lib/wurk/heartbeat.rb', line 61

def beat!
  sigs, @rtt_us = pipelined_beat
  @last_beat_at = Time.now.to_f
  if @first_heartbeat
    @first_heartbeat = false
    fire_event(:heartbeat)
  end
  fire_event(:beat, oneshot: false)
  emit_statsd_gauges
  sigs.compact
rescue StandardError => e
  @first_heartbeat = true
  handle_exception(e, { context: 'heartbeat' })
  nil
end

#stop! ⇒ Object

Best-effort teardown. SREM removes us from the live list; UNLINK drops the identity hash and its work mirror. Idempotent: if Redis is down the next process boot's ProcessSet#cleanup prunes us.



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

def stop!
  redis do |conn|
    conn.pipelined do |pipe|
      pipe.call('SREM', Keys::PROCESSES, @identity)
      pipe.call('UNLINK', @identity, "#{@identity}:work")
    end
  end
rescue StandardError
  nil
end