Class: Wurk::Middleware::Expiry

Inherits:
Object
  • Object
show all
Includes:
ServerMiddleware
Defined in:
lib/wurk/middleware/expiry.rb

Overview

Server middleware. Drops a job whose window has closed before perform gets a chance to start. Two options open such a window, both stamped onto the payload as an absolute epoch-float at push (JobUtil#finalize), and the earlier of the two is the one that counts:

* `expiry`, from Pro's `sidekiq_options expires_in:`. Once `perform` is
invoked it no longer preempts — a long-running job that started in
time finishes.
* `deadline_at`, from `sidekiq_options deadline:`. This one does preempt:
{Timeout} arms the watchdog with whatever is left of it, and the
{Wurk::Job::DeadlineExceeded} that lands in a job still running past
its cutoff unwinds to here, where it takes the same skip path. So a
deadline reaches the same terminal state whether it passed while the
job queued or while it ran — one state to reason about, one counter to
watch, and no retry either way.

The skip path:

* bumps Wurk::Processor::EXPIRED so the heartbeat flushes
`stat:expired` + `stat:expired:YYYY-MM-DD` to Redis, surfacing the
count in Wurk::Stats and the dashboard
* emits `jobs.expired` via Wurk::Metrics::Statsd (no-op when no client
is configured)
* returns without yielding — no exception, so JobRetry treats it as a
clean exit and the processor acks the UoW
* counts as a batch success: because this middleware is registered
AFTER `Wurk::Batch::ServerMiddleware`, returning unwinds back through
batch's `yield`, and batch's `ack_success` still runs on the way out

The expired job is also counted toward PROCESSED — Processor#stats's ensure block always increments PROCESSED, so EXPIRED is an additive subset (executed = processed - failed - expired). Matches Sidekiq Pro.

Spec: docs/target/sidekiq-pro.md §7.

Instance Attribute Summary

Attributes included from ServerMiddleware

#config

Instance Method Summary collapse

Methods included from ServerMiddleware

#logger, #redis, #redis_pool

Instance Method Details

#call(_job_instance, job, _queue) ⇒ Object

The rescue is scoped to the guarded call rather than the method body: only a job that carries a cutoff can be cut by one, and a job that raises DeadlineExceeded from its own code without carrying one has to unwind to JobRetry like any other failure instead of being acked.



50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
 
# File 'lib/wurk/middleware/expiry.rb', line 50

def call(_job_instance, job, _queue)
  expiry = job['expiry']
  deadline = job['deadline_at']
  return yield unless expiry || deadline

  at = cutoff(expiry, deadline)
  return yield unless at

  return drop(job) if ::Time.now.to_f > at

  begin
    yield
  rescue ::Wurk::Job::DeadlineExceeded
    drop(job)
  end
end