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.

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

Defined Under Namespace

Modules: ForkHook, InstanceMethods Classes: Drainer, Overflow

Constant Summary collapse

DEFAULT_BUFFER_CAP = 
1_000
DRAINING_KEY = 
:wurk_reliable_push_draining
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_client_factory ⇒ Object

Returns the value of attribute buffer_client_factory.



63
64
65
 
# File 'lib/wurk/client/buffered.rb', line 63

def buffer_client_factory
  @buffer_client_factory
end

Class Method Details

.buffer ⇒ Object

Internal — visible for tests. Treat as private.



289
290
291
 
# File 'lib/wurk/client/buffered.rb', line 289

def buffer
  @buffer ||= []
end

.buffer_cap ⇒ Object 



81
82
83
 
# File 'lib/wurk/client/buffered.rb', line 81

def buffer_cap
  @buffer_cap ||= DEFAULT_BUFFER_CAP
end

.buffer_cap=(value) ⇒ Object 



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

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 



93
94
95
 
# File 'lib/wurk/client/buffered.rb', line 93

def buffer_size
  buffer_mutex.synchronize { buffer.size }
end

.drain!(client) ⇒ Object

Drain payloads through raw_push on the given client. Stops on the first transient failure (ConnectionError past the pool's own retries, or a starved checkout), preserving order at the head of the buffer so the next push retries the same payload. 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.



262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
 
# File 'lib/wurk/client/buffered.rb', line 262

def drain!(client)
  drained = 0
  while (payload = pop_head)
    begin
      replayed = attempt_replay(client, payload)
    rescue StandardError
      # Non-connection failures (OOM, LOADING, READONLY…) must not
      # drop the popped payload — restore it before propagating, or
      # a recovering-but-not-ready Redis silently eats one buffered
      # job per drain tick.
      buffer_mutex.synchronize { buffer.unshift(payload) }
      raise
    end

    unless replayed
      buffer_mutex.synchronize { buffer.unshift(payload) }
      break
    end

    client.send(:emit_enqueued, [payload])
    Wurk::Metrics::Statsd.increment('jobs.recovered.push')
    drained += 1
  end
  drained
end

.drainer_running? ⇒ Boolean

Returns:

  • (Boolean) 


315
316
317
 
# File 'lib/wurk/client/buffered.rb', line 315

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

.enbuffer(payloads, client: nil) ⇒ 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. If client is provided, captures its pool for drainer to use by default.

Raises:



176
177
178
179
180
181
182
183
184
185
186
 
# File 'lib/wurk/client/buffered.rb', line 176

def enbuffer(payloads, client: nil)
  capture_pool_from_client(client)

  cap  = buffer_cap
  mode = overflow_mode
  undelivered = buffer_mutex.synchronize do
    mode == :raise ? append_within_capacity(payloads, cap) : append_dropping_oldest(payloads, cap)
  end

  raise Overflow, undelivered unless undelivered.empty?
end

.install! ⇒ Object

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



68
69
70
71
72
73
74
75
 
# File 'lib/wurk/client/buffered.rb', line 68

def install!
  install_mutex.synchronize do
    return if @installed

    Wurk::Client.prepend(InstanceMethods)
    @installed = true
  end
end

.installed? ⇒ Boolean

Returns:

  • (Boolean) 


77
78
79
 
# File 'lib/wurk/client/buffered.rb', line 77

def installed?
  @installed == true
end

.overflow_mode ⇒ Object 



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

def overflow_mode
  @overflow_mode ||= DEFAULT_OVERFLOW_MODE
end

.overflow_mode=(mode) ⇒ Object 



101
102
103
104
105
106
107
108
109
110
111
112
113
 
# File 'lib/wurk/client/buffered.rb', line 101

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 



115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
 
# File 'lib/wurk/client/buffered.rb', line 115

def reset!
  buffer_mutex.synchronize do
    @buffer = []
    @buffer_cap = nil
    @overflow_mode = nil
    @buffer_client_factory = nil
  end
  # Stop before dropping: an unstopped drainer thread survives with
  # its factory nil'd out from under it and ticks forever against
  # nothing, leaking the thread and everything its closure retains.
  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; the captured client factory goes with it, since it closes over the parent's pre-fork Redis pool.

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



154
155
156
157
158
159
160
161
162
163
164
165
166
 
# File 'lib/wurk/client/buffered.rb', line 154

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_client_factory = nil
  interval               = @drainer&.interval
  @drainer               = nil
  start_drainer!(interval: interval) if interval
  true
end

.start_drainer!(interval: Drainer::DEFAULT_INTERVAL, client_factory: nil) ⇒ 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.



299
300
301
302
303
304
305
306
 
# File 'lib/wurk/client/buffered.rb', line 299

def start_drainer!(interval: Drainer::DEFAULT_INTERVAL, client_factory: nil)
  install_mutex.synchronize do
    @drainer&.stop
    factory = client_factory || buffer_client_factory || -> { Wurk::Client.new }
    @drainer = Drainer.new(interval: interval, client_factory: factory)
    @drainer.start
  end
end

.stop_drainer! ⇒ Object 



308
309
310
311
312
313
 
# File 'lib/wurk/client/buffered.rb', line 308

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

Instance Method Details

#_fork ⇒ Object Originally defined in module ForkHook