Module: Wurk::Unique

Defined in:
lib/wurk/unique.rb

Overview

Sidekiq Enterprise unique jobs. Best-effort dedup at enqueue time keyed by a SHA256 digest of [class, queue, args] (overridable via sidekiq_unique_context; ActiveJob-wrapped payloads narrow to the wrapped class + its arguments — see active_job_context). Three lock-release strategies:

* `unique_until: :success` (default) — lock retained through retries;
server middleware DELs it on successful perform. Surviving across
a process crash is bounded by `unique_for` TTL.
* `unique_until: :start` — server middleware DELs the lock right
*before* invoking perform; a duplicate can be enqueued while the
first is running.

A job that dies automatically (retries exhausted / discarded) releases its lock via a death handler; manual UI kills keep the lock until TTL expiry (Ent wiki, Ent-Unique-Jobs).

Wire-compat (§3.9): single-key Redis layout — unique:<sha256> STRING holding the owning JID. Scheduled jobs extend the TTL by the delay so the lock covers the entire wait+execution window (§3.4).

This is the native replacement for the sidekiq-unique-jobs gem; see the migration guide for the lock: → unique_until: translation table.

Spec: docs/target/sidekiq-ent.md §3.

Examples:

Enable globally, then declare per worker

Sidekiq::Enterprise.unique!   # activate the middleware once, at boot

class ChargeJob
  include Sidekiq::Job
  sidekiq_options unique_for: 10.minutes, unique_until: :success

  # optional: customize the dedup key
  def self.sidekiq_unique_context(job)
    job["args"].first   # dedup on the first arg only
  end
end

Defined Under Namespace

Classes: ClientMiddleware, ConfigurationError, ServerMiddleware

Constant Summary collapse

KEY_PREFIX = 
'unique:'
DEFAULT_UNTIL = 
:success
VALID_UNTIL = 
%i[success start].freeze
ACTIVE_JOB_WRAPPERS =

Class names that carry an ActiveJob payload as their single arg. The canonical one Wurk writes is Sidekiq::ActiveJob::Wrapper; the others appear in payloads enqueued by older Sidekiq/Rails versions still sitting in Redis at gem-swap time.

[
  'Sidekiq::ActiveJob::Wrapper',
  'Wurk::ActiveJob::Wrapper',
  'ActiveJob::QueueAdapters::SidekiqAdapter::JobWrapper',
  'ActiveJob::QueueAdapters::WurkAdapter::JobWrapper'
].freeze
DEATH_HANDLER =

Ent parity: a job that dies automatically releases its lock so a duplicate can enqueue immediately. Manual API/UI kills keep the lock until TTL expiry (Ent wiki, Ent-Unique-Jobs) — they reach death handlers too (Sidekiq fires them on API kills), so we recognize the synthesized kill exception and skip the release for it. Atomic CAS-DEL via the shared Lua script mirrors ServerMiddleware#release.

lambda do |job, exception|
  next unless Wurk::Unique.enabled?
  next unless Wurk::Unique.coerce_ttl(job['unique_for'])
  next if exception.instance_of?(::RuntimeError) && exception.message == DeadSet::API_KILL_MESSAGE

  Wurk.redis { |conn| Wurk::Unique.release_if_owner(conn, Wurk::Unique.lock_key_for(job), job['jid']) }
end
NUMERIC_TTL =

Duration#to_s: "300" for 5.minutes, "90.0" for 1.5.minutes.

/\A\d+(?:\.\d+)?\z/

Class Method Summary collapse

Class Method Details

.coerce_ttl(value) ⇒ Object

Coerce unique_for to a numeric seconds value. Accepts Integer, Numeric, ActiveSupport::Duration (any to_i-respondent), a numeric String, or false (skip). Returns nil when uniqueness should be skipped.

The String is what every reader after the push sees for a Duration: JSON.generate writes 5.minutes as "300", in Sidekiq as here, so the server middleware and the death handler get "300" back off the queue. Refusing it left every unique_for: 5.minutes lock held until its TTL (tesote/tesote.com#519).



213
214
215
216
217
218
219
220
221
222
223
224
225
 
# File 'lib/wurk/unique.rb', line 213

def self.coerce_ttl(value)
  return nil if value.nil? || value == false
  # `.to_i`, not `value`: ActiveSupport::Duration overrides `is_a?(Integer)`
  # to return true (it delegates to its underlying value), so `1.hour` passes
  # this guard — returning the raw Duration handed redis-client a non-Integer
  # EX arg and raised TypeError at enqueue (#253).
  return value.to_i if value.is_a?(Integer) && value.positive?
  return value.to_i if value.is_a?(Numeric)
  return value.to_f.to_i if value.is_a?(String) && value.match?(NUMERIC_TTL)
  return value.to_i if duration_like?(value)

  nil
end

.digest_for(job) ⇒ Object

The identity digest on its own, without the unique: prefix. Debounce keys off the same digest under its own prefix, so a class that narrows its identity via sidekiq_unique_context narrows both policies through the one documented hook instead of growing a second, near-identical one.



137
138
139
 
# File 'lib/wurk/unique.rb', line 137

def digest_for(job)
  Digest::SHA256.hexdigest(JSON.dump(unique_context(job)))
end

.disable! ⇒ Object

Test helper — not part of the public Sidekiq surface. Clears the flag so per-test enable!/disable! does not leak across runs.



112
113
114
115
 
# File 'lib/wurk/unique.rb', line 112

def disable!
  @enabled = false
  nil
end

.duration_like?(value) ⇒ Boolean

Returns:

  • (Boolean) 


230
231
232
233
234
 
# File 'lib/wurk/unique.rb', line 230

def self.duration_like?(value)
  return false unless value.respond_to?(:to_i)

  value.respond_to?(:since) || value.class.name.to_s.include?('Duration')
end

.enable!(config = Wurk.configuration) ⇒ Object

config selects the chains and death handlers, installed once each; the enabled flag remains process-global, including for other configs whose chains already carry the middleware.



104
105
106
107
108
 
# File 'lib/wurk/unique.rb', line 104

def enable!(config = Wurk.configuration)
  @enabled = true
  register_middleware!(config)
  true
end

.enabled? ⇒ Boolean

Returns:

  • (Boolean) 


97
98
99
 
# File 'lib/wurk/unique.rb', line 97

def enabled?
  @enabled == true
end

.lock_key(klass, queue, args) ⇒ Object

Compute the lock key for an arbitrary (queue, klass, args) triple, bypassing sidekiq_unique_context. Public so operators can compute the key of a worker that uses the default context (docs §8).



120
121
122
123
 
# File 'lib/wurk/unique.rb', line 120

def lock_key(klass, queue, args)
  context = [klass.to_s, queue.to_s, args]
  "#{KEY_PREFIX}#{Digest::SHA256.hexdigest(JSON.dump(context))}"
end

.lock_key_for(job) ⇒ Object

Compute the lock key from a job payload, honoring sidekiq_unique_context when the worker class is loaded and defines it.



128
129
130
 
# File 'lib/wurk/unique.rb', line 128

def lock_key_for(job)
  "#{KEY_PREFIX}#{digest_for(job)}"
end

.locked?(queue_or_klass, klass_or_args = nil, args = nil) ⇒ String?

Returns owning jid, or nil when the lock is free.

Returns:

  • (String, nil) —

    owning jid, or nil when the lock is free.



241
242
243
244
245
246
247
248
 
# File 'lib/wurk/unique.rb', line 241

def self.locked?(queue_or_klass, klass_or_args = nil, args = nil)
  queue, klass, payload = normalize_locked_args(queue_or_klass, klass_or_args, args)
  # Routed through lock_key_for, not lock_key: the probe must honor
  # `sidekiq_unique_context` (and the ActiveJob default) or it reports
  # "free" for every worker that customizes its digest.
  key = lock_key_for('class' => klass.to_s, 'queue' => queue.to_s, 'args' => payload)
  Wurk.redis { |c| c.call('GET', key) }
end

.release_if_owner(conn, key, jid) ⇒ Object

Atomic compare-and-delete of a unique lock key. Two-command GET-then-DEL is not a real CAS — the key can expire between the GET and DEL, letting a fresh enqueue grab it, and the bare DEL would then drop the new owner's lock. Routed through a single Lua script (Wurk::Lua::RELEASE_IF_OWNER) shared by ServerMiddleware#release (normal success/start release) and DEATH_HANDLER (automatic-death release) so the two paths cannot drift.



88
89
90
 
# File 'lib/wurk/unique.rb', line 88

def self.release_if_owner(conn, key, jid)
  Wurk::Lua::Loader.eval_cached(conn, :release_if_owner, keys: [key], argv: [jid])
end

.unique_context(job) ⇒ Object

Default: [class, queue, args]. Workers may override by defining self.sidekiq_unique_context(job) returning any JSON-serializable value (e.g. a subset of args). Spec §3.5. ActiveJob-wrapped payloads get a narrowed default (see active_job_context) because their raw args can never repeat.



146
147
148
149
150
151
 
# File 'lib/wurk/unique.rb', line 146

def unique_context(job)
  klass = resolve_class(job['class'])
  return klass.sidekiq_unique_context(job) if klass.respond_to?(:sidekiq_unique_context)

  active_job_context(job) || [job['class'], job['queue'], job['args']]
end