Module: Wurk::Debounce

Defined in:
lib/wurk/debounce.rb

Overview

Debounce — collapse a burst of enqueues into one job carrying the last payload, fired after a period of quiet. A Wurk extra, not a Sidekiq surface, and deliberately not a second flavour of Unique: unique rejects the duplicate and keeps the first, debounce replaces the pending job and keeps the last.

The classic use is a re-derivation triggered by something a user does repeatedly — reindex a record, warm a cache, recompute a projection. Fifty edits in a minute should produce one rebuild at the end, not fifty.

Redis schema — one new key, plus an ordinary entry in the ZSET that already exists:

debounce:<digest>   HASH   member (the pending job JSON), first (epoch)
schedule            ZSET   the collapsed job, exactly as `perform_in`
                         would have written it

Nothing reading schedule can tell a debounced job from any other scheduled one — ScheduledSet, the dashboard and a stock Sidekiq poller all handle it unchanged. There is no parallel delayed structure.

Identity is Unique's digest of [class, queue, args], so a class that already narrows its key with sidekiq_unique_context narrows its debounce key the same way — that hook is the supported "collapse on a subset of the args" escape hatch, and there is no second one.

Defined Under Namespace

Classes: Outcome

Constant Summary collapse

GRACE =

How long debounce:<digest> outlives the entry it points at.

The key's only job is to remember which member to pull back out of schedule when the burst is extended, so it must not expire while that member is still pending — a burst that lost its key ZADDs a second entry next to the first and stops collapsing. The window that matters is therefore "fire time → poller promotes it", which the scheduler keeps at roughly average_scheduled_poll_interval across the cluster however many processes are running. Five minutes is far past that; overshooting costs one ~100-byte key per idle debounce identity, and the script treats an outlived key as a finished burst anyway.

300

Class Method Summary collapse

Class Method Details

.key_for(job) ⇒ String

Returns debounce:<digest> for that job's identity.

Parameters:

  • job (Hash) —

    a normalized job payload

Returns:

  • (String) —

    debounce:<digest> for that job's identity



64
 
# File 'lib/wurk/debounce.rb', line 64

def key_for(job) = Keys.debounce(Unique.digest_for(job))

.schedule(job, wait:, max_wait: nil, pool: nil) ⇒ Outcome

Collapse job into the pending entry for its key, or open a new burst.

Parameters:

  • job (Hash) —

    normalized payload; stored minus at/enqueued_at, the same bytes Client#push_scheduled would have written

  • wait (Numeric) —

    seconds of quiet before the job fires; every further enqueue of the same key pushes this out again

  • max_wait (Numeric, nil) (defaults to: nil) —

    hard cap measured from the first enqueue of the burst. nil leaves the job able to starve — a key re-enqueued faster than wait never fires — so callers that cannot rule that out should always set it.

  • pool (#with, nil) (defaults to: nil) —

    defaults to this process's pool

Returns:



78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
 
# File 'lib/wurk/debounce.rb', line 78

def schedule(job, wait:, max_wait: nil, pool: nil)
  seconds = positive_seconds!(wait, 'wait')
  cap = max_wait.nil? ? 0.0 : positive_seconds!(max_wait, 'max_wait')
  reject_shorter_cap!(seconds, cap)

  argv = [seconds, cap, JobUtil.scheduled_member(job), GRACE]
  # Replay-safe: a re-run finds the member it just wrote, ZREMs it and
  # ZADDs it back, so a pool retry after a lost reply converges on one
  # entry rather than a duplicate — the whole point of doing this in a
  # script rather than a pipeline.
  #
  # The replay is not byte-identical: it recomputes `now`, so the fire
  # time moves out by the retry's own latency (and `first` is carried
  # forward, so `max_wait` still caps where it lands). That is the claim
  # worth making anyway. A lost reply means the write probably applied,
  # and refusing the replay only converts a bounded shift into an
  # exception out of `perform_async` for a job that is already scheduled
  # — whose caller then re-enqueues and shifts the fire time regardless.
  raw = with_pool(pool, idempotent: true) do |conn|
    Lua::Loader.eval_cached(conn, :debounce, keys: [key_for(job), Keys::SCHEDULE], argv: argv)
  end
  Outcome.new(raw[0].to_i == 1, raw[1].to_f, raw[2].to_f)
end