Module: Cosmo::Job::ClassMethods

Defined in:
lib/cosmo/job.rb,
sig/cosmo/job.rbs

Instance Method Summary collapse

Instance Method Details

#client ⇒ Client

Returns:



145
146
147
 
# File 'lib/cosmo/job.rb', line 145

def client
  @client ||= Client.instance
end

#concurrency_key(args) ⇒ ::String?

Derive the fully-scoped concurrency key for a given args array.

Parameters:

  • args (Array[untyped])

Returns:

  • (::String, nil) 


87
88
89
90
91
92
93
94
 
# File 'lib/cosmo/job.rb', line 87

def concurrency_key(args)
  config = concurrency_options
  return unless config

  base = Utils::String.underscore(name)
  suffix = config[:key]&.call(*args)
  suffix ? "#{base}/#{suffix}" : base
end

#concurrency_options ⇒ { limit: Integer, key: Proc?, duration: Integer }?

Returns a normalized concurrency config hash, or nil when not configured. Always contains :limit, :key, :duration, and :retry_in.

Returns:

  • ({ limit: Integer, key: Proc?, duration: Integer }, nil) 


73
74
75
76
77
78
79
80
81
82
83
84
 
# File 'lib/cosmo/job.rb', line 73

def concurrency_options
  value = default_options.dig(:limit, :concurrency)
  return unless value

  duration = default_options.dig(:limit, :duration).to_i
  retry_in = default_options.dig(:limit, :retry_in)&.to_i || (duration / 2)

  case value
  when Integer then { limit: value, key: nil, duration: duration, retry_in: retry_in }
  when Hash    then { limit: value.fetch(:to), key: value[:key], duration: duration, retry_in: retry_in }
  end
end

#default_options ⇒ Hash[Symbol, untyped]

Returns:

  • (Hash[Symbol, untyped]) 


139
140
141
 
# File 'lib/cosmo/job.rb', line 139

def default_options
  @default_options ||= (superclass.respond_to?(:default_options) ? superclass.default_options : Data::DEFAULTS.merge(retry: Data.default_retry)).dup
end

#limits_concurrency? ⇒ Boolean

Returns:

  • (Boolean) 


60
61
62
 
# File 'lib/cosmo/job.rb', line 60

def limits_concurrency?
  !!concurrency_options
end

#options(**config) ⇒ Hash[Symbol, untyped] Also known as: cosmo_options

Parameters:

  • config (Hash) —

    a customizable set of options

Options Hash (**config):

  • :stream (Symbol) —

    NATS stream to publish to (default: :default)

  • :retry (Integer, Boolean) —

    max delivery attempts before giving up (default: max_retries from cosmo.yml, or 3 if unset). false is treated as 0 (no retries). Should stay comfortably under the assigned stream's consumer max_deliver (a coarse, shared safety ceiling, not a per-job budget) -- a job whose retry: exceeds it is capped and dead-lettered a delivery early, with a warning logged.

  • :dead (Boolean) —

    move to dead-letter stream after retries exhausted (default: true)

  • :limit (Hash) —

    execution limits:

    limit: { duration: 30 } limit: { duration: 30, concurrency: 3 } limit: { duration: 30, concurrency: { to: 3, key: ->(id) { id } } } limit: { duration: 30, concurrency: 3, retry_in: 5 }

  • :"limit[:duration]" (Integer) —

    hard execution timeout in seconds. The job thread is killed after this many seconds and counts as a failed attempt (retried with exponential backoff, moved to DLQ after retries exhausted).

  • :"limit[:concurrency]" (Integer, Hash) —

    caps how many instances run at once across all workers. Jobs that cannot acquire a slot are NAK'd (see retry_in) so they are not re-delivered until the slot is likely free. Requires duration. Pass an Integer for a class-wide cap, or { to: N, key: ->(args) {} } to scope per key.

  • :"limit[:retry_in]" (Integer) —

    seconds to wait before NATS redelivers a job that was NAK'd for lack of a concurrency slot (default: half of duration). Counts against the same delivery counter as any other retry -- a job stuck behind the concurrency limit for enough consecutive attempts is dropped/DLQ'd exactly like one that keeps failing outright.

  • :retry_in (Proc) —

    ->(count, exception) { } returns a number of seconds to wait before redelivering a failed job. count is a 1-based delivery attempt that just failed. Falls back to the default backoff (attempt**4 + 15 seconds) if not set or the proc returns a non-numeric/non-positive value, or if it raises.

    Caveat when combined with limit[:concurrency]: when there's no free slot to run in, the message is put back on the stream using the limit[:retry_in] delay described above, and that counts as an attempt too -- the same count goes up whether the job was turned away for lack of a free slot (via limit[:retry_in]) or actually ran and failed. So a job that gets turned away twice for lack of a slot, then finally runs and fails, calls this handler with count == 3, not 1. Don't read count as "how many times perform has actually run and failed" when concurrency limits are in play.

Returns:

  • (Hash[Symbol, untyped])

Raises:



49
50
51
52
53
54
55
56
57
 
# File 'lib/cosmo/job.rb', line 49

def options(**config)
  if config[:limit] && config.dig(:limit, :concurrency) && !config.dig(:limit, :duration).to_i.positive?
    raise ArgumentError, "limit: duration is required when concurrency is set"
  end

  raise ArgumentError, "retry_in must be callable, e.g. ->(count, exception) { ... }" if config[:retry_in] && !config[:retry_in].respond_to?(:call)

  default_options.merge!(config)
end

#perform(*args, async: true, **options) ⇒ ::String?

Parameters:

  • args (Object)
  • async: (Boolean) (defaults to: true)
  • options (Object)

Returns:

  • (::String, nil) 


96
97
98
99
100
101
102
103
104
105
106
107
108
109
 
# File 'lib/cosmo/job.rb', line 96

def perform(*args, async: true, **options)
  batch = Batch.current if async
  options[:batch_id] = batch.bid if batch
  data = Data.new(name, args, default_options.merge(options))
  unless async
    payload = Utils::Json.parse(data.to_args[1])
    raise ArgumentError, "Cannot parse payload" unless payload

    new.perform(*payload[:args])
    return
  end

  publish(data, batch)
end

#perform_async(*args) ⇒ ::String

Parameters:

  • args (Object)

Returns:

  • (::String) 


123
124
125
 
# File 'lib/cosmo/job.rb', line 123

def perform_async(*args)
  perform(*args)
end

#perform_at(timestamp, *args) ⇒ ::String

Parameters:

  • timestamp (Integer, Time)
  • args (Object)

Returns:

  • (::String) 


127
128
129
 
# File 'lib/cosmo/job.rb', line 127

def perform_at(timestamp, *args)
  perform(*args, at: timestamp)
end

#perform_in(interval, *args) ⇒ ::String

Parameters:

  • interval (Integer, Float)
  • args (Object)

Returns:

  • (::String) 


131
132
133
 
# File 'lib/cosmo/job.rb', line 131

def perform_in(interval, *args)
  perform(*args, in: interval)
end

#perform_sync(*args) ⇒ void

This method returns an undefined value.

Parameters:

  • args (Object) 


135
136
137
 
# File 'lib/cosmo/job.rb', line 135

def perform_sync(*args)
  perform(*args, async: false)
end

#publish(data, batch) ⇒ void

This method returns an undefined value.

The batch is reserved a pending slot before we know the publish will succeed (must happen in that order -- see Batch#jobs). Roll it back if it never actually made it onto the stream, so the batch doesn't hang waiting for a completion that will never arrive.

Parameters:



115
116
117
118
119
120
121
 
# File 'lib/cosmo/job.rb', line 115

def publish(data, batch)
  batch&.register_job!
  Publisher.publish_job(data)
rescue StandardError
  batch&.rollback_job!
  raise
end

#retry_in(_data = nil) ⇒ Proc?

Returns the retry_in Proc/lambda (taking (count, exception)) configured for this job class, or nil when unset. Overridable by wrapper job classes (e.g. the ActiveJob executor) that need to resolve it from something other than self.

Parameters:

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

Returns:

  • (Proc, nil) 


67
68
69
 
# File 'lib/cosmo/job.rb', line 67

def retry_in(_data = nil)
  default_options[:retry_in]
end