Class: Wurk::Batch::ServerMiddleware

Inherits:
Object
  • Object
show all
Includes:
Middleware::ServerMiddleware
Defined in:
lib/wurk/batch/server_middleware.rb

Overview

Server middleware. Runs around perform for any job carrying a bid. On success → BATCH_ACK_SUCCESS → when that drains the batch, fire its callbacks (Callbacks.maybe_fire).

On a job raising (and thus heading to retry), records a transient failure → BATCH_ACK_FAILED → the jid joins b-<bid>-failed and failures reflects the count of currently-failing jobs. A later successful retry clears it; a terminal death moves it to b-<bid>-died. Clean handled exits (JobRetry::Skip — from the interrupt handler or a Limiter::Rescheduled re-enqueue — and cooperative IterableJob interruption) are re-raised as neither success nor failure.

The success ack runs outside the job's rescue: once perform returned, nothing the batch bookkeeping raises may be mistaken for the job failing, or a job that already did its work is retried and runs twice. The ack itself is retried in-process and only raises (sending the job to retry, at-least-once) when Redis stays unreachable — a jid left in the live set would hold the batch open forever. The fire after it never raises: it is re-driven in-process and then reported to the error handlers.

Invalidated batches short-circuit: the job is skipped without raising — counts as a "success" for batch purposes per spec §12. A callback job is never skipped: it rides in its parent batch (see Callbacks), and a cancelled parent must not swallow a child's callback.

Death handling lives in Wurk::Batch::DeathHandler (registered as a config death_handler) because death is signalled from the retry layer, not from this middleware's rescue path.

Constant Summary collapse

CALLBACK_JOB = 
'Wurk::Batch::CallbackJob'

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Instance Attribute Details

#config ⇒ Object Originally defined in module Middleware::ServerMiddleware

Returns the value of attribute config.

Class Method Details

.outer_of ⇒ Object

The middleware this one must stay OUTSIDE of — the ordering anchor Chain#add consults before appending.

Expiry SKIPS an expired job by returning without yielding: inside the batch onion the batch's ack_success never runs, the batch's pending never drops and the batch never completes. The Limiter turns an over-limit job into a Limiter::Rescheduled (a JobRetry::Skip): inside the batch onion the batch's generic rescue StandardError books a spurious failure for a job that never ran.

A plain add appends, so re-registering this middleware (a configure_server block, a test helper) would move it inside both and break the batch's accounting. The anchor keeps it outside.

Resolved at call time, not at load time: Expiry is loaded after this file (lib/wurk.rb), so naming the constant here would raise NameError.



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

def self.outer_of
  anchors = []
  anchors << Wurk::Middleware::Expiry if defined?(Wurk::Middleware::Expiry)
  anchors << Wurk::Limiter::ServerMiddleware if defined?(Wurk::Limiter::ServerMiddleware)
  anchors
end

.shared_instance? ⇒ Boolean

One instance per chain entry, shared by every job (Middleware::Chain::Entry): #call keeps its state in locals and writes no ivar.

Returns:

  • (Boolean) 


44
 
# File 'lib/wurk/batch/server_middleware.rb', line 44

def self.shared_instance? = true

Instance Method Details

#call(_worker, job, _queue) ⇒ Object 



71
72
73
74
75
76
77
78
 
# File 'lib/wurk/batch/server_middleware.rb', line 71

def call(_worker, job, _queue, &)
  bid = job['bid']
  return yield unless bid

  jid = job['jid']
  run(bid, jid, &) if job['class'] == CALLBACK_JOB || !invalidated?(bid)
  ack_success(bid, jid)
end

#logger ⇒ Object Originally defined in module Middleware::ServerMiddleware

#redis(idempotent: false) ⇒ Object Originally defined in module Middleware::ServerMiddleware

#redis_pool ⇒ Object Originally defined in module Middleware::ServerMiddleware