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

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

Instance Attribute Summary collapse

Instance Method Summary collapse

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

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.



50
51
52
 
# File 'lib/wurk/fetcher/unit_of_work.rb', line 50

def acknowledge
  fetcher.defer_ack(self)
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.



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

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.



83
84
85
86
 
# File 'lib/wurk/fetcher/unit_of_work.rb', line 83

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

#requeue ⇒ Object 



101
102
103
 
# File 'lib/wurk/fetcher/unit_of_work.rb', line 101

def requeue
  config.redis { |conn| conn.call('RPUSH', queue, job) }
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.



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

def write_ack(pipe)
  pipe.call('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