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):

* READONLY / NOREPLICAS / UNBLOCKED — a failover happened; close and retry
once immediately so redis-client redials the new primary (spec §26).
* ConnectionError (incl. CannotConnect / Read- / WriteTimeout) — a blip;
close and retry with exponential backoff up to CONN_MAX_ATTEMPTS, then
raise. At-least-once tolerates the rare duplicate and JobRetry re-runs
the job anyway, so retrying at the pool layer is strictly better.
* ConnectionPool::TimeoutError — checkout starved; retry once after a
short jittered pause, then raise (sizing is the fix, not queuing).

Every retry 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.

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)/
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: }.



79
80
81
82
83
84
85
86
87
 
# File 'lib/wurk/redis_pool.rb', line 79

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.



69
70
71
 
# File 'lib/wurk/redis_pool.rb', line 69

def client_config
  @client_config
end

#name ⇒ Object (readonly)

Returns the value of attribute name.



69
70
71
 
# File 'lib/wurk/redis_pool.rb', line 69

def name
  @name
end

#pool_timeout ⇒ Object (readonly)

Returns the value of attribute pool_timeout.



69
70
71
 
# File 'lib/wurk/redis_pool.rb', line 69

def pool_timeout
  @pool_timeout
end

#size ⇒ Object (readonly)

Returns the value of attribute size.



69
70
71
 
# File 'lib/wurk/redis_pool.rb', line 69

def size
  @size
end

#url ⇒ Object (readonly)

Returns the value of attribute url.



69
70
71
 
# File 'lib/wurk/redis_pool.rb', line 69

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.



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

def available
  @pool.available
end

#disconnect! ⇒ Object 



108
109
110
 
# File 'lib/wurk/redis_pool.rb', line 108

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.)



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

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

#with(&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.



92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
 
# File 'lib/wurk/redis_pool.rb', line 92

def with(&block)
  checkout_retried = false
  begin
    @pool.with { |conn| run(conn, &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