Class: Wurk::RedisPool

Inherits:
Object
  • Object
show all
Defined in:
lib/wurk/redis_pool.rb

Overview

Per-process pool over redis-client + connection_pool. Never share a socket across forks: the parent closes the pool before fork, each child opens a fresh one (see docs/idea/03-process-model.md, steps 3 and 5).

#with absorbs transient Redis failures (production incident #101), but only where replaying the caller's block cannot change what the server already did — the block is arbitrary Ruby, so a replay re-issues every command in it:

* READONLY / NOREPLICAS / UNBLOCKED — a failover happened; close and retry
once immediately so redis-client redials the new primary (spec §26).
* CannotConnect / Failover — raised while dialing, so the command that hit
one never reached a server and cannot have applied; close and retry with
exponential backoff up to CONN_MAX_ATTEMPTS, then raise.
* Read-/WriteTimeout and bare ConnectionError — the command may already
have applied server-side, so the pool does not replay the block. That
bounds block replay only: by the time one of these reaches the pool,
redis-client has already re-sent the in-flight command once on a fresh
socket (reconnect_attempts, see DEFAULT_RECONNECT_ATTEMPTS), so a lost
reply can still double-apply that one command. Blocks that are safe to
re-run (pure reads, an LMOVE the reaper reclaims, owner-CAS scripts) opt
back into the backoff with `with(idempotent: true)`.
* ConnectionPool::TimeoutError — checkout starved; retry once after a
short jittered pause, then raise (sizing is the fix, not queuing). Only
a checkout of *this* pool that never handed out a connection retries: the
same error escaping a nested checkout of another pool from inside the
block is raised as-is, because the block's earlier commands have run.

Those proofs are about the command that raised, not the block around it: a block is several round trips, and redis-client re-dials mid-block, so a CannotConnect can surface on the second pipeline of a block whose first one already landed. So the pool also watches the connection's round-trip odometer (RedisClientAdapter::CompatClient#round_trips) and refuses to replay a non-idempotent block that has already completed one. Every retry, refused replay, and final give-up is reported through the injected on_error telemetry hook (Wurk::Configuration#on_redis_error).

Constant Summary collapse

DEFAULT_URL =

rubocop:disable Metrics/ClassLength

ENV.fetch('REDIS_URL', 'redis://localhost:6379/0')
DEFAULT_NAME = 
'default'
DEFAULT_POOL_TIMEOUT =

ConnectionPool checkout wait — how long #with blocks for a free slot.

1.0
DEFAULT_CONNECT_TIMEOUT =

Socket-level timeouts handed to RedisClient, split apart from the checkout wait above. read/write are deliberately wider than connect so a briefly- slow-but-alive Redis (RDB fork pause, a large BLMOVE payload) doesn't spuriously ReadTimeout — the production incident (#101) the single dual-use timeout caused.

reconnect_attempts: 1 is Sidekiq's own reconnect_attempts ||= 1. It re-dials a dropped socket once and re-sends the one in-flight command, which is what lets every pooled connection survive a Redis restart, a failover or an idle-socket reap without surfacing an error. The price is that a reply lost mid-command (a ReadTimeout included) can apply that command twice: a duplicate LPUSH or a double INCR. 0 would close that window but turn every stale socket into an error the pool cannot replay for a non-idempotent block, so it stays at parity.

1.0
DEFAULT_READ_TIMEOUT = 
2.5
DEFAULT_WRITE_TIMEOUT = 
2.5
DEFAULT_RECONNECT_ATTEMPTS = 
1
DEFAULT_CLIENT_CONFIG =

The floor every pool starts from; any key the host passed wins over it — including command_builder, so an app that has its own keeps it.

{
  url: DEFAULT_URL,
  connect_timeout: DEFAULT_CONNECT_TIMEOUT,
  read_timeout: DEFAULT_READ_TIMEOUT,
  write_timeout: DEFAULT_WRITE_TIMEOUT,
  reconnect_attempts: DEFAULT_RECONNECT_ATTEMPTS,
  command_builder: CommandBuilder
}.freeze
RETRYABLE_MSG =

Server-side messages where the connection is closed and the block retried exactly once. READONLY is itself a RedisClient::ConnectionError subclass, so this message match must be tested BEFORE the generic ConnectionError backoff below (otherwise a failover would sleep instead of redialing).

/\A(READONLY|NOREPLICAS|UNBLOCKED)/
PRE_APPLY_ERRORS =

ConnectionErrors that can only be raised while dialing, so the block provably never applied and replaying it is safe whatever it contains. CannotConnect covers every connect-phase failure (redis-client converts a stalled handshake into it); Failover is the Sentinel resolver rejecting a server whose role changed. Any other ConnectionError — Read-/WriteTimeout or a reset mid-command — leaves the outcome unknown.

[RedisClient::CannotConnectError, RedisClient::FailoverError].freeze
REPLAY_PLANS =

The two retry_plan verdicts that re-run the block; the rest raise.

%i[failover backoff].freeze
CONN_MAX_ATTEMPTS =

ConnectionError backoff: CONN_MAX_ATTEMPTS total tries, sleeping (BASE * 2**attempt) + rand*JITTER before each retry. The 1.0s + 2.0s pair rides out a sub-4s blip; the jitter de-syncs a fleet reconnecting at once.

3
CONN_BACKOFF_BASE = 
0.5
CONN_BACKOFF_JITTER = 
0.25
POOL_RETRY_MIN =

Checkout-timeout retry: one retry after a jittered pause in [POOL_RETRY_MIN, POOL_RETRY_MIN + POOL_RETRY_SPREAD). No loop — sustained checkout starvation is a sizing bug, not something to queue behind.

0.1
POOL_RETRY_SPREAD = 
0.2
REAP_MAX_INTERVAL =

Idle reaping (:redis_idle_timeout) sweeps at half the timeout, so a connection is closed within 1.5x of going idle, but never less often than this — Sidekiq reaps on its 10s heartbeat.

10.0

Class Attribute Summary collapse

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(size:, name: DEFAULT_NAME, on_error: nil, redis_idle_timeout: nil, **options) ⇒ RedisPool

Takes the standard Sidekiq config.redis hash: pool_timeout tunes the ConnectionPool checkout; connect_timeout/read_timeout/write_timeout/ reconnect_attempts plus any other redis-client key (driver, ssl_params, sentinels, …) reach the client. Sidekiq-only spellings (network_timeout, master_name, logger, …) are translated or dropped by Wurk::RedisOptions; a key redis-client would reject raises there with the key named. on_error is an optional callable fired per retry / final give-up with { error:, attempt:, retried:, pool: }.

redis_idle_timeout (seconds, Wurk::Configuration) closes a checked-in connection once it has sat unused that long; the next checkout dials a fresh one. It is not redis-client's own idle_timeout, which only re-validates a stale socket with a PING and stays a config.redis key.



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

def initialize(size:, name: DEFAULT_NAME, on_error: nil, redis_idle_timeout: nil, **options)
  @size               = size
  @name               = name
  @on_error           = on_error
  @redis_idle_timeout = self.class.validate_idle_timeout(redis_idle_timeout)
  @pool_timeout       = options.fetch(:pool_timeout, DEFAULT_POOL_TIMEOUT)
  @client_config      = build_client_config(options)
  @url                = @client_config[:url]
  @pool               = ConnectionPool.new(size: size, timeout: @pool_timeout) { build_client }
  @reaper_lock        = Thread::Mutex.new
  @reaper             = nil
  @reaper_closed      = false
  Metrics::PoolRegistry.register(self)
end

Class Attribute Details

.guard ⇒ Object

A redis-client middleware every pool built from now on carries, after any the host configured. Set only by the boot of a command that must not write Redis (Preflight::RedisGuard); nil everywhere else.



123
124
125
 
# File 'lib/wurk/redis_pool.rb', line 123

def guard
  @guard
end

Instance Attribute Details

#client_config ⇒ Object (readonly)

Returns the value of attribute client_config.



117
118
119
 
# File 'lib/wurk/redis_pool.rb', line 117

def client_config
  @client_config
end

#name ⇒ Object (readonly)

Returns the value of attribute name.



117
118
119
 
# File 'lib/wurk/redis_pool.rb', line 117

def name
  @name
end

#pool_timeout ⇒ Object (readonly)

Returns the value of attribute pool_timeout.



117
118
119
 
# File 'lib/wurk/redis_pool.rb', line 117

def pool_timeout
  @pool_timeout
end

#redis_idle_timeout ⇒ Object (readonly)

Returns the value of attribute redis_idle_timeout.



117
118
119
 
# File 'lib/wurk/redis_pool.rb', line 117

def redis_idle_timeout
  @redis_idle_timeout
end

#size ⇒ Object (readonly)

Returns the value of attribute size.



117
118
119
 
# File 'lib/wurk/redis_pool.rb', line 117

def size
  @size
end

#url ⇒ Object (readonly)

Returns the value of attribute url.



117
118
119
 
# File 'lib/wurk/redis_pool.rb', line 117

def url
  @url
end

Class Method Details

.validate_idle_timeout(seconds) ⇒ Object

Checked here as well as by connection_pool, whose own ArgumentError would only surface inside the reaper thread and silently end it. Configuration calls it too, so a bad value fails in the parent instead of crash-looping every forked child.

Raises:

  • (ArgumentError) 


227
228
229
230
231
232
 
# File 'lib/wurk/redis_pool.rb', line 227

def self.validate_idle_timeout(seconds)
  return nil if seconds.nil?
  return seconds if seconds.is_a?(Numeric) && seconds.positive?

  raise ArgumentError, "redis_idle_timeout must be a positive number of seconds or nil, got #{seconds.inspect}"
end

Instance Method Details

#available ⇒ Object

Free (unchecked-out) slots right now. Local and cheap — no Redis round trip — so a monitor can poll it without perturbing the pool. A reaped connection's slot stays free, so reaping leaves this count where it was; #idle is the number that drops.



214
215
216
 
# File 'lib/wurk/redis_pool.rb', line 214

def available
  @pool.available
end

#disconnect! ⇒ Object 



190
191
192
193
194
 
# File 'lib/wurk/redis_pool.rb', line 190

def disconnect!
  Metrics::PoolRegistry.deregister(self)
  stop_reaper
  @pool.shutdown { |conn| safe_close(conn) }
end

#idle ⇒ Object

Open connections sitting checked in — what idle reaping closes.



219
220
221
 
# File 'lib/wurk/redis_pool.rb', line 219

def idle
  @pool.idle
end

#info ⇒ Object

Redis INFO parsed to a Hash, with this pool's own size / available slot counts merged in — one call gives a heartbeat both Redis health and local pool saturation. (Real Redis INFO has no size/available field.)



205
206
207
208
 
# File 'lib/wurk/redis_pool.rb', line 205

def info
  with(idempotent: true) { |conn| parse_info(conn.call('INFO')) }
    .merge('size' => @size, 'available' => available)
end

#reload! ⇒ Object

Non-terminal: closes idle sockets before a fork; the pool stays usable, so host threads holding it never see PoolShuttingDownError.



198
199
200
 
# File 'lib/wurk/redis_pool.rb', line 198

def reload!
  @pool.reload { |conn| safe_close(conn) }
end

#with(idempotent: false, &block) ⇒ Object

Checkout a connection and run the block. Our own checkout raises ConnectionPool::TimeoutError before the block runs, so it is caught out here (the in-block #run rescue never sees it) — one retry, then raise. entered tells that apart from the same class escaping the block — a nested checkout of another, starved pool — where the block's earlier commands already ran and a replay would re-issue them (an INCR counted twice, a job pushed twice).

idempotent: true asserts the block can be re-run after a command may already have applied server-side, which buys back the full ConnectionError backoff. Only claim it for pure reads or writes whose repeat is a no-op.



166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
 
# File 'lib/wurk/redis_pool.rb', line 166

def with(idempotent: false, &block)
  ensure_reaper if @redis_idle_timeout
  checkout_retried = false
  entered = false
  begin
    @pool.with do |conn|
      entered = true
      run(conn, idempotent, &block)
    end
  rescue ConnectionPool::TimeoutError => e
    raise if entered

    Metrics::ProcessCounters.increment([:pool_checkout_timeouts, @name])
    if checkout_retried
      notify_error(e, attempt: 2, retried: false)
      raise
    end
    checkout_retried = true
    notify_error(e, attempt: 1, retried: true)
    sleep(checkout_delay)
    retry
  end
end