Class: Wurk::Scheduled::Enq

Inherits:
Object
  • Object
show all
Includes:
Component
Defined in:
lib/wurk/scheduled.rb

Overview

Drains both SETS each call, running every promoted job through the client middleware chain.

Claims rather than pops: scheduled_claim rescores due members to a lease expiry, the job is pushed, and only then is the member ZREMed. Between any two of those steps the job exists in its set, so a lost reply (the re-sent claim just leases a further batch), an out-of-memory push or a kill -9 ends in a re-promotion once the lease lapses. The failure mode is therefore a duplicate and a delay of up to CLAIM_LEASE, never a lost job. Other pollers skip a claimed member because its score has left their due window.

Constant Summary collapse

LUA_ZPOPBYSCORE =

Public per spec §16; the claim replaced it here.

Wurk::Lua::ZPOPBYSCORE
CLAIM_BATCH =

Bounds what one crash can duplicate. The ZREMs share one round trip per batch, which is what makes this cheaper than a pop per job.

10
CLAIM_LEASE =

Seconds until a claimed member is due again. Far above a healthy batch, which takes milliseconds, so a second poller does not re-promote it.

60
RELEASE_BACKDATE = 
1.0
TRANSIENT_ERRORS = 
[RedisClient::Error, ConnectionPool::TimeoutError].freeze
STORE_REFUSAL =

The whole store is refusing, not this job. A plain CommandError with one of these prefixes says the same as OutOfMemoryError (Dragonfly's reply at maxmemory is ERR Out of memory).

/\A(?:LOADING|ERR Out of memory|OOM|MASTERDOWN|MISCONF|BUSY)/

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(container) ⇒ Enq

Returns a new instance of Enq.



60
61
62
63
64
65
 
# File 'lib/wurk/scheduled.rb', line 60

def initialize(container)
  @config = container
  @done = false
  @client = Client.new(config: container)
  @lease = (container[:scheduled_claim_lease] if container.respond_to?(:[])) || CLAIM_LEASE
end

Instance Attribute Details

#config ⇒ Object (readonly) Originally defined in module Component

Returns the value of attribute config.

Instance Method Details

#default_tag(dir = Dir.pwd) ⇒ Object Originally defined in module Component

#enqueue_jobs(sorted_sets = SETS) ⇒ Object

now is captured once per set so a slow loop on one ZSET can't keep grabbing newly-scheduled jobs from a moving window.



69
70
71
 
# File 'lib/wurk/scheduled.rb', line 69

def enqueue_jobs(sorted_sets = SETS)
  sorted_sets.each { |sset| drain_set(sset) }
end

#fire_event(event, oneshot: true, reverse: false, reraise: false) ⇒ Object Originally defined in module Component

Invokes lifecycle hooks for event. Hooks run in registration order (or LIFO when reverse: true, used for teardown). A raise in one hook is reported via handle_exception and does NOT stop the next hook unless reraise: true (used in tests / fail-fast boot). oneshot: true clears the bucket after dispatch so the event can't fire twice.

#handle_exception(ex, ctx = {}) ⇒ Object Originally defined in module Component

#hostname ⇒ Object Originally defined in module Component

#identity ⇒ Object Originally defined in module Component

#leader? ⇒ Boolean Originally defined in module Component

True iff this process currently holds the cluster dear-leader lock. Cached per Component instance for LEADER_CACHE_TTL_MS (~5s): cron and the metrics rollups call this every tick, and an uncached GET would double their Redis traffic at short intervals for no benefit — the lock's own renewal cadence (60s+, spec §6.1) easily tolerates a few-second-stale read. Returns false unconditionally when WURK_LEADER=false (or SIDEKIQ_LEADER=false) is set on the process (opt-out hot-standby). Any Redis error is swallowed → false, so a transient partition can't propagate as an exception into user code.

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

Returns:

  • (Boolean)

#logger ⇒ Object Originally defined in module Component

--- delegated to config -------------------------------------------

#mono_ms ⇒ Object Originally defined in module Component

#process_nonce ⇒ Object Originally defined in module Component

#real_ms ⇒ Object Originally defined in module Component

--- clocks ---------------------------------------------------------

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

#safe_thread(name, priority: nil, &block) ⇒ Object Originally defined in module Component

Spawns a named thread that runs block under watchdog(name). The parent must retain the returned Thread; otherwise GC may not, but report_on_exception is disabled so we don't double-log on death.

Priority resolution matches Sidekiq (component.rb:44-48): explicit argument, then config.thread_priority, then -1. Ruby's default of 0 buys a 100ms timeslice; each negative step halves it, so -1 keeps a CPU-heavy capsule from starving its siblings for a whole tick.

#terminate ⇒ Object 



73
74
75
 
# File 'lib/wurk/scheduled.rb', line 73

def terminate
  @done = true
end

#tid ⇒ Object Originally defined in module Component

#watchdog(last_words) ⇒ Object Originally defined in module Component

Wraps a block at a thread boundary: any unhandled exception is reported via handle_exception (so it lands in error_handlers / the log) and then re-raised. last_words is the component label included in the context.