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) 


116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
 
# File 'lib/cosmo/job/processor.rb', line 116

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: options[:retry_in] * Config::NANO)
  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

#consumers ⇒ Array[untyped]

Returns:

  • (Array[untyped]) 


177
178
179
180
 
# File 'lib/cosmo/job/processor.rb', line 177

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

#drop_message(message, data) ⇒ void

This method returns an undefined value.

Parameters:

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


160
161
162
163
 
# File 'lib/cosmo/job/processor.rb', line 160

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) 


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

def fetch_subjects(config)
  config[:subject]
end

#fetch_timeout(_config) ⇒ Float

Parameters:

  • config (Hash[Symbol, untyped])

Returns:

  • (Float) 


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

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

#handle_failure(message, data) ⇒ Boolean

rubocop:disable Naming/PredicateMethod

Parameters:

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

Returns:

  • (Boolean) 


134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
 
# File 'lib/cosmo/job/processor.rb', line 134

def handle_failure(message, data) # rubocop:disable Naming/PredicateMethod
  current_attempt = message.metadata.num_delivered
  max_retries = data[:retry].to_i + 1

  if current_attempt < max_retries
    # NATS will auto-retry with delay (exponential backoff based on current attempt).
    # When max_deliver is reached, NATS stops redelivering the message and marks it as "max deliveries exceeded".
    # The message is effectively abandoned by NATS — it stays in the stream (consuming a slot) but will never be delivered again to that consumer.
    delay_ns = ((current_attempt**4) + 15) * Config::NANO
    message.nak(delay: delay_ns)
    return false
  end

  data[:dead] ? move_message(message, data) : drop_message(message, data)
  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) 


165
166
167
168
169
170
171
 
# File 'lib/cosmo/job/processor.rb', line 165

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

#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) 


54
55
56
57
58
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
 
# File 'lib/cosmo/job/processor.rb', line 54

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)
    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 = worker_class.new.tap do |worker|
      worker.jid = data[:jid]
      worker.enqueued_at = message.metadata.timestamp
      worker.attempt = message.metadata.num_delivered
      worker.scheduled_by = message.header&.dig("Nats-Scheduler")
    end
    perform_job(instance, data: data, message: message, duration: duration)

    message.ack
    Logger.with(elapsed: sw.elapsed_seconds) { Logger.info "done" }
    true
  rescue Timeout::Error
    Logger.with(elapsed: sw.elapsed_seconds) { Logger.info "fail[timeout]" }
    dropped = handle_failure(message, data)
    false if dropped
  rescue StandardError => e
    Logger.debug e
    Logger.with(elapsed: sw.elapsed_seconds) { Logger.info "fail[error]" }
    dropped = handle_failure(message, data)
    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

#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
 
# 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
        delay_ns = (execute_at - now) * 1_000_000_000
        message.nak(delay: delay_ns)
      end
    end

    break unless running?
  end
end

#scheduler? ⇒ Boolean

Returns:

  • (Boolean) 


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

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]) 


151
152
153
154
155
156
157
158
 
# File 'lib/cosmo/job/processor.rb', line 151

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

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

This method returns an undefined value.

Parameters:

  • message (Object)

Yields:

Yield Returns:

  • (Object) 


190
191
192
193
194
 
# File 'lib/cosmo/job/processor.rb', line 190

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