Class: Cosmo::Batch

Inherits:
Object
  • Object
show all
Extended by:
Dispatcher
Defined in:
lib/cosmo/batch.rb,
lib/cosmo/batch/callback.rb,
lib/cosmo/batch/dispatcher.rb,
sig/cosmo/batch.rbs,
sig/cosmo/batch/callback.rbs,
sig/cosmo/batch/dispatcher.rbs

Overview

Groups jobs together and runs a callback once they've all finished.

batch = Cosmo::Batch.new
batch.jobs { MyJob.perform_async(1); MyJob.perform_async(2) }
batch.on(:complete, MyCallback, user_id: 1)

:complete fires once every job is done, pass or fail. :success only fires if none of them ended up dead-lettered or dropped.

The callback is a plain Ruby class, not a job. It just needs an on_complete/on_success(status, options) method:

class MyCallback
def on_complete(status, options)
  Notifier.notify(options[:user_id], "#{status[:succeeded]}/#{status[:total]} done")
end
end

It still runs on the worker pool, not on the thread that finished the batch - it's wrapped in an internal Batch::Callback job.

status - { bid:, total:, succeeded:, failed: }
options - arbitrary custom arguments passed to #on

Only jobs enqueued inside a #jobs block count. Nested batches must be created explicitly, by passing the parent's batch_id:

Cosmo::Batch.new(parent: batch_id)

A nested batch counts as one job of its parent. If anything fails in it, the failure is propagated to the parent counter either. The parent succeeds when everything succeeds. Always call #jobs at least once, even with nothing in it - a batch that never gets a #jobs call never closes and never fires.

How this is actually stored in NATS

Every batch has an id (bid, a random hex string). State lives in two places, both keyed by that bid:

  1. Three atomic counters (via API::Counter, namespace "batch", backed by the shared _cosmostats stream):

    <bid>.total    - how many jobs have ever joined this batch
<bid>.pending  - how many of them are still running
<bid>.failed   - how many ended up dead-lettered or dropped

    Each increment/decrement is one atomic NATS publish, and the reply tells you the resulting value right away - no separate read needed. Whichever decrement of pending happens to bring it down to exactly 0 is the one that finalizes the batch (see #release_pending). Since it's a single atomic counter, exactly one decrement can ever be "the one that hits zero", so the batch can't finish twice or too early.

  2. A KV bucket (BUCKET, ttl'd so old batches clean themselves up):

    <bid>.meta               - { parent_id, created_at }, written once at creation
<bid>.callback.success   - { class, opts } from #on(:success, ...)
<bid>.callback.complete  - { class, opts } from #on(:complete, ...)
<bid>.ready              - { total, succeeded, failed }, written once
                          by #finalize, right after reading the counters
<bid>.fired.success      - exists once the :success callback has run
<bid>.fired.complete     - exists once the :complete callback has run
<bid>.finalized          - exists once #finalize has run for this bid

    The fired.* and finalized keys are write-once: kv.create raises if the key already exists. That's what stops the same callback (or the same finalize) from running twice when two things race to trigger it - a job finishing at the same moment someone calls #on, for example.

Defined Under Namespace

Modules: Dispatcher Classes: Callback

Constant Summary collapse

BUCKET =

Returns:

  • (::String) 
"cosmo_jobs_batches"
DEFAULT_EXPIRY =

Returns:

  • (::String) 
"3d"
EVENTS =

Returns:

  • (::Array[Symbol]) 
%i[success complete].freeze

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Methods included from Dispatcher

dispatch_callback, finalize, fireable?, parent_propagate, purge_counters

Constructor Details

#initialize(parent: nil) ⇒ Batch

Returns a new instance of Batch.

Parameters:

  • parent: (::String, nil) (defaults to: nil) 


105
106
107
108
109
110
 
# File 'lib/cosmo/batch.rb', line 105

def initialize(parent: nil)
  @bid = SecureRandom.hex(8)
  @parent_id = parent
  kv.set("#{@bid}.meta", Utils::Json.dump({ parent_id: @parent_id, created_at: Time.now.to_i }))
  link(@parent_id) if @parent_id
end

Instance Attribute Details

#bid ⇒ ::String (readonly)

Returns the value of attribute bid.

Returns:

  • (::String) 


103
104
105
 
# File 'lib/cosmo/batch.rb', line 103

def bid
  @bid
end

#parent_id ⇒ ::String? (readonly)

Returns the value of attribute parent_id.

Returns:

  • (::String, nil) 


103
104
105
 
# File 'lib/cosmo/batch.rb', line 103

def parent_id
  @parent_id
end

Class Method Details

.counter ⇒ API::Counter

Returns:



95
96
97
 
# File 'lib/cosmo/batch.rb', line 95

def self.counter
  @counter ||= API::Counter.new("batch")
end

.current ⇒ Batch?

Returns:



83
84
85
 
# File 'lib/cosmo/batch.rb', line 83

def self.current
  Thread.current[:cosmo_batch]
end

.kv ⇒ API::KV

Returns:



99
100
101
 
# File 'lib/cosmo/batch.rb', line 99

def self.kv
  @kv ||= API::KV.new(BUCKET, ttl: Utils::Duration.parse(Config[:batch_expiry] || DEFAULT_EXPIRY))
end

.notify(bid, jid, success:) ⇒ void

This method returns an undefined value.

Called by Job::Processor when a job is done for good - acked, or dead-lettered/dropped. Never call this for a retry. JID is used to dedupe: if the same job's ack gets lost, and it's redelivered, we don't want to count it twice.

Parameters:

  • bid (::String)
  • jid (::String)
  • success: (Boolean) 


90
91
92
93
 
# File 'lib/cosmo/batch.rb', line 90

def self.notify(bid, jid, success:)
  counter.increment("#{bid}.failed", msg_id: "batch.#{bid}.#{jid}.failed") unless success
  release_pending(bid, msg_id: "batch.#{bid}.#{jid}.pending")
end

Instance Method Details

#counter ⇒ API::Counter

Bridges instance methods to the private class methods Dispatcher adds to Batch

Returns:



158
 
# File 'lib/cosmo/batch.rb', line 158

def counter = self.class.counter

#jobs { ... } ⇒ void

This method returns an undefined value.

Every perform_async/perform_at/perform_in called inside this block joins the batch. Holds a placeholder "pending" slot for the whole block, so the batch can't finish while you're still adding jobs, even if the first one completes instantly. The placeholder never counts toward total, so it doesn't skew the stats. Safe to call more than once. If the block raises, we still release the placeholder (so the batch isn't stuck) and let the error propagate as normal.

Yields:

Yield Returns:

  • (void) 


119
120
121
122
123
124
125
126
127
128
129
 
# File 'lib/cosmo/batch.rb', line 119

def jobs
  previous = Thread.current[:cosmo_batch]
  Thread.current[:cosmo_batch] = self
  counter.increment("#{bid}.pending")
  begin
    yield
  ensure
    Thread.current[:cosmo_batch] = previous
    release_pending
  end
end

#kv ⇒ API::KV

Returns:



159
 
# File 'lib/cosmo/batch.rb', line 159

def kv = self.class.kv

#link(parent_id) ⇒ void

This method returns an undefined value.

Parameters:

  • parent_id (::String) 


160
 
# File 'lib/cosmo/batch.rb', line 160

def link(parent_id) = self.class.send(:link, parent_id)

#on(event, klass, **opts) ⇒ void

This method returns an undefined value.

Registers a callback for event (:success or :complete). Can be called before or after #jobs - if the batch already finished, it fires right away.

Parameters:

  • event (Symbol)
  • klass (Class)
  • (Object)

Raises:



134
135
136
137
138
139
 
# File 'lib/cosmo/batch.rb', line 134

def on(event, klass, **opts)
  raise ArgumentError, "event must be :success or :complete" unless EVENTS.include?(event)

  kv.set("#{bid}.callback.#{event}", Utils::Json.dump({ class: klass.name, opts: opts }))
  try_fire(event)
end

#register_job! ⇒ void

This method returns an undefined value.

Called by Job#perform for each job added inside #jobs.



142
143
144
145
 
# File 'lib/cosmo/batch.rb', line 142

def register_job!
  counter.increment("#{bid}.total")
  counter.increment("#{bid}.pending")
end

#release_pending ⇒ void

This method returns an undefined value.



161
 
# File 'lib/cosmo/batch.rb', line 161

def release_pending = self.class.send(:release_pending, bid)

#rollback_job! ⇒ void

This method returns an undefined value.

Undoes #register_job! when the publish itself failed, so the job never really joined the batch. Unlike a real completion, this also reverts total, not just pending.



150
151
152
153
 
# File 'lib/cosmo/batch.rb', line 150

def rollback_job!
  counter.decrement("#{bid}.total")
  release_pending
end

#try_fire(event) ⇒ void

This method returns an undefined value.

Parameters:

  • event (Symbol) 


162
 
# File 'lib/cosmo/batch.rb', line 162

def try_fire(event) = self.class.send(:try_fire, bid, event)