Class: Pgbus::Client::ResizablePool

Inherits:
Object
  • Object
show all
Defined in:
lib/pgbus/client/resizable_pool.rb

Overview

Owns the current streams PGMQ::Client behind a Concurrent::AtomicReference and resizes it by BUILDING A NEW pool, atomically swapping the reference, then bounded-draining + shutting down the old one (issue #323 spike).

connection_pool 2.x cannot resize in place (@size is frozen at construction; only shutdown/reload/available exist), so growing REQUIRES a new PGMQ::Client + a reference swap.

DRAIN CORRECTNESS — the load-bearing detail. We do NOT trust connection_pool's available == size as "drained": pgmq-ruby's #with_connection retries a lost connection (checkout → checkin → checkout again), so between the two checkouts the connection sits idle in the queue and available momentarily equals size while a produce/read is still mid-flight. Closing the pool in that window makes the retry's checkout raise PoolShuttingDownError → a genuinely LOST durable broadcast (the PgBouncer/failover mid-INSERT case). Instead each reader op is bracketed by an in-flight AtomicFixnum on the CAPTURED pool, and drain waits until THAT hits zero. This is immune to the internal retry and independent of the connection_pool version.

DURABLE BROADCASTS are loss-safe by construction: #produce captures the Pool once, runs its whole retrying checkout→INSERT→COMMIT→checkin on that pool, then decrements. Both old and new pools connect to the same Postgres, so any INSERT commits; and close waits for the captured pool's inflight to reach zero, so a pool is never shut while a produce is in flight on it.

Defined Under Namespace

Classes: Pool, SwapStats

Constant Summary collapse

DRAIN_POLL =

10ms

0.01

Instance Method Summary collapse

Constructor Details

#initialize(pgmq, shared:, drain_timeout:, logger:, clock: ::Process) ⇒ ResizablePool

shared: is accepted for symmetry with the Client's two connection paths but is not stored — the shared-AR alias guard lives in #close_current via the job_pool: identity check, and #swap is simply never called on the shared path (Client#resize_streams_pool short-circuits there).



45
46
47
48
49
50
51
52
53
 
# File 'lib/pgbus/client/resizable_pool.rb', line 45

def initialize(pgmq, shared:, drain_timeout:, logger:, clock: ::Process) # rubocop:disable Lint/UnusedMethodArgument
  @ref = Concurrent::AtomicReference.new(new_pool(pgmq))
  @drain_timeout = drain_timeout
  @logger = logger
  @clock = clock
  @swap_mutex = Mutex.new # serializes swap AND close_current; reads never take it
  @swap_count = 0
  @last = nil
end

Instance Method Details

#close_current(job_pool:) ⇒ Object

Teardown of the current pool (from Client#close), race-free vs swap. Skips the close when the current pool aliases the job pool (shared-AR path).



97
98
99
100
101
102
 
# File 'lib/pgbus/client/resizable_pool.rb', line 97

def close_current(job_pool:)
  @swap_mutex.synchronize do
    pool = @ref.get
    pool.pgmq.close unless pool.pgmq.equal?(job_pool)
  end
end

#current ⇒ Object

Hot path: one volatile load, no lock.



56
 
# File 'lib/pgbus/client/resizable_pool.rb', line 56

def current = @ref.get

#produce ⇒ Object

Bracketed reader ops — inflight decrements only after the WHOLE retrying call returns (so drain can't close under an in-flight produce/read).



60
61
62
63
64
65
66
67
68
 
# File 'lib/pgbus/client/resizable_pool.rb', line 60

def produce(*, **)
  pool = @ref.get
  pool.inflight.increment
  begin
    pool.pgmq.produce(*, **)
  ensure
    pool.inflight.decrement
  end
end

#stats ⇒ Object 



80
 
# File 'lib/pgbus/client/resizable_pool.rb', line 80

def stats = @ref.get.pgmq.stats

#stats_snapshot ⇒ Object 



104
105
106
107
 
# File 'lib/pgbus/client/resizable_pool.rb', line 104

def stats_snapshot
  @last || SwapStats.new(swap_count: 0, last_drain_seconds: 0.0, last_conns_closed: 0,
                         last_from_size: nil, last_to_size: nil, last_drained: nil)
end

#swap(new_pgmq, from_size:, to_size:) ⇒ Object

Swap in a freshly-built PGMQ::Client, then drain + close the old Pool.



83
84
85
86
87
88
89
90
91
92
93
 
# File 'lib/pgbus/client/resizable_pool.rb', line 83

def swap(new_pgmq, from_size:, to_size:)
  @swap_mutex.synchronize do
    old = @ref.get
    @ref.set(new_pool(new_pgmq)) # new checkouts now resolve to the new pool
    drained, secs, closed = drain_and_close(old)
    @swap_count += 1
    @last = SwapStats.new(swap_count: @swap_count, last_drain_seconds: secs,
                          last_conns_closed: closed, last_from_size: from_size,
                          last_to_size: to_size, last_drained: drained)
  end
end

#with_connection ⇒ Object 



70
71
72
73
74
75
76
77
78
 
# File 'lib/pgbus/client/resizable_pool.rb', line 70

def with_connection(&)
  pool = @ref.get
  pool.inflight.increment
  begin
    pool.pgmq.with_connection(&)
  ensure
    pool.inflight.decrement
  end
end