Class: Neo4j::Driver::Bolt::Pool

Inherits:
Object
  • Object
show all
Defined in:
lib/neo4j/driver/bolt/pool.rb

Overview

Hardened connection pool. Wraps ConnectionPool::TimedStack and layers the three guarantees production callers care about:

max_connection_lifetime  → connections older than this are
                         dropped on acquire (caps server-side
                         resource growth for long-running
                         processes).
connection_liveness_check_timeout → if a connection has been
                         idle longer than this, RESET-probe
                         before handing it out (catches
                         "server reaped me, client didn't
                         notice yet").
connection_acquisition_timeout → bounded wait when the pool
                         is at its size cap (raises
                         ClientException, not a hang).

Mirrors org.neo4j.driver.internal.async.pool.NettyChannelPool in the Java reference. The acquisition-timeout slice was already in place via TimedStack; this class adds the lifetime + liveness gates and centralises the construction across the two providers.

Constant Summary collapse

DEFAULT_ACQUISITION_TIMEOUT = 
60

Instance Method Summary collapse

Constructor Details

#initialize(size:, options:, connect_factory:, clock: Internal::Clock.new) ⇒ Pool

connect_factory is a callable returning a freshly-opened Bolt::Connection. Wrapped, not inherited, because TimedStack owns the threading + size-cap primitive and doesn't expose hooks for "decide whether to hand out the popped item".



33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
 
# File 'lib/neo4j/driver/bolt/pool.rb', line 33

def initialize(size:, options:, connect_factory:, clock: Internal::Clock.new)
  @options = options
  @clock = clock
  # connect_factory takes the auth token a fresh connection should
  # authenticate with. pop hands it to the create block via a
  # thread-local: the create block runs synchronously inside
  # @stack.pop on the *calling* thread, so a thread-local is read
  # back by the same thread that set it and can't be clobbered by a
  # concurrent pop on another thread (a shared ivar could — this
  # pool is shared across sessions). Keyed by object_id so distinct
  # pools don't collide on the same thread. This lets a session
  # create a fresh connection with its own identity (per-session
  # auth) or the manager's current token in a single get_token call,
  # instead of connecting as the manager and re-authing.
  @connect_factory = connect_factory
  @auth_key = :"bolt_pool_next_auth_#{object_id}"
  # Companion thread-local (same pattern as @auth_key) carrying the
  # acquisition deadline a freshly-built connection's handshake must
  # honour — so one budget spans the home-db guess + fallback.
  @deadline_key = :"bolt_pool_next_deadline_#{object_id}"
  # Metrics counters (driver.metrics / testkit GetConnectionPoolMetrics):
  # @created = live connections this pool has built (idle + in use),
  # @leased = currently checked out. idle = created - leased. Guarded by
  # @metrics_lock since pop/push/discard run on many session threads.
  # @created is bumped inside the create block (only on a successful
  # factory call — TimedStack rolls back its own count if it raises).
  @metrics_lock = Mutex.new
  @created = 0
  @leased = 0
  @stack = ConnectionPool::TimedStack.new(size: size) do
    @connect_factory.call(Thread.current[@auth_key], Thread.current[@deadline_key]).tap { bump(created: 1) }
  end
end

Instance Method Details

#discard(connection) ⇒ Object

Close a checked-out connection without putting it back. Mirrors Java's pool-discard: used when the connection is in a known-bad state (server FAILED, write-failure on a NotALeader, etc.) so we don't poison the pool. Frees the TimedStack slot so the next pop can lazily build a fresh one.



136
137
138
139
140
141
142
 
# File 'lib/neo4j/driver/bolt/pool.rb', line 136

def discard(connection)
  close_quietly(connection)
  @stack.decrement_created
  # A checked-out connection removed for good: no longer created, no
  # longer leased.
  bump(created: -1, leased: -1)
end

#metrics_snapshot ⇒ Object

A consistent [in_use, idle] snapshot read under one lock — both counters move together on discard, so reading them separately could observe an intermediate (e.g. negative idle) state.



70
71
72
 
# File 'lib/neo4j/driver/bolt/pool.rb', line 70

def metrics_snapshot
  @metrics_lock.synchronize { [@leased, @created - @leased] }
end

#pop(auth: nil, deadline: nil) ⇒ Object

Pop a connection that's young enough and confirmed alive. Loops until a usable one is found or the acquisition-timeout budget runs out — every discarded slot opens room for the factory to make a fresh one.



78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
 
# File 'lib/neo4j/driver/bolt/pool.rb', line 78

def pop(auth: nil, deadline: nil)
  # The token a freshly-built connection should authenticate with,
  # handed to the create block via a thread-local that lives only
  # for this pop (see #initialize). Cleared in `ensure` so it never
  # leaks into a later create on the same thread.
  Thread.current[@auth_key] = auth
  # A caller-supplied deadline lets one acquisition-timeout budget span
  # several pops (the home-db optimistic acquire + its fallback), and
  # bounds a freshly-built connection's handshake via the create block.
  deadline ||= current_monotonic + acquisition_timeout
  Thread.current[@deadline_key] = deadline
  loop do
    timeout = [deadline - current_monotonic, 0].max
    conn = @stack.pop(timeout: timeout)
    return prepare(conn) if usable?(conn)

    discard_on_pop(conn)
    # Loop with whatever budget remains. A flapping server
    # would otherwise burn the whole pool in one acquire.
  end
rescue ::Timeout::Error
  raise Exceptions::ClientException,
        "Unable to acquire connection from the pool within configured maximum time of #{format_acquisition_timeout}"
ensure
  Thread.current[@auth_key] = nil
  Thread.current[@deadline_key] = nil
end

#push(connection) ⇒ Object 



106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
 
# File 'lib/neo4j/driver/bolt/pool.rb', line 106

def push(connection)
  return unless connection

  # RESET on return to the pool, clearing any residual server-side state
  # so the next lessee gets a clean connection. Reference drivers do this
  # unless they advertise Optimization:MinimalResets (MRI doesn't), and
  # testkit's scripts encode the expectation (#RESET_ON_POOL_RETURN#).
  # It also keeps the liveness probe distinct from the return-RESET, so a
  # silently-dead connection is actually detected on re-acquire
  # (test_should_drop_connections_failing_liveness_check).
  #
  # If that RESET fails the connection is dead — discard it (closing the
  # socket) rather than pooling a broken connection (test_fail_on_reset).
  begin
    connection.reset!(propagate: true)
  rescue StandardError
    discard(connection)
    return
  end
  connection.idle_since = current_monotonic
  @stack.push(connection)
  bump(leased: -1) # returned to the pool, now idle
end

#shutdown(&block) ⇒ Object 



144
145
146
147
148
149
 
# File 'lib/neo4j/driver/bolt/pool.rb', line 144

def shutdown(&block)
  @stack.shutdown(&block)
  # Every connection is closed now — keep the metrics truthful so a
  # post-close snapshot reports nothing live.
  @metrics_lock.synchronize { @created = @leased = 0 }
end