Class: Wurk::Batch::Status

Inherits:
Object
  • Object
show all
Defined in:
lib/wurk/batch/status.rb

Overview

Snapshot of a batch's state — read-only view backed by HGETALL of b-<bid> plus the supporting JIDs/failed/died sets. Mirrors the Sidekiq::Batch::Status surface from docs/target/sidekiq-pro.md §2.5.

#data returns the JSON-friendly hash served by the polling endpoint. #join blocks the current thread until complete? — test/util only. #delete removes the batch and every reference to it.

Constant Summary collapse

JOIN_POLL_INTERVAL = 
0.5

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(bid, data: nil) ⇒ Status

data: is the b-<bid> hash already read, by a listing that fetched a page of them in one pipeline (BatchSet#snapshot); without it the hash is read here.

Raises:

  • (ArgumentError) 


24
25
26
27
28
29
 
# File 'lib/wurk/batch/status.rb', line 24

def initialize(bid, data: nil)
  raise ArgumentError, 'bid required' if bid.nil? || bid.to_s.empty?

  @bid = bid.to_s
  data ? self.data_hash = data : reload!
end

Instance Attribute Details

#bid ⇒ Object (readonly)

Returns the value of attribute bid.



19
20
21
 
# File 'lib/wurk/batch/status.rb', line 19

def bid
  @bid
end

Class Method Details

.summary_commands(bid) ⇒ Object

The four set sizes #summary takes, as one pipeline's commands.



32
33
34
 
# File 'lib/wurk/batch/status.rb', line 32

def self.summary_commands(bid)
  %w[failed died kids jids].map { |set| ['SCARD', "b-#{bid}-#{set}"] }
end

Instance Method Details

#callback_queue ⇒ Object 



50
 
# File 'lib/wurk/batch/status.rb', line 50

def callback_queue   = @data['callback_queue']

#child_count ⇒ Object 



80
81
82
 
# File 'lib/wurk/batch/status.rb', line 80

def child_count
  Wurk.redis { |conn| conn.call('SCARD', "b-#{@bid}-kids") }.to_i
end

#complete? ⇒ Boolean

:complete fires when the live jids set is empty (every job has either succeeded or died). Hash field is set by the server callback; falling back to a recompute keeps Status accurate even if the callback dispatch is in flight.

Returns:

  • (Boolean) 


57
58
59
60
61
 
# File 'lib/wurk/batch/status.rb', line 57

def complete?
  return true if @data['complete'] == '1'

  complete_with?(live_jids_count)
end

#complete_at ⇒ Object 



45
 
# File 'lib/wurk/batch/status.rb', line 45

def complete_at      = numeric_or_nil(@data['complete_at'])

#created_at ⇒ Object 



44
 
# File 'lib/wurk/batch/status.rb', line 44

def created_at       = numeric_or_nil(@data['created_at'])

#data ⇒ Object

JSON-serializable snapshot used by the polling middleware / web UI. Field names are wire-compat with Sidekiq Pro's BatchStatus. The four set reads it needs ride one pipeline.



96
97
98
99
 
# File 'lib/wurk/batch/status.rb', line 96

def data
  failed, dead, kids, live = set_snapshot
  header(live).merge('failed_jids' => failed, 'dead_jids' => dead, 'child_count' => kids.to_i)
end

#dead_jids ⇒ Object 



76
77
78
 
# File 'lib/wurk/batch/status.rb', line 76

def dead_jids
  Wurk.redis { |conn| conn.call('SMEMBERS', "b-#{@bid}-died") }
end

#death_at ⇒ Object 



47
 
# File 'lib/wurk/batch/status.rb', line 47

def death_at         = numeric_or_nil(@data['death_at'])

#delete ⇒ Object

Nukes every key for this batch, its index entries and tag indexes, and its membership in the parent's -kids/-pkids — one atomic script (BATCH_DELETE). Dangerous if jobs are still in flight: they'll succeed/fail without a batch to ack against and this batch's callbacks won't fire. The parent is not left waiting on a child that no longer exists: when this was the last thing it waited on, its callbacks fire here.



131
132
133
134
135
136
137
138
139
140
141
 
# File 'lib/wurk/batch/status.rb', line 131

def delete
  reload!
  parent = parent_bid.to_s
  own    = Batch.keys_for(@bid)
  pending, live, kids = Wurk.redis do |conn|
    Wurk::Lua::Loader.eval_cached(conn, :batch_delete, keys: delete_keys(own, parent),
                                                       argv: [@bid, own.size, parent.empty? ? '0' : '1'])
  end.map(&:to_i)
  Callbacks.maybe_fire(parent, pending: pending, live: live, kids: kids) unless parent.empty?
  nil
end

#description ⇒ Object 



48
 
# File 'lib/wurk/batch/status.rb', line 48

def description      = @data['description']

#exists? ⇒ Boolean

False when no b-<bid> hash exists — a well-formed bid that was never created (or has expired). Lets callers 404 instead of serving an all-zero phantom batch.

Returns:

  • (Boolean) 


39
 
# File 'lib/wurk/batch/status.rb', line 39

def exists? = !@data.empty?

#failed_jids ⇒ Object 



63
64
65
 
# File 'lib/wurk/batch/status.rb', line 63

def failed_jids
  Wurk.redis { |conn| conn.call('SMEMBERS', "b-#{@bid}-failed") }
end

#failure_info ⇒ Object

Deprecated pre-Pro8 surface (spec §2.5): an array of per-failure error detail. The Pro8 data model (§2.8) drops the b-<bid>-failinfo hash in favour of the failed_jids set, which Wurk tracks — so the per-jid error payload is intentionally not persisted and this returns []. Kept so drop-in callers referencing #failure_info don't NameError.



72
73
74
 
# File 'lib/wurk/batch/status.rb', line 72

def failure_info
  []
end

#failures ⇒ Object 



43
 
# File 'lib/wurk/batch/status.rb', line 43

def failures         = @data['failures'].to_i

#invalidated? ⇒ Boolean

Returns:

  • (Boolean) 


51
 
# File 'lib/wurk/batch/status.rb', line 51

def invalidated?     = @data['invalidated'] == '1'

#join ⇒ Object

Blocks the current thread until complete? is true. Test/util only — polling Redis from a worker thread would defeat the whole point of asynchronous batches.



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

def join
  loop do
    reload!
    break if complete?

    sleep JOIN_POLL_INTERVAL
  end
end

#parent_bid ⇒ Object 



49
 
# File 'lib/wurk/batch/status.rb', line 49

def parent_bid       = @data['parent_bid']

#pending ⇒ Object 



42
 
# File 'lib/wurk/batch/status.rb', line 42

def pending          = @data['pending'].to_i

#reload! ⇒ Object 



143
144
145
146
 
# File 'lib/wurk/batch/status.rb', line 143

def reload!
  self.data_hash = Wurk.redis { |conn| conn.call('HGETALL', "b-#{@bid}") }
  self
end

#success_at ⇒ Object 



46
 
# File 'lib/wurk/batch/status.rb', line 46

def success_at       = numeric_or_nil(@data['success_at'])

#summary(sizes = nil) ⇒ Object

#data for a listing row: the failed and dead jid sets by size rather than by member, so a page of batches with ten thousand failures each is 25 rows and not 250,000 jids. sizes are the replies to .summary_commands when the caller pipelined them for a whole page.



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

def summary(sizes = nil)
  failed, dead, kids, live = sizes || Wurk.redis do |conn|
    conn.pipelined { |pipe| Status.summary_commands(@bid).each { |cmd| pipe.call(*cmd) } }
  end
  header(live).merge('failed_count' => failed.to_i, 'dead_count' => dead.to_i, 'child_count' => kids.to_i)
end

#tags ⇒ Object 



84
85
86
87
88
89
90
91
 
# File 'lib/wurk/batch/status.rb', line 84

def tags
  raw = @data['tags']
  return [] if raw.nil? || raw.empty?

  JSON.parse(raw)
rescue JSON::ParserError
  []
end

#total ⇒ Object 



41
 
# File 'lib/wurk/batch/status.rb', line 41

def total            = @data['total'].to_i