Class: Wurk::Fetcher::Reliable::UnitOfWork

Inherits:
Struct
  • Object
show all
Defined in:
lib/wurk/fetcher/unit_of_work.rb

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(queue = nil, job = nil, config = nil) ⇒ UnitOfWork

The positional form is upstream's documented UnitOfWork.new(queue, job, config), which gem test suites build directly. No fetcher means no private list, so #acknowledge has nothing to do. Anonymous ** is the cheap spelling for the fetch path, which always passes keywords: one allocation and ~230ns per job (a named **fields costs two).



49
50
51
52
53
54
55
 
# File 'lib/wurk/fetcher/unit_of_work.rb', line 49

def initialize(queue = nil, job = nil, config = nil, **)
  if queue
    super(queue: queue, queue_name: queue.delete_prefix('queue:'), job: job, config: config, **)
  else
    super(**)
  end
end

Instance Attribute Details

#config ⇒ Object

Returns the value of attribute config

Returns:

  • (Object) —

    the current value of config



42
43
44
 
# File 'lib/wurk/fetcher/unit_of_work.rb', line 42

def config
  @config
end

#fetcher ⇒ Object

Returns the value of attribute fetcher

Returns:

  • (Object) —

    the current value of fetcher



42
43
44
 
# File 'lib/wurk/fetcher/unit_of_work.rb', line 42

def fetcher
  @fetcher
end

#jid ⇒ Object

Returns the value of attribute jid

Returns:

  • (Object) —

    the current value of jid



42
43
44
 
# File 'lib/wurk/fetcher/unit_of_work.rb', line 42

def jid
  @jid
end

#job ⇒ Object

Returns the value of attribute job

Returns:

  • (Object) —

    the current value of job



42
43
44
 
# File 'lib/wurk/fetcher/unit_of_work.rb', line 42

def job
  @job
end

#private_queue ⇒ Object

Returns the value of attribute private_queue

Returns:

  • (Object) —

    the current value of private_queue



42
43
44
 
# File 'lib/wurk/fetcher/unit_of_work.rb', line 42

def private_queue
  @private_queue
end

#queue ⇒ Object

Returns the value of attribute queue

Returns:

  • (Object) —

    the current value of queue



42
43
44
 
# File 'lib/wurk/fetcher/unit_of_work.rb', line 42

def queue
  @queue
end

#queue_name ⇒ Object

Returns the value of attribute queue_name

Returns:

  • (Object) —

    the current value of queue_name



42
43
44
 
# File 'lib/wurk/fetcher/unit_of_work.rb', line 42

def queue_name
  @queue_name
end

#slot_key ⇒ Object

Returns the value of attribute slot_key

Returns:

  • (Object) —

    the current value of slot_key



42
43
44
 
# File 'lib/wurk/fetcher/unit_of_work.rb', line 42

def slot_key
  @slot_key
end

#slot_token ⇒ Object

Returns the value of attribute slot_token

Returns:

  • (Object) —

    the current value of slot_token



42
43
44
 
# File 'lib/wurk/fetcher/unit_of_work.rb', line 42

def slot_token
  @slot_token
end

Class Method Details

.claimed(queue, queue_name, private_queue, job, config, fetcher) ⇒ Object

The fetch path's constructor. Positional, filling the members straight in: the keyword form costs the caller's Hash, a second one for the ** forward, and Struct's keyword parsing — per job, for six fields whose names never change.



61
62
63
64
65
66
67
68
69
70
 
# File 'lib/wurk/fetcher/unit_of_work.rb', line 61

def self.claimed(queue, queue_name, private_queue, job, config, fetcher) # rubocop:disable Metrics/ParameterLists
  uow = allocate
  uow.queue = queue
  uow.queue_name = queue_name
  uow.private_queue = private_queue
  uow.job = job
  uow.config = config
  uow.fetcher = fetcher
  uow
end

Instance Method Details

#acknowledge ⇒ Object

Deferred, never skipped: the LREM goes back to the fetcher, which pipelines it in front of the next fetch's LMOVE instead of spending a round trip of its own. Ordering against the job is unchanged — the LREM still happens only after success or retry handling (Pro §3.2) — so all that moves is the wall clock. Every path that stops fetching flushes first; see #flush_pending_acks.



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

def acknowledge
  fetcher&.defer_ack(self)
end

#release_claim ⇒ Object

The Processor has settled this unit, one way or the other: stop vouching for its payload. See Fetcher::SelfReconcile.



158
159
160
 
# File 'lib/wurk/fetcher/unit_of_work.rb', line 158

def release_claim
  ::Thread.current.thread_variable_set(CLAIM_KEY, nil)
end

#release_slot ⇒ Object

The same release for the one path that deliberately does not ACK: Wurk::Shutdown, where the payload stays in the private list to be requeued and re-run somewhere else. Whoever re-runs it needs the capacity, so this cannot wait out the TTL — and there is no ACK left to ride, hence the round trip. Bounded to the jobs a hard shutdown kills mid-flight.



122
123
124
125
126
127
 
# File 'lib/wurk/fetcher/unit_of_work.rb', line 122

def release_slot
  return unless slot_key

  QueueSlot::HELD.drop(slot_token, slot_key)
  config.redis(idempotent: true) { |conn| conn.call('ZREM', slot_key, slot_token) }
end

#release_slot_in(pipe) ⇒ Object

Give a global-concurrency slot back inside a pipeline the caller already has open — the ACK's, so a capped job costs no round trip more on the way out than an uncapped one does on the way in. The ZREM lands ahead of the fetch it is pipelined with, so a thread re-competing for the queue it just ran releases before it asks, never after.

Idempotent by construction (QueueSlot#release): it names one member, so a replayed ACK frees nothing that has since been handed on. The member names this claim rather than this thread, which is what makes that true of a stale ACK too — one handed back by a failed flush (Fetcher::Reliable#restore_pending_acks) after the thread has already claimed its next slot on the same queue. See QueueSlot.claim_token.



111
112
113
114
 
# File 'lib/wurk/fetcher/unit_of_work.rb', line 111

def release_slot_in(pipe)
  QueueSlot::HELD.drop(slot_token, slot_key)
  pipe.call('ZREM', slot_key, slot_token)
end

#requeue ⇒ Object

Puts a claimed job back without running it — the public half of the Sidekiq UnitOfWork contract that throttling/rate-limiting gems call (sidekiq-throttled's requeue path, anything holding a BasicFetch UoW, which is this one under the Sidekiq::BasicFetch alias). Sidekiq's BasicFetch has no private list, so its bare RPUSH is the whole move; here the private copy has to go in the same step, or the job sits in both lists and runs twice — once from the public queue, once when the reaper reclaims this process's private list.

RPUSH then LREM in one script, unconditionally: unlike bulk_requeue (which races a Processor ACKing the same UoW and so guards the push on the private copy), the caller here owns the unit, and a job whose ACK already went out must still be re-queued rather than dropped. A script because a MULTI would apply the LREM after a refused RPUSH, losing the job. Unconditional is also why it goes by call_once: redis-client re-sends on a dropped socket, and the replay would push the job a second time. A raise leaves the outcome unknown, and both outcomes are safe: moved, or still in the private list, where Fetcher::SelfReconcile finds it.



148
149
150
151
152
153
154
 
# File 'lib/wurk/fetcher/unit_of_work.rb', line 148

def requeue
  config.redis do |conn|
    next conn.call_once('RPUSH', queue, job) unless private_queue

    Wurk::Lua::Loader.eval_once(conn, :requeue_unit, keys: [private_queue, queue], argv: [job])
  end
end

#write_ack(pipe) ⇒ Object

Queue this unit's ACK into an already-open pipeline. The counter DEL rides the same round trip rather than taking one of its own: a per-job call would be a fetch+execute regression for the sake of a key that exists for roughly no jobs. See Middleware::PoisonPill.

Sending it only for the jobs that own one is not available to us: whether a job was reclaimed lives in the counter, and a reclaimed payload is byte-identical to a first-attempt one (the job JSON is wire-frozen, so the reaper cannot flag it). Reading the counter to decide would spend the very round trip the DEL is riding for free.



92
93
94
95
96
97
 
# File 'lib/wurk/fetcher/unit_of_work.rb', line 92

def write_ack(pipe)
  pipe.call_v(['LREM', private_queue, LREM_COUNT, job])
  job_jid = jid.to_s
  Middleware::PoisonPill.clear_in(pipe, job_jid) unless job_jid.empty?
  release_slot_in(pipe) if slot_key
end