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 these raise. Replaying would double-push a
job, double-count a stat, or drop a member ZPOPed by the lost reply.
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).

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 = 
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 re-dials a dropped socket once; note that redis-client's re-dial also re-sends the one in-flight command, so the apply-safety split below bounds block replay, not command replay.

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.

{
  url: DEFAULT_URL,
  connect_timeout: DEFAULT_CONNECT_TIMEOUT,
  read_timeout: DEFAULT_READ_TIMEOUT,
  write_timeout: DEFAULT_WRITE_TIMEOUT,
  reconnect_attempts: DEFAULT_RECONNECT_ATTEMPTS
}.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

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(size:, name: DEFAULT_NAME, on_error: 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: }.



104
105
106
107
108
109
110
111
112
 
# File 'lib/wurk/redis_pool.rb', line 104

def initialize(size:, name: DEFAULT_NAME, on_error: nil, **options)
  @size          = size
  @name          = name
  @on_error      = on_error
  @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 }
end

Instance Attribute Details

#client_config ⇒ Object (readonly)

Returns the value of attribute client_config.



94
95
96
 
# File 'lib/wurk/redis_pool.rb', line 94

def client_config
  @client_config
end

#name ⇒ Object (readonly)

Returns the value of attribute name.



94
95
96
 
# File 'lib/wurk/redis_pool.rb', line 94

def name
  @name
end

#pool_timeout ⇒ Object (readonly)

Returns the value of attribute pool_timeout.



94
95
96
 
# File 'lib/wurk/redis_pool.rb', line 94

def pool_timeout
  @pool_timeout
end

#size ⇒ Object (readonly)

Returns the value of attribute size.



94
95
96
 
# File 'lib/wurk/redis_pool.rb', line 94

def size
  @size
end

#url ⇒ Object (readonly)

Returns the value of attribute url.



94
95
96
 
# File 'lib/wurk/redis_pool.rb', line 94

def url
  @url
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.



151
152
153
 
# File 'lib/wurk/redis_pool.rb', line 151

def available
  @pool.available
end

#disconnect! ⇒ Object 



137
138
139
 
# File 'lib/wurk/redis_pool.rb', line 137

def disconnect!
  @pool.shutdown { |conn| safe_close(conn) }
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.)



144
145
146
147
 
# File 'lib/wurk/redis_pool.rb', line 144

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

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

Checkout a connection and run the block. ConnectionPool::TimeoutError is raised by @pool.with before the block runs, so it is caught out here (the in-block #run rescue never sees it) — one retry, then raise.

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.



121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
 
# File 'lib/wurk/redis_pool.rb', line 121

def with(idempotent: false, &block)
  checkout_retried = false
  begin
    @pool.with { |conn| run(conn, idempotent, &block) }
  rescue ConnectionPool::TimeoutError => e
    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