Module: Wurk::Client::Buffered

Includes:
ForkHook
Defined in:
lib/wurk/client/buffered.rb

Overview

Pro feature parity: in-process ring buffer that catches enqueue failures during a Redis outage and replays them on the next push. Activated globally — Wurk::Client.reliable_push!. Buffer is per-process, in-memory only; crash = lost. Does NOT cover batch creation or batch-context pushes (bid on payload): BATCH_PUSH has atomic counter side-effects we can't safely replay.

A failed push opens a circuit for that Redis: later pushes buffer without touching the socket, so one caller pays the connection backoff and the rest do not. The Drainer thread, started on the first failure if none is running, is what probes the server and closes the circuit.

One buffer serves every pool, so each entry remembers the Redis it was headed for (its Origin) and is only ever replayed there: a sharded app pushing through Client.via(shard_a) must not have its outage backlog land on shard B because B's producer happened to push next.

Spec: docs/target/sidekiq-pro.md §5.

Defined Under Namespace

Modules: ForkHook, InstanceMethods Classes: Circuit, Drainer, DropLog, Entry, Origin, Overflow

Constant Summary collapse

DEFAULT_BUFFER_CAP =

rubocop:disable Metrics/ModuleLength

1_000
DRAINING_KEY = 
:wurk_reliable_push_draining
UNBUFFERED_KEY = 
:wurk_reliable_push_unbuffered
REPLAY_ACKED_KEY =

What a failed replay's raw_push had already landed, handed from the prepended raw_push to Buffered.attempt_replay.

:wurk_reliable_push_replay_acked
REPLAY_SLICE =

Payloads per replayed raw_push. A backlog reaches Redis in about N / REPLAY_SLICE calls, each one pipeline per queue.

100
EXIT_FLUSH_SECONDS =

A clean exit tries to deliver what is still in memory, but never holds the process up for longer than this: a hung Redis would otherwise stall shutdown.

5
FLUSH_POLL =

Paces flush's retries as well as its checks, so a server that keeps refusing gets a few attempts in that window and not hundreds of reports.

0.25
OVERFLOW_MODES =

Overflow modes. :drop_oldest is the spec default (Sidekiq Pro §5 ring buffer). :raise lets callers decide what to do on backpressure — Wurk extension surfaced for issue #19's "over-cap pushes raise so callers can decide" requirement.

%i[drop_oldest raise].freeze
DEFAULT_OVERFLOW_MODE = 
:drop_oldest
NOTHING_UNDELIVERED =

Returned by the append helpers when everything fit.

[].freeze

Class Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Class Attribute Details

.buffer_bytes_cap ⇒ Object

Optional second cap, in bytes of payload JSON (Wurk-only; nil, the default, leaves only the entry cap). The entry cap alone lets a burst of large payloads hold cap x payload size in a web process for the whole outage. Over either cap the overflow mode applies; a payload larger than this cap on its own never fits, so it is the one dropped (or raised).



213
214
215
 
# File 'lib/wurk/client/buffered.rb', line 213

def buffer_bytes_cap
  @buffer_bytes_cap
end

.circuit ⇒ Object (readonly)

Returns the value of attribute circuit.



470
471
472
 
# File 'lib/wurk/client/buffered.rb', line 470

def circuit
  @circuit
end

.drops ⇒ Object (readonly)

Returns the value of attribute drops.



470
471
472
 
# File 'lib/wurk/client/buffered.rb', line 470

def drops
  @drops
end

Class Method Details

.buffer ⇒ Object

Internal — visible for tests. Treat as private. Holds Entry structs.



466
467
468
 
# File 'lib/wurk/client/buffered.rb', line 466

def buffer
  @buffer ||= []
end

.buffer_bytes ⇒ Object 



227
228
229
 
# File 'lib/wurk/client/buffered.rb', line 227

def buffer_bytes
  buffer_mutex.synchronize { @buffer_bytes }
end

.buffer_cap ⇒ Object 



195
196
197
 
# File 'lib/wurk/client/buffered.rb', line 195

def buffer_cap
  @buffer_cap ||= DEFAULT_BUFFER_CAP
end

.buffer_cap=(value) ⇒ Object 



199
200
201
202
203
204
205
 
# File 'lib/wurk/client/buffered.rb', line 199

def buffer_cap=(value)
  unless value.is_a?(Integer) && value.positive?
    raise ArgumentError, 'reliable_push_buffer must be a positive Integer'
  end

  @buffer_cap = value
end

.buffer_size ⇒ Object 



223
224
225
 
# File 'lib/wurk/client/buffered.rb', line 223

def buffer_size
  buffer_mutex.synchronize { buffer.size }
end

.drain!(client, probe: false) ⇒ Object

Replay, through raw_push on client, the buffered payloads headed for the Redis that client writes to — oldest first; entries for any other Redis are left in place, so a push to a healthy shard neither misroutes nor waits on a dead one's backlog. Stops on the first transient failure (ConnectionError past the pool's own retries, or a starved checkout) and puts that payload back at the head, so the next push retries it first. Emits statsd jobs.recovered.push per drained payload, plus the jobs.enqueued the buffering push deliberately did not emit — the replay is where the job actually reaches Redis, so a buffered-then-drained job counts once as enqueued and once as recovered.

Skipped outright while the circuit is open, unless this is the probe.



405
406
407
408
409
410
411
412
413
 
# File 'lib/wurk/client/buffered.rb', line 405

def drain!(client, probe: false)
  target = client.send(:pool)
  return 0 if !probe && circuit.open?(target)
  return 0 if buffer_mutex.synchronize { buffer.empty? }

  drained = replay_all(client, target)
  drops.end_burst if drained.positive?
  drained
end

.drain_all! ⇒ Object

The background drainer's pass: every origin in the buffer, each through a client of its own, so one Redis still down does not hold back another that has recovered. Returns the total replayed; the first non-transient error is re-raised after every origin had its turn.



450
451
452
453
454
455
456
457
458
459
460
461
462
463
 
# File 'lib/wurk/client/buffered.rb', line 450

def drain_all!
  failure = nil
  pools = buffered_pools
  drained = pools.sum do |pool|
    drain!(Wurk::Client.new(pool: pool), probe: true)
  rescue StandardError => e
    failure ||= e
    0
  end
  circuit.ping_all(except: pools)
  raise failure if failure

  drained
end

.drainer_running? ⇒ Boolean

Returns:

  • (Boolean) 


494
495
496
 
# File 'lib/wurk/client/buffered.rb', line 494

def drainer_running?
  install_mutex.synchronize { @drainer&.running? == true }
end

.enbuffer(payloads, client:) ⇒ Object

Append payloads to the buffer. Behavior on cap exhaustion depends on overflow_mode:

* :drop_oldest (default, spec) — ring buffer, oldest evicted.
* :raise                       — fills the remaining capacity, then
                               raises one Overflow carrying every
                               payload that did not fit.

Drops batched payloads — caller is expected to re-raise for those. client is the one whose push failed; every payload is tagged with the Redis that client was writing to (see Origin).



317
318
319
320
321
322
323
324
325
326
327
328
 
# File 'lib/wurk/client/buffered.rb', line 317

def enbuffer(payloads, client:)
  entries = entries_for(payloads, client)
  caps = [buffer_cap, buffer_bytes_cap]
  mode = overflow_mode
  undelivered = buffer_mutex.synchronize do
    mode == :raise ? append_within_capacity(entries, *caps) : append_dropping_oldest(entries, *caps)
  end

  raise Overflow, undelivered.map(&:payload) unless undelivered.empty?
ensure
  drops.report(*caps)
end

.flush(timeout: EXIT_FLUSH_SECONDS) ⇒ Object

Replays whatever is still buffered, whichever circuit it is behind, for up to timeout. Returns how many jobs are left. Runs on a clean exit; a Launcher or host shutting down earlier can call it too. Forked children inherit the at_exit hook, but start with an empty buffer of their own.



434
435
436
437
438
439
 
# File 'lib/wurk/client/buffered.rb', line 434

def flush(timeout: EXIT_FLUSH_SECONDS)
  return 0 if undelivered_count.zero?

  retry_until_delivered(flush_clock + timeout)
  undelivered_count.tap { |left| report_undelivered(left) if left.positive? }
end

.install! ⇒ Object

Idempotent. Prepends the wrapper module into Wurk::Client so push / push_bulk / push_prepared drain the buffer before each call and raw_push catches connection errors. Safe to call from multiple threads.



181
182
183
184
185
186
187
188
189
 
# File 'lib/wurk/client/buffered.rb', line 181

def install!
  install_mutex.synchronize do
    return if @installed

    Wurk::Client.prepend(InstanceMethods)
    @installed = true
    at_exit { Buffered.flush }
  end
end

.installed? ⇒ Boolean

Returns:

  • (Boolean) 


191
192
193
 
# File 'lib/wurk/client/buffered.rb', line 191

def installed?
  @installed == true
end

.overflow_mode ⇒ Object 



231
232
233
 
# File 'lib/wurk/client/buffered.rb', line 231

def overflow_mode
  @overflow_mode ||= DEFAULT_OVERFLOW_MODE
end

.overflow_mode=(mode) ⇒ Object 



235
236
237
238
239
240
241
242
243
244
245
246
247
 
# File 'lib/wurk/client/buffered.rb', line 235

def overflow_mode=(mode)
  begin
    mode = mode.to_sym
  rescue NoMethodError, TypeError
    raise ArgumentError, "overflow_mode must be one of #{OVERFLOW_MODES.inspect}"
  end

  unless OVERFLOW_MODES.include?(mode)
    raise ArgumentError, "overflow_mode must be one of #{OVERFLOW_MODES.inspect}"
  end

  @overflow_mode = mode
end

.reset! ⇒ Object 



249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
 
# File 'lib/wurk/client/buffered.rb', line 249

def reset!
  buffer_mutex.synchronize do
    @buffer = []
    @buffer_bytes = 0
    @buffer_cap = nil
    @buffer_bytes_cap = nil
    @overflow_mode = nil
    @replaying = 0
  end
  @circuit = Circuit.new
  @drops = DropLog.new
  # Stop before dropping: an unstopped drainer thread would otherwise
  # tick on forever, unreachable, leaking the thread.
  install_mutex.synchronize do
    @drainer&.stop
    @drainer = nil
  end
end

.reset_after_fork! ⇒ Object

Fork hook, called from the Process._fork prepend below and from Swarm::ChildBoot#reconnect_after_fork. Whichever runs first wins and returns true; the pid guard makes the other a no-op returning false, so a caller can tell which one rebuilt the state.

A child inherits a copy of every ivar here: the buffered payloads, the Drainer (whose thread did not survive the fork), and both mutexes (see their definition for why replacing them matters).

The child DROPS its inherited payloads rather than replaying them: the parent still holds the same buffer and replays it on its own next push, so a child that also drained would enqueue every buffered job once per fork — (children + 1) x N duplicates. Only the parent replays.

@drainer is dropped, never stopped — its @lock carries the same inherited-mutex hazard. A parent-configured drainer is replaced by an equivalent fresh one so an opted-in child keeps flushing the buffer it fills itself.

Deliberately unsynchronized: the child has exactly one thread here, and waiting on the very mutex being replaced is what would hang it.



290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
 
# File 'lib/wurk/client/buffered.rb', line 290

def reset_after_fork! # rubocop:disable Naming/PredicateMethod
  return false if @owner_pid == ::Process.pid

  @owner_pid             = ::Process.pid
  @install_mutex         = Mutex.new
  @buffer_mutex          = Mutex.new
  @buffer                = []
  @buffer_bytes          = 0
  @replaying             = 0
  @circuit               = Circuit.new
  @drops                 = DropLog.new
  interval               = @drainer&.interval unless @implicit_drainer
  @implicit_drainer      = false
  @drainer               = nil
  start_drainer!(interval: interval) if interval
  true
end

.start_drainer!(interval: Drainer::DEFAULT_INTERVAL) ⇒ Object

Start a background drain thread that wakes every interval seconds and tries to flush the buffer. Idempotent — replaces any prior drainer with one at the new interval. Issue #19 requirement: "Background drain thread flushes on reconnect" — handles the case where push activity stops mid-outage so the passive (drain-on-next-push) path never fires.



478
479
480
481
482
483
484
485
 
# File 'lib/wurk/client/buffered.rb', line 478

def start_drainer!(interval: Drainer::DEFAULT_INTERVAL)
  install_mutex.synchronize do
    @drainer&.stop
    @implicit_drainer = false
    @drainer = Drainer.new(interval: interval)
    @drainer.start
  end
end

.stop_drainer! ⇒ Object 



487
488
489
490
491
492
 
# File 'lib/wurk/client/buffered.rb', line 487

def stop_drainer!
  install_mutex.synchronize do
    @drainer&.stop
    @drainer = nil
  end
end

.trip!(pool) ⇒ Object 



441
442
443
444
 
# File 'lib/wurk/client/buffered.rb', line 441

def trip!(pool)
  circuit.trip(pool)
  ensure_drainer
end

.unbuffered ⇒ Object

Pushes inside the block reach Redis or raise. For a caller that goes on to delete its own copy of the job once push returns, which must not count memory as delivered: a crash would then lose the only copy.



418
419
420
421
422
423
424
 
# File 'lib/wurk/client/buffered.rb', line 418

def unbuffered
  previous = Thread.current[UNBUFFERED_KEY]
  Thread.current[UNBUFFERED_KEY] = true
  yield
ensure
  Thread.current[UNBUFFERED_KEY] = previous
end

.unbuffered? ⇒ Boolean

Returns:

  • (Boolean) 


426
427
428
 
# File 'lib/wurk/client/buffered.rb', line 426

def unbuffered?
  Thread.current[UNBUFFERED_KEY] == true
end

Instance Method Details

#_fork ⇒ Object Originally defined in module ForkHook