Class: Wurk::Batch

Inherits:
Object
  • Object
show all
Defined in:
lib/wurk/batch.rb,
lib/wurk/batch_set.rb,
lib/wurk/batch/empty.rb,
lib/wurk/batch/buffer.rb,
lib/wurk/batch/status.rb,
lib/wurk/batch/callbacks.rb,
lib/wurk/batch/callback_job.rb,
lib/wurk/batch/death_handler.rb,
lib/wurk/batch/client_middleware.rb,
lib/wurk/batch/server_middleware.rb

Overview

Sidekiq Pro Batches. Group jobs, attach success/complete/death callbacks, track progress. Spec: docs/target/sidekiq-pro.md §2.

Lifecycle:

1. `Batch.new` allocates a fresh BID; the batch is `mutable?` until the
 first `#jobs` block flushes — that flush HSETs the core hash, ZADDs
 to `batches`, and writes tag indexes.
2. `Batch.new(bid)` reopens an existing batch (legal from inside a job
 or callback only). `mutable?` is false because reopening implies the
 first flush already happened.
3. `#jobs { ... }` collects `Job.perform_async` calls via the client
 middleware (Thread.current[:wurk_current_batch] is the signal),
 atomically registering each via BATCH_PUSH.
4. Workers ack on success → BATCH_ACK_SUCCESS → pending--.
 Death handler acks on permanent failure → BATCH_ACK_COMPLETE.
5. When live jids hits zero → fire `:complete`. When pending also
 hits zero with zero deaths → fire `:success`.

Nested batches: a job opening its OWN batch (batch.jobs { ... }) increments live counters on the existing batch. A callback opening its PARENT batch links via parent_bid and adds child BID to b--kids.

Examples:

Define a batch with a success callback

batch = Sidekiq::Batch.new
batch.description = "Nightly import"
batch.on(:success, ImportCallback, "user_id" => user.id)
batch.jobs do
  rows.each { |r| ImportRowJob.perform_async(r.id) }
end
batch.bid   # => the batch id, persisted in Redis

Defined Under Namespace

Modules: Callbacks Classes: Buffer, CallbackJob, ClientMiddleware, DeadSet, DeathHandler, Empty, ServerMiddleware, Status

Constant Summary collapse

DEFAULT_EXPIRY_SECONDS =

rubocop:disable Metrics/ClassLength

30 * 24 * 60 * 60
POST_SUCCESS_EXPIRY_SECONDS = 
24 * 60 * 60
CALLBACK_NOTIFY_TTL = 
30 * 24 * 60 * 60
INDEX_MAX =

Member ceiling for the two batch index ZSETs (batches, dead-batches). The score axis in .trim_index retires entries in step with the batch data itself, so this is only the backstop for a workload creating batches faster than that window retires them. Deliberately generous: a cap that bites drops batches whose data is still live out of BatchSet, and at this scale the per-batch hashes dwarf the index anyway.

1_000_000
CALLBACKS_MAX =

Ceiling on the callbacks array of one batch hash, enforced by BATCH_APPEND_CALLBACK. Every registration re-encodes the whole array, and every entry becomes a callback job when the event fires, so an unbounded array is both a hot-path cost and a fan-out. Far above any legitimate batch — real ones register a handful — so hitting it means a loop is registering callbacks it should have registered once.

1_000
BID_BYTES =

Bid is URL-safe base64 of 10 random bytes — matches Sidekiq Pro's BID generator. Length matters: third-party gems that key off bid prefix (sharded batches in Pro 8) inspect the first character.

10
VALID_EVENTS = 
%i[success complete death].freeze
KEY_SUFFIXES =

Every key a batch owns, for the sweep paths: Status#delete (UNLINK), Callbacks#apply_linger and DeathHandler.restamp_ttls (EXPIRE).

The 'live' set tracks jobs that have not yet reached a terminal state. When it's empty, every job has either succeeded or died → :complete is allowed to fire.

complete/success/death are the callback dedup markers written by Callbacks#dedup_set; they belong to the batch and must die with it. notify/cbsucc/tags are Sidekiq Pro's own key layout (spec §2.8) that Wurk never writes — Wurk dedups on the three markers above and indexes tags at tags:<tag>. They stay listed so a Redis dataset carried over from Sidekiq Pro on the gem swap gets swept too; EXPIRE/UNLINK of a missing key is a no-op for batches Wurk created itself.

%w[jids failed died complete success death notify cbsucc kids pkids tags].freeze
THREAD_KEY = 
:wurk_current_batch
BUFFER_KEY =

Set on the current thread (to a Buffer) only inside an autoflush #jobs block. Client#raw_push reads it: when present, batched pushes accumulate here instead of round-tripping per job.

:wurk_batch_buffer

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(bid = nil) ⇒ Batch

Returns a new instance of Batch.



134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
 
# File 'lib/wurk/batch.rb', line 134

def initialize(bid = nil)
  @bid              = bid || SecureRandom.urlsafe_base64(BID_BYTES)
  @existing         = !bid.nil?
  @description      = nil
  @callback_queue   = 'default'
  @callback_class   = nil
  @tags             = []
  @autoflush        = nil
  @linger           = nil
  @parent_bid       = nil
  @callbacks        = []
  # Dedup index over `@callbacks`, keyed on the encoded entry. Only the
  # pre-flush staging path feeds it — once flushed, Redis holds the array
  # and BATCH_APPEND_CALLBACK does the deduping — so a batch reopened by
  # bid never pays to build it.
  @callback_index   = Set.new
  @expires_in       = DEFAULT_EXPIRY_SECONDS
  @mutable          = !@existing
  @flushed_once     = @existing
  load_existing! if @existing
end

Instance Attribute Details

#autoflush ⇒ Object

Returns the value of attribute autoflush.



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

def autoflush
  @autoflush
end

#bid ⇒ Object (readonly)

Returns the value of attribute bid.



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

def bid
  @bid
end

#callback_class ⇒ Object

Returns the value of attribute callback_class.



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

def callback_class
  @callback_class
end

#callback_queue ⇒ Object

Returns the value of attribute callback_queue.



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

def callback_queue
  @callback_queue
end

#description ⇒ Object

Returns the value of attribute description.



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

def description
  @description
end

#linger ⇒ Object

Returns the value of attribute linger.



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

def linger
  @linger
end

#parent_bid ⇒ Object (readonly)

Returns the value of attribute parent_bid.



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

def parent_bid
  @parent_bid
end

Class Method Details

.keys_for(bid) ⇒ Object 



93
94
95
96
 
# File 'lib/wurk/batch.rb', line 93

def self.keys_for(bid)
  base = "b-#{bid}"
  [base, *KEY_SUFFIXES.map { |s| "#{base}-#{s}" }]
end

.trim_bounds(max: nil, timeout: nil) ⇒ Object

The same two bounds as arguments, for a writer that cannot append to a pipeline: Wurk::Flow's creation script has to trim from inside its own atomic write, and one policy read out of here beats a second copy of the cutoff arithmetic that drifts the day this one changes.



129
130
131
132
 
# File 'lib/wurk/batch.rb', line 129

def self.trim_bounds(max: nil, timeout: nil)
  cutoff = ::Process.clock_gettime(::Process::CLOCK_REALTIME) - (timeout || DEFAULT_EXPIRY_SECONDS)
  ["(#{cutoff}", -(max || INDEX_MAX)]
end

.trim_index(pipe, key, max: nil, timeout: nil) ⇒ Object

Two-axis trim of a batch index ZSET (batches, dead-batches), in the shape of the morgue trim (DeadSet#trim): ZREMRANGEBYSCORE evicts entries older than timeout, ZREMRANGEBYRANK 0 -max caps the member count — and, like the morgue, that bound keeps max - 1 of a full set. Appended to the caller's pipeline so bounding the index costs neither writer an extra round trip.

Nothing else ever shrinks either set: Status#delete and the death-recovery ZREM are manual, so an index entry outlives the batch it points at and both sets grow for the life of the Redis without this.

Both index in epoch seconds — batches from CLOCK_REALTIME, dead-batches from Time.now.to_f — so one cutoff serves both. The default window is the batch hash TTL: past it b-<bid> is gone and the entry only yields an empty Status. A batch that overrode expires_in beyond that window outlives its index entry — still reachable by bid, just no longer enumerated by BatchSet.

max: / timeout: override the defaults for one call, so parallel tests can drive the trim on isolated limits without mutating the process-global Wurk.configuration.



119
120
121
122
123
 
# File 'lib/wurk/batch.rb', line 119

def self.trim_index(pipe, key, max: nil, timeout: nil)
  score, rank = trim_bounds(max: max, timeout: timeout)
  pipe.call('ZREMRANGEBYSCORE', key, '-inf', score)
  pipe.call('ZREMRANGEBYRANK', key, 0, rank)
end

Instance Method Details

#expires_in(duration) ⇒ Object 



232
233
234
235
 
# File 'lib/wurk/batch.rb', line 232

def expires_in(duration)
  @expires_in = duration.to_i
  self
end

#include?(jid) ⇒ Boolean

Returns:

  • (Boolean) 


197
198
199
 
# File 'lib/wurk/batch.rb', line 197

def include?(jid)
  Wurk.redis { |conn| conn.call('SISMEMBER', "b-#{@bid}-jids", jid) }.to_i.positive?
end

#invalidate_all ⇒ Object

Mark batch invalid. Pending jobs still exist in their queues; the server middleware short-circuits them when it observes the flag. Cascades to descendant batches via b--kids.



219
220
221
222
 
# File 'lib/wurk/batch.rb', line 219

def invalidate_all
  cascade_invalidate(@bid)
  nil
end

#jobs(&block) ⇒ Object

Atomic enqueue block. Inside the block, Job.perform_async finds this batch via Thread.current and stamps bid onto the payload — the client middleware then uses BATCH_PUSH to register and push atomically. Empty blocks synthesise a Batch::Empty no-op so callbacks still fire.

Raises:

  • (ArgumentError) 


261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
 
# File 'lib/wurk/batch.rb', line 261

def jobs(&block)
  raise ArgumentError, 'jobs requires a block' unless block

  ensure_first_flush!
  pre_count = job_count
  collect_jobs(&block)
  # By the time we check, the buffer (if any) has flushed, so `total`
  # reflects everything the block pushed — a flat count is reliable.
  # Scheduled (`perform_in`) jobs count here too: BATCH_SCHEDULE moves
  # `total` at creation, so a scheduled-only block does not misfire the
  # marker as if it were empty.
  enqueue_empty_marker if job_count == pre_count
  @mutable = false
  self
end

#mutable? ⇒ Boolean

Returns:

  • (Boolean) 


193
194
195
 
# File 'lib/wurk/batch.rb', line 193

def mutable?
  @mutable
end

#on(event, callback, options = {}) ⇒ Object

Register a callback. Any number of distinct callbacks may be attached to one event; re-registering an identical [event, target, options] triple is a no-op, and past CALLBACKS_MAX entries the registration is dropped with a warning. The callback target may be a Class, "Foo#bar" string spec, or anything responding to name. options must be JSON-serializable.

Raises:

  • (ArgumentError) 


243
244
245
246
247
248
249
250
251
252
253
254
 
# File 'lib/wurk/batch.rb', line 243

def on(event, callback, options = {})
  sym = event.to_sym
  raise ArgumentError, "invalid event #{event.inspect}" unless VALID_EVENTS.include?(sym)
  raise ArgumentError, 'callback options must be a Hash' unless options.is_a?(Hash)

  entry = [sym.to_s, callback_target(callback), options]
  # Before the first flush the array lives only in memory; after it, Redis
  # is authoritative and `@callbacks` is a stale mirror nothing reads —
  # appending to it there would just leak one entry per registration.
  @flushed_once ? persist_callback!(entry) : stage_callback(entry)
  self
end

#parent ⇒ Object 



187
188
189
190
191
 
# File 'lib/wurk/batch.rb', line 187

def parent
  return nil if @parent_bid.nil? || @parent_bid.empty?

  Batch.new(@parent_bid)
end

#remove_jobs(*jids) ⇒ Object

Remove jobs from the batch. Decrements pending/total by exactly the count of jids actually removed (idempotent for repeated calls).



203
204
205
206
207
208
209
210
211
212
213
214
 
# File 'lib/wurk/batch.rb', line 203

def remove_jobs(*jids)
  return 0 if jids.empty?

  Wurk.redis do |conn|
    removed = conn.call('SREM', "b-#{@bid}-jids", *jids).to_i
    if removed.positive?
      conn.call('HINCRBY', "b-#{@bid}", 'pending', -removed)
      conn.call('HINCRBY', "b-#{@bid}", 'total', -removed)
    end
    removed
  end
end

#status ⇒ Object 



228
229
230
 
# File 'lib/wurk/batch.rb', line 228

def status
  Status.new(@bid)
end

#tags ⇒ Object 



162
163
164
 
# File 'lib/wurk/batch.rb', line 162

def tags
  @tags.dup
end

#tags=(value) ⇒ Object

Hash assignment writes strings — Sidekiq's UI / third-party gems expect String tags. Array-coercion lets callers pass a String or Set.



158
159
160
 
# File 'lib/wurk/batch.rb', line 158

def tags=(value)
  @tags = Array(value).map(&:to_s)
end

#valid? ⇒ Boolean

Returns:

  • (Boolean) 


224
225
226
 
# File 'lib/wurk/batch.rb', line 224

def valid?
  Wurk.redis { |conn| conn.call('HGET', "b-#{@bid}", 'invalidated') } != '1'
end