Module: Wurk::Collapse

Defined in:
lib/wurk/collapse.rb

Overview

The enqueue door for Wurk's two collapse policies. Debounce and Throttle are the mechanisms; this is the single option a worker class writes to reach them, and the one place the company they cannot keep is refused.

sidekiq_options collapse: { policy: :debounce, wait: 5, max_wait: 60 }
sidekiq_options collapse: { policy: :throttle, slot: 60 }

Why collapse: rather than unique:

The slice plan sketched unique: { policy: … }, on the reasonable grounds that all three policies share one identity digest and one narrowing hook (sidekiq_unique_context). That name is taken. sidekiq-unique-jobs v8.1.0 — the gem this feature replaces, and one Wurk's ecosystem suite runs against — still recognizes sidekiq_options unique: as the deprecated spelling of its own lock: (SidekiqUniqueJobs::Lock::Validator::DEPRECATED_KEYS maps :unique => :lock). Claiming the key would make this validator reject a host's working pre-v6 declaration, which is the drop-in contract breaking on exactly the gem it is meant to make unnecessary. collapse is unclaimed there, and says what both policies do.

One policy per class

collapse: and unique_for: answer the same question differently — collapse a burst, or reject the duplicate — and they do not compose: a lock drops the very re-enqueue debounce needs in order to extend a burst, and which of the two won would come down to client-chain registration order. A class declares one; declaring both is refused where it was written.

What a collapsed enqueue returns

nil, from both policies. For a dropped throttle that is Throttle's documented answer and Unique's long-standing one. For a debounce it is the less obvious half: this push's payload really is in schedule, under this push's jid — but only until the next enqueue of the same identity replaces it, so a jid handed back here invites a caller to poll, or cancel, a member a sibling push has already ZREM'd. Debounce.schedule returns its Debounce::Outcome for code that needs the detail; the enqueue door does not.

Defined Under Namespace

Classes: ClientMiddleware, ConfigurationError, Policy

Constant Summary collapse

OPTION =

The job-hash key, and the sidekiq_options key it is written as.

'collapse'
PARAMS =

Per-policy parameter rules. whole marks a slot boundary, which is a wall-clock landmark every producer has to land on identically and so cannot be a fraction of a second (Throttle refuses one too).

Values are checked here and not left to the two modules, which raise at enqueue: a slot: 1.5 that only fails the first time a rarely-exercised worker is pushed has already shipped.

{
  debounce: { required: %i[wait].freeze, optional: %i[max_wait].freeze, whole: false },
  throttle: { required: %i[slot].freeze, optional: [].freeze, whole: true }
}.freeze

Class Method Summary collapse

Class Method Details

.policy_for(options) ⇒ Policy?

The collapse policy a set of job options declares, or nil for none.

Parsing and validating are one pass, shared by every door: both sidekiq_options copies (worker and ActiveJob), which keep only the raising, and ClientMiddleware, which acts on the result. One definition means "valid where it was written" and "usable where it is read" cannot drift apart — the same reason JobUtil.positive_seconds? is shared rather than reimplemented per reader.

Strict on every axis, deliberately: slot: written on a debounce is a typo for wait:, and a silently ignored one leaves a worker collapsing on a default nobody chose.

Parameters:

  • options (Hash) —

    string-keyed job options or a job payload

Returns:

Raises:



95
96
97
98
99
100
101
102
103
104
105
106
 
# File 'lib/wurk/collapse.rb', line 95

def policy_for(options)
  declared = options[OPTION]
  return nil if declared.nil?

  reject_unique!(options, declared)
  params = symbolized(declared)
  name = policy_name!(params.delete(:policy), declared)
  check_keys!(name, params, declared)
  check_seconds!(name, params, declared)
  reject_short_cap!(params, declared) if name == :debounce
  Policy.new(name, params)
end

.reject_bulk!(klass, declared) ⇒ Object

push_bulk exists to amortize one round trip across many jobs: a same-queue bulk of 1,000 is still the one SADD + variadic LPUSH pipeline a single push is. A collapse policy is a decision per job — an EVALSHA whose KEYS are that job's own identity — so it cannot ride that pipeline, and applying it per item turns the one round trip into N. Measured here, 1,000 jobs onto a loopback Redis, INFO commandstats either side:

push_bulk, no policy   ~14 ms    2 commands,     1 round trip
per-item collapse     ~320 ms    6,000 commands, 1,000 round trips

~22x wall clock and three orders of magnitude of round trips, and both grow with the batch. That is a structural cost, not a constant worth tuning away, which is what settles step 6 of the slice plan against applying policies per item.

It is also meaningless in the shape bulk is used: perform_bulk pushes one class with N distinct argument lists, and distinct arguments are distinct identities, so nothing collapses with anything. Every jid in the returned array would come back nil, which in push_bulk's contract means "middleware halted this one" — indistinguishable from a debounce that did write. Raising is the only outcome that is neither a silent regression nor a lie about what happened.

Raises:



132
133
134
135
136
137
138
139
140
 
# File 'lib/wurk/collapse.rb', line 132

def reject_bulk!(klass, declared)
  return if declared.nil?

  raise ConfigurationError,
        "#{klass} declares `collapse: #{declared.inspect}` and cannot be enqueued with push_bulk: " \
        'a collapse policy decides per job, which would cost one Redis round trip per item where ' \
        'bulk enqueue costs one for the whole batch, and every returned jid would be nil. ' \
        'Push these jobs individually, or drop the policy on this worker.'
end