Ruby NATS worker: n threads pulling from m streams

nats_worker.rb

# bin/nats_worker
require_relative "../config/environment"
Rails.application.eager_load!

def backoff_delay(attempt)
  # 1s, 2s, 4s… capped at 3 days, with 50–100% jitter
  [2**(attempt - 1), 3 * 24 * 3600].min * (0.5 + rand * 0.5)
end

nc = NATS.connect(ENV.fetch("NATS_URL"))
js = nc.jetstream
stopping = false
%w[TERM INT].each { |sig| trap(sig) { stopping = true } }

threads = Integer(ENV.fetch("WORKER_THREADS", 5)).times.map do
  Thread.new do
    sub = js.pull_subscribe(nil, "rails-workers", stream: "EVENTS")
    until stopping
      begin
        msg = sub.fetch(1, timeout: 2).first
        job_metadata = { subject: msg.subject, attempt: msg.metadata.num_delivered }
        begin
          Rails.application.executor.wrap { Handlers.dispatch(msg.subject, msg.data) }
          msg.ack
          O11y.log(job_metadata.merge(status: :ok))
        rescue => e
          O11y.log(job_metadata.merge(status: :error, error: e.class.name))
          ErrorTracker.notify(e, job_metadata)
          msg.nak(delay: backoff_delay(job_metadata[:attempt]))
        end
      rescue NATS::IO::Timeout
        # nothing to fetch, loop again
      end
    end
  end
end

threads.each(&:join)
nc.drain