Class: Neo4j::Driver::Direct::ConnectionProvider

Inherits:
Object
  • Object
show all
Defined in:
lib/neo4j/driver/direct/connection_provider.rb

Overview

Single-server connection provider for bolt:// URIs.

Sibling of Routing::LoadBalancer — both implement the same interface (acquire / release / verify_connectivity / supports_multi_db? / close) so Driver can hold either polymorphically without branching on scheme.

Direct ignores the access_mode, database, and bookmarks kwargs to acquire: all sessions hit the same server, so role/database/bookmark-aware routing is not its concern. The kwargs are accepted so call sites stay polymorphic with Routing::LoadBalancer.

Instance Method Summary collapse

Constructor Details

#initialize(uri, auth_manager, options = {}, domain_name_resolver: nil, clock: Internal::Clock.new) ⇒ ConnectionProvider

domain_name_resolver is the factory-injected hostname->IPs hook (nil = system DNS); baked into every connection this provider builds.



20
21
22
23
24
25
26
27
28
29
30
31
32
33
 
# File 'lib/neo4j/driver/direct/connection_provider.rb', line 20

def initialize(uri, auth_manager, options = {}, domain_name_resolver: nil, clock: Internal::Clock.new)
  @uri = uri
  @auth_manager = auth_manager
  @options = options
  @clock = clock
  @domain_name_resolver = domain_name_resolver
  # Authorization-expired generation counter (see Connection#auth_epoch).
  # Bumped when any connection reports AuthorizationExpired so every
  # other pooled connection re-authenticates on its next acquire.
  # Guarded by a mutex: the pool is shared across threads and on JRuby
  # (mri-on-jruby) there's no GIL, so a bare `+= 1` could lose bumps.
  @auth_epoch = 0
  @auth_epoch_mutex = Mutex.new
end

Instance Method Details

#acquire(access_mode: nil, database: nil, bookmarks: nil, imp_user: nil, auth: nil, deadline: nil) ⇒ Object

imp_user is accepted for signature parity with Routing::LoadBalancer#acquire but unused here: the direct path has no discovery, so impersonation is enforced on RUN/BEGIN by the protocol handler (Bolt::Protocol::Base#enforce_impersonation_support!).

Raises:



77
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
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
 
# File 'lib/neo4j/driver/direct/connection_provider.rb', line 77

def acquire(access_mode: nil, database: nil, bookmarks: nil, imp_user: nil, auth: nil, deadline: nil)
  # See Routing::LoadBalancer#acquire for the rationale.
  raise Exceptions::IllegalStateException, 'Driver is closed' if @closed

  # `auth` is the per-session override (nil = use the manager's
  # current token). `effective` is the identity this acquire should
  # hand out; resolving it consults the manager (get_auth_count == 1
  # for the default identity, 0 when the session carries its own
  # token). A freshly-built connection authenticates with it directly
  # (no LOGOFF/LOGON); a reused one is re-authed to it via
  # ensure_identity on Bolt 5.1+.
  #
  # On Bolt 5.0 there is no in-place re-auth, so a *reused* connection
  # keeps its creation-time identity. If the manager has since rotated
  # its token, that connection is stale — discard it and loop, letting
  # the pool build a fresh one authenticated as the current identity
  # (Java's backwards-compatible behaviour). Re-resolving `effective`
  # each turn means the replacement issues its own get_token, as the
  # managed-auth contract expects (one extra get_auth per rotation).
  loop do
    # Snapshot the epoch once per turn so the staleness check and the
    # connection's stamp (in ensure_identity) use one consistent value.
    epoch = auth_epoch
    effective = auth || @auth_manager.get_token
    conn = pool.pop(auth: effective, deadline: deadline)
    begin
      ensure_identity(conn, effective, session_auth: auth, epoch: epoch)
    rescue StandardError
      # Identity enforcement can raise (per-session auth on Bolt < 5.1,
      # or a re-auth LOGON failure). Free the just-checked-out slot
      # instead of leaking it for the driver's lifetime, then re-raise.
      pool.discard(conn)
      raise
    end
    # On Bolt 5.1+ ensure_identity already brought the connection to
    # the right identity (and re-auth generation). On 5.0 it can't, so
    # a reused connection is only usable if its token AND auth epoch
    # already match — otherwise discard and let the pool rebuild a
    # fresh one (token rotation, or an AuthorizationExpired refresh).
    return conn if conn.protocol.supports_re_auth? ||
                   (conn.auth == effective && conn.auth_epoch == epoch)

    pool.discard(conn)
  end
end

#auth_epoch ⇒ Object

Current auth generation, read atomically.



36
 
# File 'lib/neo4j/driver/direct/connection_provider.rb', line 36

def auth_epoch = @auth_epoch_mutex.synchronize { @auth_epoch }

#cache_home_db(_imp_user, _auth, _database) ⇒ Object 



48
 
# File 'lib/neo4j/driver/direct/connection_provider.rb', line 48

def cache_home_db(_imp_user, _auth, _database) = nil

#close ⇒ Object 



184
185
186
187
188
189
190
191
 
# File 'lib/neo4j/driver/direct/connection_provider.rb', line 184

def close
  @closed = true
  @pool&.shutdown do |conn|
    conn.close
  rescue StandardError
    nil
  end
end

#connection_pool_metrics ⇒ Object

Single-address pool snapshot (driver.metrics). Empty until the pool has been lazily built by the first acquire.



156
157
158
159
160
161
162
163
 
# File 'lib/neo4j/driver/direct/connection_provider.rb', line 156

def connection_pool_metrics
  return [] unless @pool

  address = Routing::ServerAddress.new(host: @uri.host,
                                       port: @uri.port || Routing::ServerAddress::DEFAULT_PORT)
  in_use, idle = @pool.metrics_snapshot
  [Internal::Metrics::ConnectionPoolMetrics.new(address, in_use, idle)]
end

#current_auth_token ⇒ Object

The current default identity, sourced from the auth-token manager (which refreshes / re-fetches as needed). Sessions re-auth pooled connections to this on acquire unless they carry their own per-session :auth_token.



54
 
# File 'lib/neo4j/driver/direct/connection_provider.rb', line 54

def current_auth_token = @auth_manager.get_token

#home_database(_bookmarks, _imp_user = nil, _auth = nil) ⇒ Object

No home-database resolution on the direct path: a bolt:// driver talks to one server, which resolves the user's home database itself when the operation omits db. Returns nil so the session leaves it unset (matches Java's DirectConnectionProvider).



42
 
# File 'lib/neo4j/driver/direct/connection_provider.rb', line 42

def home_database(_bookmarks, _imp_user = nil, _auth = nil) = nil

#home_db_guess(_imp_user, _auth) ⇒ Object 



47
 
# File 'lib/neo4j/driver/direct/connection_provider.rb', line 47

def home_db_guess(_imp_user, _auth) = nil

#on_security_exception(token, error, session_scoped = false) ⇒ Object

Feed a security failure back to the manager. Returns true when the manager considers it retryable (token refreshed), so the caller can re-auth and retry. An AuthorizationExpired failure also bumps the auth epoch: the server dropped its authorization cache for this identity, so every pooled connection must re-authenticate (with whatever token is current) before its next use.



62
63
64
65
66
67
68
69
70
71
 
# File 'lib/neo4j/driver/direct/connection_provider.rb', line 62

def on_security_exception(token, error, session_scoped = false)
  # Provider-side invalidation happens regardless of who owns the token.
  @auth_epoch_mutex.synchronize { @auth_epoch += 1 } if error.is_a?(Exceptions::AuthorizationExpiredException)
  # A per-session token wasn't issued by the manager — don't notify it
  # (testkit: handle_security_exception_count stays 0), and don't treat
  # the failure as manager-retryable.
  return false if session_scoped

  @auth_manager.handle_security_exception(token, error)
end

#release(connection) ⇒ Object 



123
124
125
126
127
128
129
130
131
132
133
134
 
# File 'lib/neo4j/driver/direct/connection_provider.rb', line 123

def release(connection)
  return unless connection

  # Honor a connection flagged for discard (e.g. after an auth
  # failure) — close it and free the slot rather than pooling a
  # compromised / server-closed connection.
  if connection.discard_on_release
    pool.discard(connection)
  else
    pool.push(connection)
  end
end

#routing_table_registry ⇒ Object

Mirrors Java: a Direct (bolt://) driver has no routing-table registry. testkit's GetRoutingTable / ForcedRoutingTableUpdate handlers expect a clean error here rather than NoMethodError so callers can distinguish "wrong scheme" from a real bug.

Raises:



179
180
181
182
 
# File 'lib/neo4j/driver/direct/connection_provider.rb', line 179

def routing_table_registry
  raise Exceptions::ClientException,
        'Routing table is only available on routing (neo4j://) drivers'
end

#ssr_enabled? ⇒ Boolean

No home-db cache on the direct path (one server, no discovery to skip): never guess, never track SSR. Keeps Session polymorphic across providers.

Returns:

  • (Boolean) 


46
 
# File 'lib/neo4j/driver/direct/connection_provider.rb', line 46

def ssr_enabled? = false

#supports_multi_db? ⇒ Boolean

True iff the negotiated Bolt protocol supports multi-database routing (Bolt 4.0+). Acquires a connection to ensure HELLO has happened.

Returns:

  • (Boolean) 


168
169
170
171
172
173
 
# File 'lib/neo4j/driver/direct/connection_provider.rb', line 168

def supports_multi_db?
  conn = acquire
  conn.protocol.supports_multiple_databases?
ensure
  release(conn)
end

#verify_connectivity ⇒ Object 



136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
 
# File 'lib/neo4j/driver/direct/connection_provider.rb', line 136

def verify_connectivity
  # Probe the connection with a RESET so a reused (pooled) connection
  # is actually exercised on the wire — matches Java's
  # verifyConnectivity and the routing LoadBalancer, and testkit's
  # test_direct_from_pool (which asserts one RESET on the pooled
  # connection). `propagate: true` surfaces a failed probe (rather than
  # reporting false success); a failed probe means a dead connection,
  # so discard it instead of returning it to the pool.
  conn = acquire
  begin
    conn.reset!(propagate: true)
  rescue StandardError
    pool.discard(conn)
    raise
  end
  release(conn)
end