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) and
 pushes them at block exit, each registered via BATCH_PUSH. A hold
 keeps the batch from draining while the block is open.
4. Workers ack on success → BATCH_ACK_SUCCESS → pending--.
 Death handler acks on permanent failure → BATCH_ACK_COMPLETE.
5. When live jids and pending child batches both hit 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, CallbackEnqueueError, 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
EMPTY_JOB = 
'Sidekiq::Batch::Empty'
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
HOLDS_KEY =

Bids whose #jobs hold this thread already owns, so a block nested in another block of the same batch neither takes a second hold nor drops the outer one when it exits.

:wurk_batch_holds
HOLD_PREFIX =

Prefix of the hold sentinel's member in b-<bid>-jids. Real jids are hex, so a sentinel can never collide with one.

'hold:'

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(bid = nil) ⇒ Batch

Returns a new instance of Batch.



171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
 
# File 'lib/wurk/batch.rb', line 171

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.



105
106
107
 
# File 'lib/wurk/batch.rb', line 105

def autoflush
  @autoflush
end

#bid ⇒ Object (readonly)

Returns the value of attribute bid.



104
105
106
 
# File 'lib/wurk/batch.rb', line 104

def bid
  @bid
end

#callback_class ⇒ Object

Returns the value of attribute callback_class.



104
105
106
 
# File 'lib/wurk/batch.rb', line 104

def callback_class
  @callback_class
end

#callback_queue ⇒ Object

Returns the value of attribute callback_queue.



105
106
107
 
# File 'lib/wurk/batch.rb', line 105

def callback_queue
  @callback_queue
end

#description ⇒ Object

Returns the value of attribute description.



105
106
107
 
# File 'lib/wurk/batch.rb', line 105

def description
  @description
end

#linger ⇒ Object

Returns the value of attribute linger.



104
105
106
 
# File 'lib/wurk/batch.rb', line 104

def linger
  @linger
end

#parent_bid ⇒ Object (readonly)

Returns the value of attribute parent_bid.



104
105
106
 
# File 'lib/wurk/batch.rb', line 104

def parent_bid
  @parent_bid
end

Class Method Details

.ack_success(conn, bid, jid) ⇒ Object

BATCH_ACK_SUCCESS for jid, as Integers: [removed, pending, live, kids]. The last three are the fire-gate inputs Callbacks.maybe_fire takes.



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

def self.ack_success(conn, bid, jid)
  Wurk::Lua::Loader.eval_cached(
    conn, :batch_ack_success,
    keys: ["b-#{bid}", "b-#{bid}-jids", "b-#{bid}-failed", "b-#{bid}-pkids"], argv: [jid]
  ).map(&:to_i)
end

.keys_for(bid) ⇒ Object 



107
108
109
110
 
# File 'lib/wurk/batch.rb', line 107

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.



166
167
168
169
 
# File 'lib/wurk/batch.rb', line 166

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.



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

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

.with_thread_batch(batch, buffer) ⇒ Object

Runs the block with batch as the thread's active batch (the client middleware stamps its bid) and buffer collecting batched pushes — either may be nil — restoring whatever an enclosing block had set.



115
116
117
118
119
120
121
122
123
124
 
# File 'lib/wurk/batch.rb', line 115

def self.with_thread_batch(batch, buffer)
  previous    = Thread.current[THREAD_KEY]
  prev_buffer = Thread.current[BUFFER_KEY]
  Thread.current[THREAD_KEY] = batch
  Thread.current[BUFFER_KEY] = buffer
  yield
ensure
  Thread.current[THREAD_KEY] = previous
  Thread.current[BUFFER_KEY] = prev_buffer
end

Instance Method Details

#expires_in(duration) ⇒ Object 



272
273
274
275
 
# File 'lib/wurk/batch.rb', line 272

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

#include?(jid) ⇒ Boolean

Returns:

  • (Boolean) 


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

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 and acks them as successes (spec §12), so the batch still drains and fires. Cascades to descendant batches via b--kids.



259
260
261
262
 
# File 'lib/wurk/batch.rb', line 259

def invalidate_all
  cascade_invalidate(@bid)
  nil
end

#jobs(&block) ⇒ Object

Atomic enqueue block (spec §2.3). Inside the block, Job.perform_async finds this batch via Thread.current and stamps bid onto the payload; the pushes are collected and flushed at block exit, each through BATCH_PUSH. A block that raises pushes nothing it collected. autoflush = N flushes every N jobs instead, trading that atomicity for bounded memory. A block that creates the batch and adds nothing to it — no job, no child batch — synthesises a Batch::Empty no-op (spec §2.3); a block re-entering an existing batch needs none, the batch already has members (or drains when the hold is released).

For the whole block the batch carries a hold (BATCH_HOLD), so a job pushed early — a flushed autoflush slice, a scheduled job, a nested child batch — cannot ack the batch empty and fire its callbacks before the rest is in. The hold is released at block exit through the same ack path as a job. When the block that created the batch raises, the hold stays: nothing the block collected was pushed, and a batch that fired for whatever slipped out early would be the partial batch §2.3 rules out. A block re-entering an existing batch releases it either way, or the batch's own jobs could never fire it.

Raises:

  • (ArgumentError) 


315
316
317
318
319
320
321
322
323
324
325
326
327
 
# File 'lib/wurk/batch.rb', line 315

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

  threshold = autoflush_threshold
  created   = !@flushed_once
  ensure_first_flush!
  with_hold(release_on_error: !created) do
    collect_jobs(threshold, &block)
    enqueue_empty_marker if created && untouched?
  end
  @mutable = false
  self
end

#mutable? ⇒ Boolean

Returns:

  • (Boolean) 


230
231
232
 
# File 'lib/wurk/batch.rb', line 230

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) 


283
284
285
286
287
288
289
290
291
292
293
294
 
# File 'lib/wurk/batch.rb', line 283

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 



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

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), in one atomic script. Removing the last live jid drains the batch like an ack would, so the callbacks fire here rather than never.



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

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

  removed, pending, live, kids = Wurk.redis do |conn|
    Wurk::Lua::Loader.eval_cached(
      conn, :batch_remove_jobs,
      keys: ["b-#{@bid}", "b-#{@bid}-jids", "b-#{@bid}-failed", "b-#{@bid}-pkids"], argv: jids
    )
  end.map(&:to_i)
  Callbacks.maybe_fire(@bid, pending: pending, live: live, kids: kids) if removed.positive?
  removed
end

#status ⇒ Object 



268
269
270
 
# File 'lib/wurk/batch.rb', line 268

def status
  Status.new(@bid)
end

#tags ⇒ Object 



199
200
201
 
# File 'lib/wurk/batch.rb', line 199

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.



195
196
197
 
# File 'lib/wurk/batch.rb', line 195

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

#valid? ⇒ Boolean

Returns:

  • (Boolean) 


264
265
266
 
# File 'lib/wurk/batch.rb', line 264

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