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

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

Overview

Carries the public queue key, the raw (still-JSON) job payload, the capsule we use to reach Redis, and the fetcher that holds this unit's ACK until it can ride a pipeline. ACK removes from the private list; requeue pushes back to the public queue head so the job is next pulled. LREM count=1 is idempotent for our payloads since each job's JSON contains a unique jid.

queue_name (the queue without its queue: prefix) and private_queue come off the fetcher's per-queue cache at build time: both are pure functions of the queue this unit came from, so deriving them here would be a per-job cost for a per-queue fact.

jid is filled in by the Processor once it has parsed the payload — the fetcher never parses. It is only used to retire the job's poison-pill recovery counter, so an ACK without one is still a complete ACK.

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



72
73
74
 
# File 'lib/wurk/fetcher/reliable.rb', line 72

def config
  @config
end

#fetcher ⇒ Object

Returns the value of attribute fetcher

Returns:

  • (Object) —

    the current value of fetcher



72
73
74
 
# File 'lib/wurk/fetcher/reliable.rb', line 72

def fetcher
  @fetcher
end

#jid ⇒ Object

Returns the value of attribute jid

Returns:

  • (Object) —

    the current value of jid



72
73
74
 
# File 'lib/wurk/fetcher/reliable.rb', line 72

def jid
  @jid
end

#job ⇒ Object

Returns the value of attribute job

Returns:

  • (Object) —

    the current value of job



72
73
74
 
# File 'lib/wurk/fetcher/reliable.rb', line 72

def job
  @job
end

#private_queue ⇒ Object

Returns the value of attribute private_queue

Returns:

  • (Object) —

    the current value of private_queue



72
73
74
 
# File 'lib/wurk/fetcher/reliable.rb', line 72

def private_queue
  @private_queue
end

#queue ⇒ Object

Returns the value of attribute queue

Returns:

  • (Object) —

    the current value of queue



72
73
74
 
# File 'lib/wurk/fetcher/reliable.rb', line 72

def queue
  @queue
end

#queue_name ⇒ Object

Returns the value of attribute queue_name

Returns:

  • (Object) —

    the current value of queue_name



72
73
74
 
# File 'lib/wurk/fetcher/reliable.rb', line 72

def queue_name
  @queue_name
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.



80
81
82
 
# File 'lib/wurk/fetcher/reliable.rb', line 80

def acknowledge
  fetcher.defer_ack(self)
end

#requeue ⇒ Object 



100
101
102
 
# File 'lib/wurk/fetcher/reliable.rb', line 100

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.



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

def write_ack(pipe)
  pipe.call('LREM', private_queue, 1, job)
  job_jid = jid.to_s
  Middleware::PoisonPill.clear_in(pipe, job_jid) unless job_jid.empty?
end