Class: Cosmo::Job::Processor

Inherits:
Processor show all
Includes:
Sentry::JobProcessorMiddleware
Defined in:
lib/cosmo/job/processor.rb,
sig/cosmo/job/processor.rbs

Constant Summary

Constants included from Sentry::JobProcessorMiddleware

Sentry::JobProcessorMiddleware::NAME_PREFIX, Sentry::JobProcessorMiddleware::OP_NAME, Sentry::JobProcessorMiddleware::SPAN_ORIGIN, Sentry::JobProcessorMiddleware::STATUS_FAIL, Sentry::JobProcessorMiddleware::STATUS_OK

Constants inherited from Processor

Processor::STREAMS_PAUSED_IDLE_SLEEP, Processor::STREAM_EMPTY_BACKOFF_MAX, Processor::STREAM_PAUSED_RECHECK_TTL

Instance Method Summary collapse

Methods inherited from Processor

#client, #consumer_state, #fetch, #initialize, #lock, run, #run, #run_loop, #running?, #stop, #stopwatch, #work_loop

Constructor Details

This class inherits a constructor from Cosmo::Processor

Instance Method Details

#acquire_concurrency_slot(worker_class, message, data) ⇒ ::String, false

Tries to acquire a concurrency slot for the job. Returns the slot key (String) on success, or false if all slots are taken (a message is NAK'd with a delay of retry_in before returning

Parameters:

  • worker_class (Object)
  • message (Object)
  • data (Hash[Symbol, untyped])

Returns:

  • (::String, false) 


128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
 
# File 'lib/cosmo/job/processor.rb', line 128

def acquire_concurrency_slot(worker_class, message, data)
  options = worker_class.concurrency_options
  key = worker_class.concurrency_key(data[:args])

  slot = Limit.instance.acquire(key, jid: data[:jid], limit: options[:limit], duration: options[:duration])
  return slot if slot

  message.nak(delay: Config.to_ns(options[:retry_in]))
  Logger.debug "concurrency limit reached for #{data[:class]}, re-queueing back #{data[:jid]}"
  false
rescue NATS::Error => e
  # Unexpected KV failure (e.g. transient NATS error). NAK immediately so
  # the message is retried rather than stuck in-flight until ack_wait expires.
  Logger.error e
  message.nak
  false
end

#consumer_entry(stream_name) ⇒ [untyped, Hash[Symbol, untyped], untyped]?

Parameters:

  • stream_name (Object)

Returns:

  • ([untyped, Hash[Symbol, untyped], untyped], nil) 


187
188
189
 
# File 'lib/cosmo/job/processor.rb', line 187

def consumer_entry(stream_name)
  @consumers.find { |(_, config, _)| config[:stream].to_s == stream_name.to_s }
end

#consumers ⇒ Array[untyped]

Returns:

  • (Array[untyped]) 


232
233
234
235
 
# File 'lib/cosmo/job/processor.rb', line 232

def consumers
  @weights ||= @consumers.filter_map { |(_, c, _)| [c[:stream]] * [c[:priority].to_i, 1].max }.flatten
  @weights.shuffle.map { |s| @consumers.find { |(_, c, _)| c[:stream] == s } }
end

#default_retry_delay(current_attempt) ⇒ Integer

Parameters:

  • current_attempt (Integer)

Returns:

  • (Integer) 


202
203
204
 
# File 'lib/cosmo/job/processor.rb', line 202

def default_retry_delay(current_attempt)
  (current_attempt**4) + 15
end

#deliver_cap(stream_name, desired_retries) ⇒ Integer?

Returns the consumer's configured max_deliver when it's lower than the job's own configured retry count (so we should give up a bit early instead of NAK'ing into a redelivery that'll never come), or nil when the job's own retry count is already the binding constraint.

Parameters:

  • stream_name (Object)
  • desired_retries (Integer)

Returns:

  • (Integer, nil) 


182
183
184
185
 
# File 'lib/cosmo/job/processor.rb', line 182

def deliver_cap(stream_name, desired_retries)
  max_deliver = consumer_entry(stream_name)&.dig(1, :max_deliver).to_i
  max_deliver if max_deliver.positive? && max_deliver < desired_retries
end

#drop_message(message, data) ⇒ void

This method returns an undefined value.

Parameters:

  • message (Object)
  • data (Hash[Symbol, untyped]) 


215
216
217
218
 
# File 'lib/cosmo/job/processor.rb', line 215

def drop_message(message, data)
  message.term
  Logger.debug "job dropped #{data[:jid]}"
end

#fetch_subjects(config) ⇒ Object

Parameters:

  • config (Hash[Symbol, untyped])

Returns:

  • (Object) 


237
238
239
 
# File 'lib/cosmo/job/processor.rb', line 237

def fetch_subjects(config)
  config[:subject]
end

#fetch_timeout(_config) ⇒ Float

Parameters:

  • config (Hash[Symbol, untyped])

Returns:

  • (Float) 


241
242
243
 
# File 'lib/cosmo/job/processor.rb', line 241

def fetch_timeout(_config)
  ENV.fetch("COSMO_JOBS_FETCH_TIMEOUT", 0.1).to_f
end

#handle_failure(worker_class, message, data, exception) ⇒ Boolean

rubocop:disable Naming/PredicateMethod

Parameters:

  • worker_class (Object)
  • message (Object)
  • data (Hash[Symbol, untyped])
  • exception (Exception, nil)

Returns:

  • (Boolean) 


146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
 
# File 'lib/cosmo/job/processor.rb', line 146

def handle_failure(worker_class, message, data, exception) # rubocop:disable Naming/PredicateMethod
  current_attempt = message.metadata.num_delivered
  desired_retries = data[:retry].to_i + 1
  capped_at = deliver_cap(message.metadata.stream, desired_retries)

  if current_attempt < (capped_at || desired_retries)
    nak_message(worker_class, message, data, current_attempt, exception)
    return false
  end

  warn_capped(message, data, capped_at) if capped_at
  data[:dead] ? move_message(message, data) : drop_message(message, data)
  notify_batch(data, success: false)
  true
end

#move_message(message, data = nil) ⇒ void

This method returns an undefined value.

Parameters:

  • message (Object)
  • data (Hash[Symbol, untyped], nil) (defaults to: nil) 


220
221
222
223
224
225
226
 
# File 'lib/cosmo/job/processor.rb', line 220

def move_message(message, data = nil)
  klass = data ? Utils::String.underscore(data[:class]) : "default"
  headers = { "X-Stream" => message.metadata.stream, "X-Subject" => message.subject }
  Client.instance.publish("jobs.dead.#{klass}", message.data, header: headers)
  message.ack
  Logger.debug "job moved #{data&.dig(:jid)} to DLQ"
end

#nak_message(worker_class, message, data, current_attempt, exception) ⇒ void

This method returns an undefined value.

The message is NAK'd with an explicit delay (default backoff, or the job class's own retry_in handler).

Parameters:

  • worker_class (Object)
  • message (Object)
  • data (Hash[Symbol, untyped])
  • current_attempt (Integer)
  • exception (Exception, nil) 


169
170
171
 
# File 'lib/cosmo/job/processor.rb', line 169

def nak_message(worker_class, message, data, current_attempt, exception)
  message.nak(delay: Config.to_ns(retry_delay(worker_class, data, current_attempt, exception)))
end

#perform_job(job_instance, data:, message:, duration: nil) ⇒ Object

rubocop:disable Lint/UnusedMethodArgument

Parameters:

  • job_instance (Cosmo::Job)
  • data (Hash)
  • message (NATS::Msg)
  • duration (Float, nil) (defaults to: nil)
  • data: (Hash[Symbol, untyped])
  • message: (Object)
  • duration: (Float, nil) (defaults to: nil)

Returns:

  • (Object) 


257
258
259
260
261
262
263
 
# File 'lib/cosmo/job/processor.rb', line 257

def perform_job(job_instance, data:, message:, duration: nil)
  if duration
    Timeout.timeout(duration) { job_instance.perform(*data[:args]) }
  else
    job_instance.perform(*data[:args])
  end
end

#process(messages, _) ⇒ void

This method returns an undefined value.

rubocop:disable Metrics/AbcSize, Metrics/MethodLength, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity

Parameters:

  • messages (Array[untyped])
  • processor (Object) 


59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
 
# File 'lib/cosmo/job/processor.rb', line 59

def process(messages, _) # rubocop:disable Metrics/AbcSize, Metrics/MethodLength, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity
  message = messages.first
  Logger.debug "received messages #{messages.inspect}"
  data = Utils::Json.parse(message.data)
  unless data
    Logger.error ArgumentError.new("malformed payload")
    move_message(message)
    return
  end

  worker_class = Utils::String.safe_constantize(data[:class])
  unless worker_class
    Logger.error ArgumentError.new("#{data[:class]} class not found")
    move_message(message, data)
    notify_batch(data, success: false)
    return
  end

  if worker_class.limits_concurrency?
    slot = acquire_concurrency_slot(worker_class, message, data)
    return if slot == false
  end

  duration = worker_class.default_options[:limit]&.dig(:duration)&.to_i

  with_stats(message) do
    sw = stopwatch
    Logger.with(jid: data[:jid])
    Logger.info "start"

    instance = build_worker(worker_class, data, message)
    perform_job(instance, data: data, message: message, duration: duration)

    message.ack
    notify_batch(data, success: true)
    Logger.with(elapsed: sw.elapsed_seconds) { Logger.info "done" }
    true
  rescue Timeout::Error => e
    Logger.with(elapsed: sw.elapsed_seconds) { Logger.info "fail[timeout]" }
    dropped = handle_failure(worker_class, message, data, e)
    false if dropped
  rescue StandardError => e
    Logger.debug e
    Logger.with(elapsed: sw.elapsed_seconds) { Logger.info "fail[error]" }
    dropped = handle_failure(worker_class, message, data, e)
    false if dropped
  rescue Exception # rubocop:disable Lint/RescueException
    Logger.with(elapsed: sw.elapsed_seconds) { Logger.info "fail[exception]" }
    raise
  end
ensure
  Limit.instance.release(slot) if slot
  Logger.without(:jid)
  Logger.debug "processed message #{message.inspect}"
end

#retry_delay(worker_class, data, current_attempt, exception) ⇒ Numeric

Parameters:

  • worker_class (Object)
  • data (Hash[Symbol, untyped])
  • current_attempt (Integer)
  • exception (Exception, nil)

Returns:

  • (Numeric) 


191
192
193
194
195
196
197
198
199
200
 
# File 'lib/cosmo/job/processor.rb', line 191

def retry_delay(worker_class, data, current_attempt, exception)
  handler = worker_class.retry_in(data)
  return default_retry_delay(current_attempt) unless handler

  delay = handler.call(current_attempt, exception)
  delay.is_a?(Numeric) && delay.positive? ? delay : default_retry_delay(current_attempt)
rescue StandardError => e
  Logger.error e
  default_retry_delay(current_attempt)
end

#schedule_loop ⇒ void

This method returns an undefined value.

rubocop:disable Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity, Metrics/MethodLength, Metrics/AbcSize



24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
 
# File 'lib/cosmo/job/processor.rb', line 24

def schedule_loop # rubocop:disable Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity, Metrics/MethodLength, Metrics/AbcSize
  config = Config.dig(:consumers, :jobs, :scheduled)
  return unless config

  subscription, = subscribe(:scheduled, config)
  while running?
    break unless running?

    now = Time.now.to_i
    timeout = ENV.fetch("COSMO_JOBS_SCHEDULER_FETCH_TIMEOUT", 5).to_f
    messages = fetch(subscription, batch_size: 100, timeout:)
    messages&.each do |message|
      headers = message.header.except("X-Stream", "X-Subject", "X-Execute-At", "Nats-Expected-Stream")
      stream, subject, execute_at = message.header.values_at("X-Stream", "X-Subject", "X-Execute-At")
      headers["Nats-Expected-Stream"] = stream
      execute_at = execute_at.to_i

      if now >= execute_at
        client.publish(subject, message.data, headers: headers)
        message.ack
      else
        message.nak(delay: Config.to_ns(execute_at - now))
      end
    rescue StandardError => e
      # A transient failure here (e.g. a JetStream publish timeout) must not be allowed
      # to escape #each and kill this thread — schedule_loop only runs once per processor,
      # so an unhandled exception would silently stop all future scheduled-job dispatch.
      Logger.error e
      message.nak rescue nil
    end

    break unless running?
  end
end

#scheduler? ⇒ Boolean

Returns:

  • (Boolean) 


228
229
230
 
# File 'lib/cosmo/job/processor.rb', line 228

def scheduler?
  true
end

#setup ⇒ void

This method returns an undefined value.



10
11
12
13
14
15
16
17
18
19
20
21
22
 
# File 'lib/cosmo/job/processor.rb', line 10

def setup
  # Initialize singletons before starting to process messages
  API::Busy.instance
  API::Counter.instance
  Limit.instance

  jobs_config = Config.dig(:consumers, :jobs)
  jobs_config&.each do |stream_name, config|
    next if stream_name == :scheduled # scheduled jobs are handled in schedule_loop

    @consumers << subscribe(stream_name, config)
  end
end

#subscribe(stream_name, config) ⇒ [untyped, Hash[Symbol, untyped], nil]

Parameters:

  • stream_name (Symbol)
  • config (Hash[Symbol, untyped])

Returns:

  • ([untyped, Hash[Symbol, untyped], nil]) 


206
207
208
209
210
211
212
213
 
# File 'lib/cosmo/job/processor.rb', line 206

def subscribe(stream_name, config)
  config = config.dup
  config[:batch_size] = 1
  config[:stream] = stream_name
  config[:consumer] = "consumer-#{stream_name}"
  subscription = client.subscribe(config[:subject], config[:consumer], config.except(:subject, :priority, :stream, :batch_size, :consumer))
  [subscription, config, nil]
end

#warn_capped(message, data, capped_at) ⇒ void

This method returns an undefined value.

Parameters:

  • message (Object)
  • data (Hash[Symbol, untyped])
  • capped_at (Integer) 


173
174
175
176
177
 
# File 'lib/cosmo/job/processor.rb', line 173

def warn_capped(message, data, capped_at)
  consumer_name = consumer_entry(message.metadata.stream)&.dig(1, :consumer)
  Logger.warn "#{data[:class]} configured retry: #{data[:retry]} exceeds max_deliver: #{capped_at} " \
              "on #{consumer_name}; giving up early to avoid a stranded message"
end

#with_stats(message) { ... } ⇒ void

This method returns an undefined value.

Parameters:

  • message (Object)

Yields:

Yield Returns:

  • (Object) 


245
246
247
248
249
 
# File 'lib/cosmo/job/processor.rb', line 245

def with_stats(message, &block)
  API::Busy.instance.with(message) do
    API::Counter.instance.with(&block)
  end
end