Class: Zeridion::Flare::Worker::Worker

Inherits:
Object
  • Object
show all
Defined in:
lib/zeridion_flare/worker/worker.rb

Overview

The background-worker runtime — the Ruby analog of the reference SDK's hosted worker service. One bootstrap object:

require "zeridion_flare/worker"

class SendWelcomeEmail
include Zeridion::Flare::Job
flare_options queue: "email", max_attempts: 5, timeout: 120
def perform(payload, ctx) = Mailer.welcome(payload["email"]).deliver_now
end

worker = Zeridion::Flare::Worker.new(concurrency: 5)
worker.run   # blocks; SIGTERM/SIGINT → graceful drain

Lifecycle:

#start — spawn the poll loop on a background thread, return immediately.
#run   — #start, install SIGTERM/SIGINT handlers, then block until a
       signal arrives, then #stop. (Signals are wired ONLY inside #run.)
#stop  — idempotent; stop polling and drain in-flight jobs on the
       CURRENT thread within the shutdown grace, then join everything.

The worker OWNS its HTTP clients (decision D5): a poll client with a read timeout > 30 s and no poll-timeout retry, a single-shot heartbeat client, an ack client with at most one retry, and a register client. All four reuse the existing Client transport — the worker does NOT reinvent HTTP.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(api_key: nil, base_url: nil, registry: nil, transport: nil, max_response_bytes: 10 * 1024 * 1024, **config_options) ⇒ Worker

Returns a new instance of Worker.

Parameters:

  • api_key (String, nil) (defaults to: nil) —

    falls back to FLARE_API_KEY.

  • base_url (String, nil) (defaults to: nil) —

    default https://api.zeridion.com.

  • registry (Registry, nil) (defaults to: nil) —

    explicit registry; default = every class that included Zeridion::Flare::Job / RecurringJob so far.

  • transport (Object, nil) (defaults to: nil) —

    HTTP transport (for tests / DI); passed through to every Client the worker builds.

  • max_response_bytes (Integer) (defaults to: 10 * 1024 * 1024) —

    body cap forwarded to each Client.

  • config_options —

    see Configuration#initialize (queues:, concurrency:, poll_interval_s:, default_timeout_s:, default_max_attempts:, shutdown_grace_s:, poll_read_timeout_s:, rpc_timeout_s:, hostname:, logger:, heartbeat_min_ms:).



43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
 
# File 'lib/zeridion_flare/worker/worker.rb', line 43

def initialize(api_key: nil, base_url: nil, registry: nil, transport: nil,
               max_response_bytes: 10 * 1024 * 1024, **config_options)
  @config = Configuration.new(**config_options)
  @registry = registry || Registry.from_declared
  @worker_id = Identity.generate(hostname: @config.hostname)
  @logger = @config.logger
  @stop = CancellationToken.new
  @lifecycle_mutex = Mutex.new
  @started = false
  @stopped = false
  @poll_thread = nil
  @run_latch = CancellationToken.new

  base_url ||= ::Zeridion::Flare::Client::DEFAULT_BASE_URL
  @poll_client = build_poll_client(api_key, base_url, transport, max_response_bytes)
  @heartbeat_client = build_rpc_client(api_key, base_url, transport, max_response_bytes, max_retries: 0)
  @ack_client = build_rpc_client(api_key, base_url, transport, max_response_bytes, max_retries: 1)
  @register_client = build_rpc_client(api_key, base_url, transport, max_response_bytes, max_retries: 0)

  @pool = Pool.new(concurrency: @config.concurrency)
  @executor = Executor.new(registry: @registry, logger: @logger)
  @runner = build_runner
end

Instance Attribute Details

#config ⇒ Configuration (readonly)

Returns:



71
72
73
 
# File 'lib/zeridion_flare/worker/worker.rb', line 71

def config
  @config
end

#worker_id ⇒ String (readonly)

Returns this worker's id (wrk_host_pid_rand).

Returns:

  • (String) —

    this worker's id (wrk_host_pid_rand).



68
69
70
 
# File 'lib/zeridion_flare/worker/worker.rb', line 68

def worker_id
  @worker_id
end

Instance Method Details

#run ⇒ Object

Start, install SIGTERM/SIGINT handlers, and block until a stop signal (or another thread calls #stop). On wake, drain and return. Must run on the main thread (Signal.trap is process-global).



94
95
96
97
98
99
100
101
102
 
# File 'lib/zeridion_flare/worker/worker.rb', line 94

def run
  start
  install_signal_handlers
  # Block the calling thread until stop is requested.
  @run_latch.wait(Float::INFINITY)
  stop
ensure
  uninstall_signal_handlers
end

#start ⇒ Object

Start the pool + poll loop on a background thread. Non-blocking. Idempotent. Returns self.



75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
 
# File 'lib/zeridion_flare/worker/worker.rb', line 75

def start
  @lifecycle_mutex.synchronize do
    return self if @started

    @started = true
    @pool.start
    @poll_thread = Thread.new do
      if Thread.current.respond_to?(:name=)
        Thread.current.name = "flare-poll"
      end
      @runner.run
    end
  end
  self
end

#stop ⇒ Object

Idempotent graceful shutdown: stop polling, then drain in-flight jobs on the CURRENT thread within shutdown_grace_s, then join the poll thread + pool. Safe to call from a signal context or another thread.



107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
 
# File 'lib/zeridion_flare/worker/worker.rb', line 107

def stop
  @lifecycle_mutex.synchronize do
    return if @stopped

    @stopped = true
  end

  # 1. Flip the stop token → poll loop stops claiming new jobs and
  #    cooperatively cancels freshly-claimed ones.
  @stop.cancel!
  # 2. Wake #run if it is blocking.
  @run_latch.cancel!
  # 3. Let the poll loop notice and exit.
  @poll_thread&.join(@config.poll_read_timeout_s + 5)
  # 4. Drain in-flight jobs (each acks itself) within the grace window.
  @logger&.info("shutdown_waiting in_flight=#{@pool.in_flight}") if @pool.in_flight.positive?
  all_done = @pool.shutdown(timeout: @config.shutdown_grace_s)
  @logger&.warn("shutdown_grace_exceeded — some jobs still running") unless all_done
  nil
end

#stopped? ⇒ Boolean

Returns true once #stop has completed (test convenience).

Returns:

  • (Boolean) —

    true once #stop has completed (test convenience).



129
130
131
 
# File 'lib/zeridion_flare/worker/worker.rb', line 129

def stopped?
  @lifecycle_mutex.synchronize { @stopped }
end