Class: Wurk::JobSet

Inherits:
SortedSet show all
Defined in:
lib/wurk/job_set.rb

Overview

ZSET-of-jobs view. Reverse-paged iteration so callers see newest-first (highest score, i.e. furthest-out retry/schedule). Mutations use ZREM-by-value when the exact bytes are known and a (score, jid) scan otherwise.

Spec: docs/target/sidekiq-free.md §19.5.

Direct Known Subclasses

DeadSet, RetrySet, ScheduledSet

Constant Summary collapse

FIND_SCAN_COUNT =

rubocop:disable Metrics/ClassLength

1_000

Instance Method Summary collapse

Constructor Details

This class inherits a constructor from Wurk::SortedSet

Instance Method Details

#delete_by_jid(score, jid) ⇒ Object Also known as: delete

Scan the score bracket for a jid match, ZREM the exact bytes once found. Returns true on success. Aliased as delete for Sidekiq wire-compat. Per-row JSON rescue so a single malformed entry can't shadow a valid match at the same score.



229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
 
# File 'lib/wurk/job_set.rb', line 229

def delete_by_jid(score, jid)
  Wurk.redis do |conn|
    rows = conn.call('ZRANGEBYSCORE', @name, score.to_f, score.to_f)
    rows.each do |raw|
      parsed = begin
        Wurk.load_json(raw)
      rescue ::JSON::ParserError
        nil
      end
      next unless parsed && parsed['jid'] == jid

      gone = conn.call('ZREM', @name, raw).to_i.positive?
      note_removal if gone
      return gone
    end
  end
  false
end

#delete_by_value(name, value) ⇒ Object

ZREM by exact bytes. Returns true when ≥1 element was removed. Method name is Sidekiq wire-compat — delete_by_value? would break the alias.



221
222
223
 
# File 'lib/wurk/job_set.rb', line 221

def delete_by_value(name, value)
  zrem_member(name, value)
end

#each ⇒ Object

Newest-first paged ZRANGE. Yields a SortedEntry per row.

Removing a yielded entry shifts every later rank up by one, so a plain page * PAGE_SIZE offset would skip a row per removal — each(&:delete) over 120 entries used to leave 50 behind. Like upstream, the offset is pulled back by the removals made through this set (#removed) since the iteration started.



85
86
87
88
89
90
91
92
93
94
95
96
97
 
# File 'lib/wurk/job_set.rb', line 85

def each
  return enum_for(:each) unless block_given?

  baseline = removed
  page = 0
  loop do
    slice = newest_page(page, baseline)
    slice.each { |value, score| yield SortedEntry.new(self, score, value) }
    return (page * PAGE_SIZE) + slice.size if slice.size < PAGE_SIZE

    page += 1
  end
end

#fetch(score, jid = nil) ⇒ Object

O(score) lookup. score accepts Time, Numeric, or a Range of either. Returns the matching entries (possibly multiple at the same exact score). When jid is set, narrows to the single (score, jid) pair.



172
173
174
175
176
177
178
 
# File 'lib/wurk/job_set.rb', line 172

def fetch(score, jid = nil)
  results = Wurk.redis { |conn| conn.call('ZRANGEBYSCORE', @name, *range_args(score), 'WITHSCORES') }
  entries = results.map { |value, sc| SortedEntry.new(self, sc, value) }
  return entries unless jid

  entries.select { |e| e.jid == jid }
end

#fetch_pairs(pairs) ⇒ Array<SortedEntry>

#fetch(score, jid) for every [score, jid] pair, in one pipeline where #fetch pays a round trip per pair. Internal reader for the dashboard's bulk actions (up to 1000 keys a request).

Returns:



185
186
187
188
189
190
191
192
193
194
195
196
 
# File 'lib/wurk/job_set.rb', line 185

def fetch_pairs(pairs)
  return [] if pairs.empty?

  replies = Wurk.redis do |conn|
    conn.pipelined do |pipe|
      pairs.each { |pair| pipe.call('ZRANGEBYSCORE', @name, *range_args(pair[0]), 'WITHSCORES') }
    end
  end
  pairs.zip(replies).flat_map do |(_, jid), rows|
    rows.map { |value, score| SortedEntry.new(self, score, value) }.select { |entry| entry.jid == jid }
  end
end

#find_job(jid) ⇒ Object

ZSCAN-based search by jid substring. Returns the first matching entry or nil. O(n) on the ZSET — callers iterating many jids should switch to per-jid hashes or the score-based fetch. Walks FIND_SCAN_COUNT per call rather than #scan's upstream default of 100: a jid matches one member, so the walk is the whole cost, and 1000 cuts a 1M-entry set from ~10k round trips to ~1k.



204
205
206
207
 
# File 'lib/wurk/job_set.rb', line 204

def find_job(jid)
  scan(jid, FIND_SCAN_COUNT) { |entry| return entry if entry.jid == jid }
  nil
end

#kill_all(notify_failure: false, ex: nil) ⇒ Object

Moves every job in this set to the dead set. Death handlers stay quiet unless notify_failure: true — Sidekiq 8's default (spec §19.5), so the dashboard's "Kill All" doesn't page an error tracker once per job. The dead set is trimmed once at the end rather than per entry. Returns the count of jobs moved; a job that could not be moved stays in this set (see #sweep_once).



157
158
159
160
161
162
163
164
165
166
167
 
# File 'lib/wurk/job_set.rb', line 157

def kill_all(notify_failure: false, ex: nil)
  dead = DeadSet.new
  opts = { notify_failure: notify_failure, ex: ex, trim: false }
  begin
    sweep_once do |entry|
      entry.send(:remove_job) { |message| dead.kill(message, opts) }
    end
  ensure
    dead.trim
  end
end

#pop_each ⇒ Object

ZPOPMIN loop. Each iteration pops the single oldest member (lowest score, e.g. earliest scheduled-at) and yields the raw JSON + score. Stops when the set is empty.



118
119
120
121
122
123
124
125
126
127
128
 
# File 'lib/wurk/job_set.rb', line 118

def pop_each
  loop do
    result = Wurk.redis { |conn| conn.call('ZPOPMIN', @name, 1) }
    break if result.nil? || result.empty?

    # Newer redis-client returns nested `[[value, score]]` even with COUNT 1;
    # older `[value, score]`. Normalize both.
    value, score = result.first.is_a?(Array) ? result.first : result
    yield value, score.to_f
  end
end

#remove_job(entry) ⇒ Object

Removes the exact (score, jid)-matching member. Backs SortedEntry#delete when no cached value bytes are present.

Goes around the public #delete_by_value: sidekiq-unique-jobs prepends it and answers whether a lock was released, so a member already removed read as "not removed" and the caller dropped the job without pushing it.



215
216
217
 
# File 'lib/wurk/job_set.rb', line 215

def remove_job(entry)
  zrem_member(@name, entry.value) || delete_by_jid(entry.score, entry.jid)
end

#retry_all ⇒ Object

Re-enqueues every job in this set via the client. Returns how many were re-enqueued; a job the push rejects stays in the set (see #sweep_once).

Same result as SortedEntry#retry per entry, which is upstream's loop, at a page's cost rather than an entry's: each page of #each is removed in one pipeline and pushed in one write per queue (#retry_page). A gem that hooks the per-entry path (sidekiq-unique-jobs prepends #delete_by_value to release its lock) gets that path, since a batch would skip its hook.



138
139
140
141
142
143
144
145
146
147
148
149
 
# File 'lib/wurk/job_set.rb', line 138

def retry_all
  return sweep_once(&:retry) if per_entry_hooked?

  batch = []
  done = sweep_once(batch: true) do |entry|
    batch << entry
    next 0 if batch.size < PAGE_SIZE

    retry_page(batch.slice!(0..))
  end
  done + (batch.empty? ? 0 : retry_page(batch))
end

#schedule(timestamp, message) ⇒ Object

Plain ZADD, no NX: re-scheduling an identical payload moves it to the new timestamp rather than keeping the first one. Mirrors Sidekiq::JobSet#schedule exactly — the last schedule call wins.



74
75
76
 
# File 'lib/wurk/job_set.rb', line 74

def schedule(timestamp, message)
  Wurk.redis { |conn| conn.call('ZADD', @name, timestamp.to_f.to_s, Wurk.dump_json(message)) }
end

#snapshot(offset, count) ⇒ Array(Integer, Array<SortedEntry>)

The set's size and the count entries from offset in #each's newest-first order, in one round trip rather than #each's walk from the top. Internal reader for the dashboard and the HTTP API.

Returns:



104
105
106
107
108
109
110
111
112
113
 
# File 'lib/wurk/job_set.rb', line 104

def snapshot(offset, count)
  size, rows = Wurk.redis do |conn|
    conn.pipelined do |pipe|
      pipe.call('ZCARD', @name)
      pipe.call('ZRANGE', @name, offset, offset + count - 1, 'REV', 'WITHSCORES')
    end
  end
  rows = rows.each_slice(2).to_a unless rows.first.is_a?(Array)
  [size, rows.map { |value, score| SortedEntry.new(self, score, value) }]
end