Class: Wurk::Queue

Inherits:
Object
  • Object
show all
Includes:
Enumerable, API::Fast::QueueExt
Defined in:
lib/wurk/queue.rb

Overview

Inspects / iterates one named queue against the canonical Sidekiq schema: LIST at queue:<name>, membership in the queues SET. Wire-compat is sacred — every Redis call below matches OSS exactly.

Pause/unpause is a Pro feature upstream; Wurk ships it free. Membership of the paused SET drives both paused? and the fetcher's queue filter (see Wurk::Fetcher::Reliable#queues_cmd). In-flight jobs continue to completion — pausing only stops new fetches.

Fetchers answer from a PAUSED_TTL cache rather than re-reading the SET every pass, so pause!/unpause! expire this process's copy inline and other processes pick the change up within the TTL. paused? is a reporting call and always reads Redis.

Spec: docs/target/sidekiq-free.md §19.2, sidekiq-pro.md §6.

Defined Under Namespace

Classes: Snapshot

Constant Summary collapse

PAGE_SIZE =

Page size for LRANGE traversal. Matches upstream so dashboards observing Redis traffic see the same pattern.

50

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(name = 'default') ⇒ Queue

Returns a new instance of Queue.



44
45
46
47
 
# File 'lib/wurk/queue.rb', line 44

def initialize(name = 'default')
  @name = name.to_s
  @rname = Keys.queue(@name)
end

Instance Attribute Details

#name ⇒ Object (readonly) Also known as: id

Returns the value of attribute name.



35
36
37
 
# File 'lib/wurk/queue.rb', line 35

def name
  @name
end

Class Method Details

.all ⇒ Array<Queue>

Returns one per known queue, sorted by name.

Returns:

  • (Array<Queue>) —

    one per known queue, sorted by name.



39
40
41
42
 
# File 'lib/wurk/queue.rb', line 39

def self.all
  names = Wurk.redis(idempotent: true) { |conn| conn.call('SMEMBERS', Keys::QUEUES_SET) }
  names.sort.map { |n| new(n) }
end

Instance Method Details

#as_json(_options = nil) ⇒ Object 



144
145
146
 
# File 'lib/wurk/queue.rb', line 144

def as_json(_options = nil)
  { name: @name }
end

#clear ⇒ Object Also known as: 💣

UNLINK the list + drop the queue from the queues set in one MULTI, as upstream does, so no reader sees the queue listed but gone. Method name is Sidekiq wire-compat — clear? would break the alias.

Unlike pause!/unpause!, this one can't claim apply-safety: a replay after a lost reply would UNLINK whatever a producer enqueued in between.



133
134
135
136
137
138
139
140
141
 
# File 'lib/wurk/queue.rb', line 133

def clear
  Wurk.redis do |conn|
    conn.multi do |tx|
      tx.call('UNLINK', @rname)
      tx.call('SREM', Keys::QUEUES_SET, @name)
    end
  end
  true
end

#delete_by_class(klass) ⇒ Integer Originally defined in module API::Fast::QueueExt

Returns payloads removed.

Parameters:

  • klass (Class, String, Symbol)

Returns:

  • (Integer) —

    payloads removed.

Raises:

  • (ArgumentError)

#delete_job(jid) ⇒ String? Originally defined in module API::Fast::QueueExt

Returns the deleted job's JSON, or nil when no job in the queue has that jid — Pro's contract, so if q.delete_job(jid) branches on whether anything was removed.

Returns:

  • (String, nil) —

    the deleted job's JSON, or nil when no job in the queue has that jid — Pro's contract, so if q.delete_job(jid) branches on whether anything was removed.

Raises:

  • (ArgumentError)

#each ⇒ Object

Paged LRANGE traversal. Yields JobRecord per payload. Continues paging until Redis returns < PAGE_SIZE rows.

Deleting a yielded record shifts every later index down by one; like upstream, each page start is pulled back by how much the list shrank since iteration began, so each(&:delete) empties the queue instead of skipping every other page.



105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
 
# File 'lib/wurk/queue.rb', line 105

def each
  return enum_for(:each) unless block_given?

  initial_size = size
  shrunk = 0
  page = 0
  loop do
    slice = page_from([(page * PAGE_SIZE) - shrunk, 0].max)
    slice.each { |value| yield JobRecord.new(value, @name) }
    break if slice.size < PAGE_SIZE

    page += 1
    shrunk = initial_size - size
  end
end

#find_job(jid) ⇒ Object

O(n) scan. Returns nil if no job matches.



122
123
124
125
 
# File 'lib/wurk/queue.rb', line 122

def find_job(jid)
  each { |record| return record if record.jid == jid }
  nil
end

#latency ⇒ Object

Seconds since the oldest job (tail of LIST) was enqueued. 0.0 when empty.



54
55
56
 
# File 'lib/wurk/queue.rb', line 54

def latency
  latency_of(Wurk.redis(idempotent: true) { |conn| conn.call('LRANGE', @rname, -1, -1).first })
end

#pause! ⇒ Object

Pause new fetches against this queue. Idempotent; like Pro, returns true only when this call paused it (SADD added the name), false when it was already paused. In-flight jobs are untouched.



85
86
87
88
89
 
# File 'lib/wurk/queue.rb', line 85

def pause!
  added = Wurk.redis(idempotent: true) { |conn| conn.call('SADD', Keys::PAUSED_SET, @name) }
  Fetcher::Reliable.invalidate_paused_cache!
  added == 1
end

#paused? ⇒ Boolean

True iff this queue's name is a member of the paused SET. Wurk implements the Pro contract for free; fetchers consult the same set.

Returns:

  • (Boolean) 


78
79
80
 
# File 'lib/wurk/queue.rb', line 78

def paused?
  Wurk.redis(idempotent: true) { |conn| conn.call('SISMEMBER', Keys::PAUSED_SET, @name) } == 1
end

#size ⇒ Object 



49
50
51
 
# File 'lib/wurk/queue.rb', line 49

def size
  Wurk.redis(idempotent: true) { |conn| conn.call('LLEN', @rname) }
end

#snapshot(offset, count) ⇒ Snapshot

The four gauges and the count jobs from offset (the order #each walks), read in one round trip rather than #each's page-by-page walk from the head. Internal reader for the dashboard and the HTTP API.

Returns:



63
64
65
66
67
68
69
70
71
72
73
74
 
# File 'lib/wurk/queue.rb', line 63

def snapshot(offset, count)
  size, tail, paused, rows = Wurk.redis(idempotent: true) do |conn|
    conn.pipelined do |pipe|
      pipe.call('LLEN', @rname)
      pipe.call('LRANGE', @rname, -1, -1)
      pipe.call('SISMEMBER', Keys::PAUSED_SET, @name)
      pipe.call('LRANGE', @rname, offset, offset + count - 1)
    end
  end
  Snapshot.new(size: size, latency: latency_of(tail.first), paused: paused == 1,
               records: rows.map { |value| JobRecord.new(value, @name) })
end

#unpause! ⇒ Object

Resume fetches. Idempotent; true only when this call unpaused it.



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

def unpause!
  removed = Wurk.redis(idempotent: true) { |conn| conn.call('SREM', Keys::PAUSED_SET, @name) }
  Fetcher::Reliable.invalidate_paused_cache!
  removed == 1
end