Class: Wurk::SortedEntry

Inherits:
JobRecord show all
Defined in:
lib/wurk/sorted_entry.rb

Overview

One entry inside a sorted-set view (Retry/Scheduled/Dead). Carries the member's score alongside the JobRecord so callers can re-target the exact (score, value) pair when mutating Redis — sorted-set membership is by value, but ZREM-by-value is faster than ZRANGEBYSCORE+filter.

The id field ("|") is the Sidekiq wire-compat identifier used by dashboards and third-party tooling. Don't reformat it.

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

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(parent, score, item) ⇒ SortedEntry

Returns a new instance of SortedEntry.

Parameters:

  • parent (JobSet, nil) —

    the owning set; nil when constructed bare.

  • score (Numeric) —

    ZSET score (Float seconds since epoch).

  • item (String, Hash) —

    raw JSON or parsed payload.



21
22
23
24
25
26
27
28
29
30
 
# File 'lib/wurk/sorted_entry.rb', line 21

def initialize(parent, score, item)
  super(item)
  @score = score.to_f
  @parent = parent
  # Parsed now, unlike JobRecord: sidekiq-failures reopens this method and
  # reads @item straight after it. Measured on a 20k-member walk that reads
  # no field: 2.5us per entry at 250B, 5us at 2KB, 10us at 8KB, and nothing
  # at all for the walks that do read one, which is all of them in practice.
  self.item if @value
end

Instance Attribute Details

#parent ⇒ Object (readonly)

Returns the value of attribute parent.



16
17
18
 
# File 'lib/wurk/sorted_entry.rb', line 16

def parent
  @parent
end

#score ⇒ Object (readonly)

Returns the value of attribute score.



16
17
18
 
# File 'lib/wurk/sorted_entry.rb', line 16

def score
  @score
end

Instance Method Details

#add_to_queue ⇒ Object

Removes this entry and re-enqueues it via the client with the payload untouched. Backs the scheduled/dead "add to queue" actions — Sidekiq's add_to_queue does not touch retry_count.



64
65
66
67
68
 
# File 'lib/wurk/sorted_entry.rb', line 64

def add_to_queue
  remove_job do |message|
    Client.new.push(Wurk.load_json(message))
  end
end

#at ⇒ Object 



39
 
# File 'lib/wurk/sorted_entry.rb', line 39

def at = ::Time.at(score).utc

#delete ⇒ Object

Removes this entry from the parent set. Prefers exact-value match (idempotent across duplicates with the same jid), falls back to (score, jid) when constructed without a cached value.



44
45
46
47
48
49
50
 
# File 'lib/wurk/sorted_entry.rb', line 44

def delete
  if @value
    @parent.delete_by_value(@parent.name, @value)
  else
    @parent.delete_by_jid(@score, jid)
  end
end

#error? ⇒ Boolean

Returns:

  • (Boolean) 


92
 
# File 'lib/wurk/sorted_entry.rb', line 92

def error? = !item['error_class'].nil?

#id ⇒ Object 



37
 
# File 'lib/wurk/sorted_entry.rb', line 37

def id = "#{score}|#{jid}"

#kill ⇒ Object

Removes this entry from its parent set and writes it to the dead set. Death handlers fire with the synthesized "Job killed by API" exception (Sidekiq's default) so error trackers and the batch death path observe API/UI kills.



86
87
88
89
90
 
# File 'lib/wurk/sorted_entry.rb', line 86

def kill
  remove_job do |message|
    DeadSet.new.kill(message)
  end
end

#queue ⇒ Object

JobRecord resolves the queue eagerly and only from a pre-parsed Hash; a sorted-set entry is almost always built from the raw JSON string, so read it off the payload instead.



35
 
# File 'lib/wurk/sorted_entry.rb', line 35

def queue = @queue ||= item['queue']

#reschedule(at) ⇒ Object

Shifts the score by the delta to at, returning the new score like Sidekiq's ZINCRBY does. ZADD XX INCR is that ZINCRBY minus its one surprise: on a member already promoted or deleted it returns nil instead of re-creating it, so a stale dashboard row can't resurrect a job that has since run.



57
58
59
 
# File 'lib/wurk/sorted_entry.rb', line 57

def reschedule(at)
  Wurk.redis { |conn| conn.call('ZADD', @parent.name, 'XX', 'INCR', at.to_f - @score, value) }
end

#restore_after_failed_push ⇒ Object

Puts the original bytes back at the original score, after this entry was removed and its push failed. Internal: #remove_job and JobSet's batched retry_all, which removes a page before pushing it.

Past maxmemory the push is refused, and so is a plain ZADD putting the entry back. Dragonfly refuses both forms: all that is left is the log.



100
101
102
103
104
105
106
107
108
 
# File 'lib/wurk/sorted_entry.rb', line 100

def restore_after_failed_push
  Wurk.redis do |conn|
    conn.call('ZADD', @parent.name, @score, value)
  rescue RedisClient::OutOfMemoryError => e
    Fetcher::OomDrain.zadd(conn, e, @parent.name, @score, value)
  end
rescue StandardError => e
  Wurk.logger.error("#{@parent.name} entry lost, ZADD refused (#{e.class}): #{loggable_payload}")
end

#retry ⇒ Object

Same flow but decrements retry_count first: the count was already incremented when the job entered the retry set, and the next failure bumps it again — without the decrement a manual "Retry now" would consume an attempt. Wire-compat with Sidekiq's SortedEntry#retry.



74
75
76
77
78
79
80
 
# File 'lib/wurk/sorted_entry.rb', line 74

def retry
  remove_job do |message|
    job = Wurk.load_json(message)
    job['retry_count'] = job['retry_count'].to_i - 1 if job['retry_count']
    Client.new.push(job)
  end
end