Class: Wurk::Client

Inherits:
Object
  • Object
show all
Includes:
JobUtil
Defined in:
lib/wurk/client.rb,
lib/wurk/client/buffered.rb

Overview

Enqueue interface. Pipelined LPUSH / ZADD writes against the canonical Sidekiq Redis schema — never change keys, JSON shape, or score format here: wire-compat is sacred. Most apps enqueue through the Worker DSL (MyJob.perform_async), which routes here; reach for Client directly only to push a raw job hash or to drive the bulk/scheduled path explicitly.

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

Examples:

Push a raw job hash

Wurk::Client.new.push("class" => "MyJob", "args" => [1, 2], "queue" => "default")

Bulk enqueue in one round-trip

Wurk::Client.new.push_bulk("class" => "MyJob", "args" => [[1], [2], [3]])

Defined Under Namespace

Modules: Buffered

Constant Summary collapse

DEFAULT_BATCH_SIZE =

Sidekiq mirrors these exactly. Tests against the upstream parity suite depend on the magic numbers, not just behavior.

1_000
SCHEDULED_BATCH_SIZE = 
100
SPREAD_INTERVAL_FLOOR = 
5
BATCH_PIPELINE_SLICE =

Batched (bid) payloads per EVALSHA pipeline. Ours, not Sidekiq's — it has no batches. Sized to DEFAULT_BATCH_SIZE so no existing caller's round-trip count moves: push_bulk already hands #raw_push at most that many payloads, so its batched pipeline stays exactly one round trip.

The cap is for the one path that isn't pre-sliced: autoflush = true buffers a whole Batch#jobs block, so #flush_batched can be handed an unbounded payload set. Unsliced that is one pipeline holding every command and every reply in memory at once, and — Lua being atomic and single-threaded — one uninterrupted server-side sweep that blocks every other client for its duration. Same reasoning as the LIMIT on RELIABLE_SCHEDULE_PROMOTE.

1_000
DELIVERED_KEY =

Thread-local slot holding the payloads of the current push whose Redis write is confirmed applied. Buffered subtracts them from the set it re-buffers when a later phase of the same push loses the connection, so an already-written job is never replayed into a second copy. Thread-local because one Client instance serves every producer thread; opened and closed by Buffered, the only reader, so an un-prepended Client pays a single nil check per write phase.

:wurk_client_delivered

Constants included from JobUtil

JobUtil::RETRY_FOR_MAX, JobUtil::TRANSIENT_ATTRIBUTES

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Methods included from JobUtil

#normalize_item, #now_in_millis, #validate, #verify_json

Constructor Details

#initialize(pool: nil, config: nil, chain: nil) ⇒ Client

Returns a new instance of Client.



54
55
56
57
58
 
# File 'lib/wurk/client.rb', line 54

def initialize(pool: nil, config: nil, chain: nil)
  @config     = config || Wurk.configuration
  @redis_pool = pool
  @chain      = chain || @config.client_middleware
end

Instance Attribute Details

#redis_pool ⇒ Object

Returns the value of attribute redis_pool.



52
53
54
 
# File 'lib/wurk/client.rb', line 52

def redis_pool
  @redis_pool
end

Class Method Details

.enqueue(klass) ⇒ Object 



122
 
# File 'lib/wurk/client.rb', line 122

def enqueue(klass, *) = klass.perform_async(*)

.enqueue_in(interval, klass) ⇒ Object 



132
133
134
 
# File 'lib/wurk/client.rb', line 132

def enqueue_in(interval, klass, *)
  klass.perform_in(interval, *)
end

.enqueue_to(queue, klass) ⇒ Object 



124
125
126
 
# File 'lib/wurk/client.rb', line 124

def enqueue_to(queue, klass, *)
  klass.set(queue: queue.to_s).perform_async(*)
end

.enqueue_to_in(queue, interval, klass) ⇒ Object 



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

def enqueue_to_in(queue, interval, klass, *)
  klass.set(queue: queue.to_s).perform_in(interval, *)
end

.push(item) ⇒ Object 



120
 
# File 'lib/wurk/client.rb', line 120

def push(item)        = new.push(item)

.push_bulk(items) ⇒ Object 



121
 
# File 'lib/wurk/client.rb', line 121

def push_bulk(items)  = new.push_bulk(items)

.reliable_push! ⇒ Object

Activate reliable_push! mode globally. Idempotent — call from the top level of an initializer (NOT inside Wurk.configure_*). Spec: docs/target/sidekiq-pro.md §5.



502
503
504
505
 
# File 'lib/wurk/client/buffered.rb', line 502

def reliable_push! # rubocop:disable Naming/PredicateMethod
  Buffered.install!
  true
end

.reliable_push? ⇒ Boolean

Returns:

  • (Boolean) 


507
508
509
 
# File 'lib/wurk/client/buffered.rb', line 507

def reliable_push?
  Buffered.installed?
end

.reliable_push_buffer ⇒ Object 



511
512
513
 
# File 'lib/wurk/client/buffered.rb', line 511

def reliable_push_buffer
  Buffered.buffer_cap
end

.reliable_push_buffer=(value) ⇒ Object 



515
516
517
 
# File 'lib/wurk/client/buffered.rb', line 515

def reliable_push_buffer=(value)
  Buffered.buffer_cap = value
end

.reliable_push_drainer(interval: Buffered::Drainer::DEFAULT_INTERVAL) ⇒ Object

Start an opt-in background drainer thread. Implicitly enables reliable_push! so callers don't have to chain the two. Idempotent; calling again replaces the thread with one at the new interval. Spec for reliable_push (sidekiq-pro.md §5) only requires drain on next push — this is a Wurk extension for issue #19's "Background drain thread flushes on reconnect" so producer-stopped-mid-outage buffers don't sit idle until next push.



534
535
536
537
538
 
# File 'lib/wurk/client/buffered.rb', line 534

def reliable_push_drainer(interval: Buffered::Drainer::DEFAULT_INTERVAL)
  Buffered.install!
  Buffered.start_drainer!(interval: interval)
  true
end

.reliable_push_drainer_running? ⇒ Boolean

Returns:

  • (Boolean) 


544
545
546
 
# File 'lib/wurk/client/buffered.rb', line 544

def reliable_push_drainer_running?
  Buffered.drainer_running?
end

.reliable_push_drainer_stop! ⇒ Object 



540
541
542
 
# File 'lib/wurk/client/buffered.rb', line 540

def reliable_push_drainer_stop!
  Buffered.stop_drainer!
end

.reliable_push_overflow ⇒ Object 



519
520
521
 
# File 'lib/wurk/client/buffered.rb', line 519

def reliable_push_overflow
  Buffered.overflow_mode
end

.reliable_push_overflow=(mode) ⇒ Object 



523
524
525
 
# File 'lib/wurk/client/buffered.rb', line 523

def reliable_push_overflow=(mode)
  Buffered.overflow_mode = mode
end

.via(pool) ⇒ Object

Thread-local pool override. Re-entrant calls are rejected — Sidekiq raises here too, because nested via would silently shadow. The begin/ensure guards the slot so a raise on entry doesn't clear the outer caller's pool.

Raises:

  • (ArgumentError) 


140
141
142
143
144
145
146
147
148
149
150
 
# File 'lib/wurk/client.rb', line 140

def via(pool)
  raise ArgumentError, 'pool is required' if pool.nil?
  raise 'Wurk::Client.via is not re-entrant' if Thread.current[:wurk_via_pool]

  Thread.current[:wurk_via_pool] = pool
  begin
    yield
  ensure
    Thread.current[:wurk_via_pool] = nil
  end
end

Instance Method Details

#cancel!(jid) ⇒ Object

Marks an IterableJob as cancelled. Returns the Unix epoch timestamp written. Field name + epoch-second value mirror Sidekiq::IterableJob#cancel! exactly. TTL = CANCELLATION_PERIOD so other workers observe the flag well after the dashboard click that issued the cancel.

Raises:

  • (ArgumentError) 


99
100
101
102
103
104
105
106
107
108
 
# File 'lib/wurk/client.rb', line 99

def cancel!(jid)
  raise ArgumentError, 'jid must be a non-empty String' if jid.nil? || jid.to_s.empty?

  ts = ::Process.clock_gettime(::Process::CLOCK_REALTIME).to_i
  pool.with do |conn|
    conn.call('HSET', "it-#{jid}", 'cancelled', ts)
    conn.call('EXPIRE', "it-#{jid}", Wurk::IterableJob::CANCELLATION_PERIOD)
  end
  ts
end

#flush_batched(payloads) ⇒ Object

Flush batched payloads (each carrying a bid) to Redis in one pipeline. Public entry point for Wurk::Batch's autoflush buffer — see #push_batched for the per-job BATCH_PUSH semantics it reuses.



113
114
115
116
117
 
# File 'lib/wurk/client.rb', line 113

def flush_batched(payloads)
  return if payloads.empty?

  pool.with { |conn| push_batched_pipelined(conn, payloads, now_in_millis) }
end

#middleware {|copy| ... } ⇒ Object

Returns the chain (or a duplicate when a block is given, matching Sidekiq).

Yields:

  • (copy) 


61
62
63
64
65
66
67
 
# File 'lib/wurk/client.rb', line 61

def middleware
  return @chain unless block_given?

  copy = @chain.dup
  yield copy
  copy
end

#push(item) ⇒ String?

Returns jid; nil when client middleware halts the push.

Parameters:

  • item (Hash) —

    job payload; must carry class and args, may carry at, queue, jid, etc.

Returns:

  • (String, nil) —

    jid; nil when client middleware halts the push.



71
72
73
74
75
76
77
78
79
80
 
# File 'lib/wurk/client.rb', line 71

def push(item)
  normed  = normalize_item(item)
  payload = invoke_chain(normed)
  return nil unless payload

  verify_json(payload)
  buffered = raw_push([payload])
  emit_enqueued([payload], buffered)
  payload['jid']
end

#push_bulk(items) ⇒ Array<String, nil>

Returns jids in submission order; nil entries mark middleware-halted jobs.

Parameters:

  • items (Hash) —

    keys: class, args (Array), at?, spread_interval?, batch_size?, jid?

Returns:

  • (Array<String, nil>) —

    jids in submission order; nil entries mark middleware-halted jobs.



84
85
86
87
88
89
90
91
92
93
 
# File 'lib/wurk/client.rb', line 84

def push_bulk(items)
  args = items['args'] || items[:args]
  validate_bulk_shape!(items, args)
  return [] if args.empty?

  at_values = expand_at(items, args.size)
  batch_sz  = items['batch_size'] || items[:batch_size] || (at_values ? SCHEDULED_BATCH_SIZE : DEFAULT_BATCH_SIZE)
  base      = bulk_base(items)
  flush_bulk(args, at_values, base, batch_sz)
end