Class: Neo4j::Driver::Bolt::Connection

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

Overview

Handles a single Bolt protocol connection over TCP

Defined Under Namespace

Classes: ResponseCollector

Constant Summary collapse

DEFAULT_PORT = 
7687
READ_CHUNK =

Per-read upper bound; the wire reassembles across reads, so this is just how much we ask the socket for at once.

65_536

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

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

domain_name_resolver is the non-public hostname->IPs hook the DriverFactory wires in (default nil = system DNS). It's an explicit dependency, not part of the user options, so factory-only extension points never leak into the driver's public config.



51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
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
 
# File 'lib/neo4j/driver/bolt/connection.rb', line 51

def initialize(uri, auth, options = {}, domain_name_resolver: nil, clock: Internal::Clock.new)
  @uri = URI(uri)
  # The driver's stored auth — the identity HELLO/LOGON
  # authenticated as on connect, and what Session restores via
  # authenticate(driver_auth) when no per-session :auth was
  # given but a previous lessee had switched identity.
  @driver_auth = auth
  @auth = auth
  @options = options
  @clock = clock
  @domain_name_resolver = domain_name_resolver
  @socket = nil
  # The sans-I/O core: framing + hydration, no socket. Built once the
  # handshake has negotiated a protocol (perform_handshake). This
  # Connection is the on-demand pump over it — it owns the socket and
  # moves bytes between it and the wire on the caller's thread.
  @wire = nil
  # The on-demand pull model: every request registers @collector as its
  # response handler on the wire's FIFO; @collector appends each routed
  # message here, and fetch_response drains it. (The handler seam lets a
  # streaming request register a record-routing handler instead.)
  # The dedicated reader is the sole socket reader; it routes each reply
  # to the handler the request registered on the wire's FIFO. Sync
  # replies (RUN/BEGIN/COMMIT/RESET/ROUTE terminals) go to @collector,
  # which pushes them onto @inbox — a blocking queue fetch_response pops.
  # Streaming PULLs register a StreamHandler that fills a RecordBuffer
  # instead. @inbox is a Thread::Queue: the reader pushes, the consumer
  # pops, both colorless (yields under a Fiber scheduler).
  @inbox = Thread::Queue.new
  @collector = ResponseCollector.new(@inbox)
  # LOGOFF/LOGON replies pipelined ahead of the next operation and not yet
  # consumed (Optimization:AuthPipelining).
  @pending_auth_acks = 0
  # Guards the one-shot #on_close swap: close (caller thread) and
  # mark_closed_broken (reader thread) can race, and firing the callback
  # twice would double-decrement the routing SSR tally.
  @on_close_mutex = Mutex.new
  @recv_timeout = nil # server's connection.recv_timeout_seconds hint
  @read_deadline = nil # monotonic bound for acquisition-phase reads
  # Writes go behind a mutex: the reader writes watermark follow-up
  # nothing — but the consumer writes (new query, next PULL, DISCARD)
  # while the reader reads, so one guarded writer, never two readers.
  @write_mutex = Mutex.new
  # Dedicated reader: a Thread (per-connection lifetime) spawned lazily on
  # the first request, parked on @reader_cv when nothing is in flight,
  # stopped on close. Drives #advance and routes via the wire.
  @reader = nil
  @reader_mutex = Mutex.new
  @reader_cv = ConditionVariable.new # wakes the reader: a reply is expected
  @quiescent_cv = ConditionVariable.new # wakes drainers: in_flight hit 0
  @reader_stopped = false
  @broken_error = nil # set by failure fan-out; raised to inbox poppers
  @server_version = nil
  @bolt_version = nil
  @protocol = nil
  @server_agent = nil
  @closed = false
  @created_at = nil
  @idle_since = nil
  @discard_on_release = false
  @auth_failed = false
  @security_notified = false
  @security_classification = nil
  @session_scoped_auth = false
  @auth_epoch = 0
end

Instance Attribute Details

#address ⇒ Object (readonly)

Returns the value of attribute address.



27
28
29
 
# File 'lib/neo4j/driver/bolt/connection.rb', line 27

def address
  @address
end

#auth ⇒ Object (readonly)

Current auth identity (set by HELLO/LOGON, updated by authenticate) and the driver's stored identity (set once at construction). Sessions read driver_auth as the "no per- session :auth was given" default — calling authenticate(driver_auth) on every acquire makes the pool's auth-bleed problem disappear without needing connection-pin bookkeeping.



275
276
277
 
# File 'lib/neo4j/driver/bolt/connection.rb', line 275

def auth
  @auth
end

#auth_epoch ⇒ Object

The auth "generation" this connection last authenticated at. The provider bumps its own counter on an AuthorizationExpired failure (the server invalidated its authorization cache for every connection of this identity); a pooled connection authed at an older generation must re-authenticate on next acquire even though its token is unchanged. Set by the provider's connect_factory / ensure_identity.



301
302
303
 
# File 'lib/neo4j/driver/bolt/connection.rb', line 301

def auth_epoch
  @auth_epoch
end

#auth_failed ⇒ Object

Set on a security FAILURE specifically: the server closes the connection, so callers must NOT send a RESET (it would error / surface a spurious wire error). Distinct from discard_on_release, which also covers still-alive cases (e.g. NotALeader) that DO want a RESET before the connection is dropped.



45
46
47
 
# File 'lib/neo4j/driver/bolt/connection.rb', line 45

def auth_failed
  @auth_failed
end

#created_at ⇒ Object

idle_since: stamp the pool sets when it pushes a connection back. Bolt::Pool reads it on pop to decide whether to run a liveness probe (idle longer than the configured threshold). created_at: when the TCP/Bolt handshake finished, so the pool can evict connections older than max_connection_lifetime.



33
34
35
 
# File 'lib/neo4j/driver/bolt/connection.rb', line 33

def created_at
  @created_at
end

#discard_on_release ⇒ Object

Set when this connection must not return to the pool — e.g. an auth failure (the server closes the connection after a security FAILURE, and the identity is compromised either way). The direct provider's release discards instead of pooling. (Routing's RoutedConnection carries its own discard_on_release flag.)



39
40
41
 
# File 'lib/neo4j/driver/bolt/connection.rb', line 39

def discard_on_release
  @discard_on_release
end

#driver_auth ⇒ Object (readonly)

Current auth identity (set by HELLO/LOGON, updated by authenticate) and the driver's stored identity (set once at construction). Sessions read driver_auth as the "no per- session :auth was given" default — calling authenticate(driver_auth) on every acquire makes the pool's auth-bleed problem disappear without needing connection-pin bookkeeping.



275
276
277
 
# File 'lib/neo4j/driver/bolt/connection.rb', line 275

def driver_auth
  @driver_auth
end

#idle_since ⇒ Object

idle_since: stamp the pool sets when it pushes a connection back. Bolt::Pool reads it on pop to decide whether to run a liveness probe (idle longer than the configured threshold). created_at: when the TCP/Bolt handshake finished, so the pool can evict connections older than max_connection_lifetime.



33
34
35
 
# File 'lib/neo4j/driver/bolt/connection.rb', line 33

def idle_since
  @idle_since
end

#on_close ⇒ Object

Fired once, the first time this connection tears down (clean GOODBYE, broken read, or a failed RESET). The routing provider uses it to keep its pool-wide SSR tally current. nil for the direct provider.



284
285
286
 
# File 'lib/neo4j/driver/bolt/connection.rb', line 284

def on_close
  @on_close
end

#protocol ⇒ Object (readonly)

Returns the value of attribute protocol.



27
28
29
 
# File 'lib/neo4j/driver/bolt/connection.rb', line 27

def protocol
  @protocol
end

#security_exception_handler ⇒ Object

Provider-set callback (token, error) -> Boolean: feeds a security failure back to the auth-token manager so it can invalidate / refresh the token. nil for drivers built without a managed manager (the static case never invalidates).



352
353
354
 
# File 'lib/neo4j/driver/bolt/connection.rb', line 352

def security_exception_handler
  @security_exception_handler
end

#server_agent ⇒ Object (readonly)

Returns the value of attribute server_agent.



27
28
29
 
# File 'lib/neo4j/driver/bolt/connection.rb', line 27

def server_agent
  @server_agent
end

#server_version ⇒ Object (readonly)

Returns the value of attribute server_version.



27
28
29
 
# File 'lib/neo4j/driver/bolt/connection.rb', line 27

def server_version
  @server_version
end

#session_scoped_auth ⇒ Object

True when the connection's current identity came from a per-session auth token rather than the auth-token manager's default. The manager didn't issue that token, so a security failure on such a connection must NOT be reported to it (testkit's get_auth contract: handle_security_exception_count stays 0 for session-scoped auth). Set by the provider's ensure_identity on every acquire so it tracks the current lessee of a reused pooled connection.



346
347
348
 
# File 'lib/neo4j/driver/bolt/connection.rb', line 346

def session_scoped_auth
  @session_scoped_auth
end

Instance Method Details

#alive? ⇒ Boolean

Lightweight RESET-based liveness probe. Used by Bolt::Pool when an idle connection has been parked longer than the configured liveness threshold and we want to confirm it's still usable before handing it to a session. Any wire error OR a non-SUCCESS RESET response → return false; the pool discards and creates a fresh one. NOT reset! — that swallows errors so the original failure surfaces on the next user-driven call; here we want the probe itself to report the outcome. assert_success! is needed because fetch_response returns Message::Failure / Message::Ignored objects without raising — a "soft" RESET failure would otherwise leave the connection in the pool.

Returns:

  • (Boolean) 


186
187
188
189
190
191
192
193
194
195
196
197
198
 
# File 'lib/neo4j/driver/bolt/connection.rb', line 186

def alive?
  return false if closed?

  send_message(Message.reset)
  flush
  drain_quiesced.each(&:assert_success!)
  true
rescue StandardError
  discard_socket
  @closed = true
  fire_on_close
  false
end

#authenticate(new_auth, force: false, pipelined: true) ⇒ Object

Bolt 5.1+ re-auth: LOGOFF then LOGON with new_auth. Used by Session when it has its own :auth and the pooled connection is currently authenticated as somebody else. No-op when the connection already holds the target identity — unless force (an AuthorizationExpired-driven refresh re-auths to the same token to refresh the server's authorization cache).



309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
 
# File 'lib/neo4j/driver/bolt/connection.rb', line 309

def authenticate(new_auth, force: false, pipelined: true)
  return if !force && @auth == new_auth
  unless @protocol&.supports_re_auth?
    raise Exceptions::UnsupportedFeatureException,
          "Per-session auth requires Bolt 5.1+; negotiated #{@bolt_version}"
  end

  send_message(Message.logoff)
  send_message(Message.logon(new_auth || {}))
  @auth = new_auth
  # AuthPipelining: enqueue LOGOFF + LOGON but don't flush or read their
  # replies — they ride out with the next operation's messages and are
  # consumed (via #drain_pending_auth_acks, from the next #fetch_response)
  # just before that operation reads its own reply, saving a round-trip.
  # A rejected LOGON surfaces there as the auth failure (the operation's
  # own message is IGNORED). @auth is set optimistically; a failed re-auth
  # discards the connection, so a stale value never gets reused.
  #
  # pipelined: false forces the synchronous round-trip — verify_authentication
  # re-auths then discards the connection with no operation to carry (and
  # drain) the replies, and must see the LOGON's success/failure itself.
  if pipelined
    @pending_auth_acks += 2
  else
    flush
    fetch_response.assert_success!
    fetch_response.assert_success!
  end
end

#broken? ⇒ Boolean

Cheap, non-blocking "did the peer go away?" check the pool runs before reusing an idle pooled connection. Unlike alive?, no RESET round-trip: a clean idle connection has nothing to read, so one non-blocking read returns :wait_readable and we're done. A server that closed the idle connection (e.g. a router that served a table then EXITed — test_should_successfully_acquire_rt_when_router_ip_changes) shows up as EOF here, so the pool discards it and the next acquire re-resolves and reconnects. NOOP keepalives are drained harmlessly. This is the threaded equivalent of the reactor's background reader noticing an idle close — without a reader per parked connection.

Returns:

  • (Boolean) 


210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
 
# File 'lib/neo4j/driver/bolt/connection.rb', line 210

def broken?
  return true if closed?

  loop do
    case (chunk = @socket.read_nonblock(READ_CHUNK, exception: false))
    when :wait_readable, :wait_writable
      return false # nothing pending → healthy
    when nil
      mark_closed_broken
      return true # peer closed
    else
      @wire.receive(chunk) # NOOP / stray bytes — drain and re-check
    end
  end
rescue IOError, SystemCallError
  mark_closed_broken
  true
end

#classify_failure(error) ⇒ Object

No-op routing classifier for the direct (bolt://) path — there's no routing table to feed back. Still funnels through the auth manager. Routing::RoutedConnection overrides with the real routing classification (and also notifies). Defined here so session.rb / transaction.rb / Result#on_failure can call connection.classify_failure(e) unconditionally.



413
 
# File 'lib/neo4j/driver/bolt/connection.rb', line 413

def classify_failure(error) = notify_security_exception(error)

#close ⇒ Object 



236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
 
# File 'lib/neo4j/driver/bolt/connection.rb', line 236

def close
  return if @closed

  @closed = true
  fire_on_close
  # Best-effort GOODBYE before we tear down. Frame+write directly rather
  # than via send_message (which would re-arm the reader) / fetch (GOODBYE
  # has no reply). Then stop the reader and close the socket.
  begin
    @wire&.enqueue(Message.goodbye, @collector)
    bytes = @wire&.take_outbound
    if bytes && !bytes.empty?
      @write_mutex.synchronize do
        @socket.write(bytes)
        @socket.flush
      end
    end
  rescue StandardError
    # closing anyway
  end
  stop_reader
  begin
    @socket&.close
  rescue StandardError
    nil
  end
end

#closed? ⇒ Boolean

Returns:

  • (Boolean) 


264
265
266
 
# File 'lib/neo4j/driver/bolt/connection.rb', line 264

def closed?
  @closed || @socket&.closed?
end

#connect(deadline: nil) ⇒ Object

Raises:



118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
 
# File 'lib/neo4j/driver/bolt/connection.rb', line 118

def connect(deadline: nil)
  last_error = nil
  # One monotonic acquisition deadline for the whole connect, shared by
  # every resolved-address attempt AND the handshake/HELLO reads: a
  # server stalling the handshake — or a series of stalled addresses —
  # can't collectively outlast the acquisition timeout. Each attempt
  # gets only the *remaining* budget (open_socket / bounded reads). A
  # total deadline (not a per-read timeout) so interleaved NOOP
  # keepalives can't reset the clock. Cleared once the connection is
  # ready and steady-state reads use the recv-timeout hint instead. A
  # caller-supplied deadline (home-db optimistic acquire + fallback
  # sharing one budget) wins over this connection's own.
  @read_deadline = deadline || acquisition_deadline
  resolved_addresses.each do |host, port|
    open_socket(host, port)
    perform_handshake
    perform_hello
    @read_deadline = nil
    @created_at = current_monotonic
    # Connection is READY: hand steady-state reads to the dedicated
    # reader. (Handshake/hello above read synchronously via
    # fetch_response, so a failed connect never spawns a reader.)
    start_reader
    return self
  rescue Exceptions::AuthenticationException
    # Auth is the same regardless of which address we hit — fail fast.
    discard_socket
    raise
  rescue Exceptions::ServiceUnavailableException, IOError, SystemCallError => e
    last_error = e
    discard_socket
  end

  raise Exceptions::ServiceUnavailableException, 'No addresses to connect to' if last_error.nil?

  # A Neo4jException (e.g. the handshake's ServiceUnavailable) is
  # already classified — propagate as-is. A raw transport error
  # (Errno::ECONNRESET when a plaintext client hits a TLS-only
  # server, EOFError on a mid-handshake close, …) must be wrapped
  # so callers see a DriverError rather than a bare SystemCallError.
  # Without this, native MRI's C-OpenSSL leaks Errno::ECONNRESET and
  # the testkit-backend reports a generic BackendError instead of a
  # DriverError — the one TLS test where mri-on-jruby (whose Java
  # socket layer surfaces a classified error) diverged from native
  # mri (test_secure_server_explicitly_disabled_encryption).
  raise last_error if last_error.is_a?(Exceptions::Neo4jException)

  # Chain the original transport error as `cause` (this raise is
  # outside the per-address rescue, so it's set explicitly rather
  # than auto-populated from $!). Preserves the underlying failure
  # and its backtrace behind the wrapper.
  raise Exceptions::ServiceUnavailableException,
        "Unable to connect to #{@address || @uri}: #{last_error.class}: #{last_error.message}",
        cause: last_error
end

#current_monotonic ⇒ Object

Monotonic seconds — immune to wall-clock jumps, which is what every age / idle calculation here needs. Through the Clock seam so Backend:MockTime can freeze/advance it.



232
233
234
 
# File 'lib/neo4j/driver/bolt/connection.rb', line 232

def current_monotonic
  @clock.monotonic
end

#drain_pending_auth_acks ⇒ Object

Consume the replies to a pipelined re-auth's LOGOFF + LOGON before the next operation reads its own reply. A rejected LOGON raises the auth failure here (its follow-on messages come back IGNORED).



624
625
626
627
628
629
630
 
# File 'lib/neo4j/driver/bolt/connection.rb', line 624

def drain_pending_auth_acks
  return if @pending_auth_acks.zero?

  pending = @pending_auth_acks
  @pending_auth_acks = 0
  pending.times { pop_inbox.assert_success! }
end

#fetch_all ⇒ Object 



632
633
634
 
# File 'lib/neo4j/driver/bolt/connection.rb', line 632

def fetch_all
  drain_quiesced
end

#fetch_response ⇒ Object

Return the next sync reply in request order. The dedicated reader fills reader delivers. On a connection failure the reader closes @inbox and records @broken_error, so a blocked pop wakes with nil and re-raises the classified error rather than hanging.



602
603
604
605
606
607
 
# File 'lib/neo4j/driver/bolt/connection.rb', line 602

def fetch_response
  # A pipelined re-auth's LOGOFF/LOGON replies sit ahead of this
  # operation's own reply — consume them first (AuthPipelining).
  drain_pending_auth_acks
  pop_inbox
end

#flush ⇒ Object

Defer peer-closed errors from flush so a buffered server response (e.g. a final FAILURE) gets read before we raise. Under JRuby the peer-closed state surfaces eagerly on the very next write/flush; raising here would swallow the FAILURE bytes already in the receive buffer — the test_should_error_on_database_shutdown_using_tx_run stub regression. Every normal request/response cycle pairs flush with a fetch_response (Transaction#run/commit/rollback, Result streaming, Connection#route), so a peer-gone state with nothing buffered still surfaces as ServiceUnavailableException — just from the read side. Connection#close also calls flush but discards exceptions itself (flush rescue nil), so it does not need the pair. Non-peer-closed wire errors (e.g. a timed-out write on a socket that has SO_SNDTIMEO set, or EBADF on a closed-out-from-under-us fd) are NOT silenced — they fall through and propagate. We do not set SO_SNDTIMEO and the fd is owned by us, so these are improbable in practice. Drain the wire's outbound buffer to the socket. Writes are mutex- guarded so a future prefetch reader and the consumer's writes never interleave on one socket. Peer-closed errors are deferred (not raised) so a server FAILURE buffered before the close is still read by the paired fetch_response — every request/response cycle pairs flush with a fetch (Transaction#run/commit/rollback, Result streaming, #route), so a genuinely-gone peer still surfaces as ServiceUnavailable from the read side. #close flushes with rescue nil, so it needs no pair.



573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
 
# File 'lib/neo4j/driver/bolt/connection.rb', line 573

def flush
  bytes = @wire.take_outbound
  return if bytes.empty?

  begin
    @write_mutex.synchronize do
      @socket.write(bytes)
      @socket.flush
    end
  rescue Errno::EPIPE, Errno::ECONNRESET, IOError
    # Peer-closed (EPIPE/ECONNRESET) or the reader closed the socket
    # out from under this write mid-flight (IOError "stream closed in
    # another thread", EBADF). Deferred (not raised) so a server FAILURE
    # buffered before the close is still read by the paired
    # fetch_response — see method comment.
  ensure
    # Always wake the reader, even on a failed write: a reply may be
    # expected, OR the socket is dead and the reader must run #advance to
    # hit EOF and fan the failure out — otherwise a parked reader never
    # discovers the break and a drainer in #wait_quiescent hangs.
    wake_reader
  end
end

#notify_security_exception(error) ⇒ Object

Report a security failure to the auth-token manager (if any) and let it decide retryability. Every operation site funnels errors through classify_failure, so this is the single notification point. When the manager deems the failure retryable (it has invalidated the token so the next acquire re-fetches), surface a SecurityRetryableException wrapping the original (code/message preserved, original chained as cause when the caller re-raises) — mirrors Java, and the testkit reports it as retryable.



362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
 
# File 'lib/neo4j/driver/bolt/connection.rb', line 362

def notify_security_exception(error)
  return error unless error.is_a?(Exceptions::SecurityException)

  # A security failure means this connection can't be reused — drop
  # it from the pool either way.
  @discard_on_release = true

  # AuthorizationExpired is the server's authorization-cache expiry,
  # not a closed socket — the connection stays usable, so RESET is
  # fine. Unauthorized / TokenExpired close the connection server-
  # side, so skip RESET there.
  @auth_failed = true unless error.is_a?(Exceptions::AuthorizationExpiredException)

  # Notify the auth-token manager at most once per connection. The
  # same failure is classified again as it propagates (result
  # streaming on_failure, then the tx rollback re-consuming the
  # failed result), so without this guard the manager's
  # handle_security_exception fires twice. We also cache the
  # classified result and return it on the repeat calls, so the
  # retryability/type stays stable — the first call may upgrade to
  # SecurityRetryableException, and a later call must not downgrade
  # back to the raw error. The connection is discarded after a
  # security failure, so neither the flag nor the cache needs
  # clearing.
  return @security_classification if @security_notified

  @security_notified = true
  # Always run the provider handler — it performs provider-side work
  # that must happen regardless of who owns the token, notably bumping
  # the auth epoch on AuthorizationExpired so SIBLING pooled
  # connections re-authenticate. `session_scoped_auth` is passed so the
  # handler can skip the auth-token-MANAGER notification for a
  # per-session identity (the manager didn't issue that token, so its
  # handle_security_exception_count must stay 0). A retryable verdict
  # surfaces a SecurityRetryableException wrapping the original
  # (code/message preserved, original chained as cause at re-raise);
  # the handler returns false for session-scoped auth, so no upgrade.
  @security_classification =
    if @security_exception_handler&.call(@auth, error, @session_scoped_auth)
      Exceptions::SecurityRetryableException.new(error.message, code: error.code)
    else
      error
    end
end

#pending_responses? ⇒ Boolean

True while the caller still owes a fetch: a request whose terminal the wire hasn't seen yet (in_flight), or a message already routed to the inbox but not yet popped. The drain loops spin on this.

Returns:

  • (Boolean) 


665
666
667
 
# File 'lib/neo4j/driver/bolt/connection.rb', line 665

def pending_responses?
  @wire.in_flight.positive? || !@inbox.empty?
end

#pop_inbox ⇒ Object

Pop the next sync reply. Acquisition phase (no reader yet): drive the reads ourselves. Steady state: the reader fills @inbox; block on pop until it delivers. On a connection failure the reader closes @inbox and records @broken_error, so a blocked pop wakes with nil and re-raises.

Raises:

  • (@broken_error) 


613
614
615
616
617
618
619
 
# File 'lib/neo4j/driver/bolt/connection.rb', line 613

def pop_inbox
  advance while @reader.nil? && @inbox.empty?
  message = @inbox.pop
  raise @broken_error if message.nil? && @broken_error

  message
end

#reset!(propagate: false) ⇒ Object

Recover from a FAILED server state. Sends RESET and drains all pending responses (including any IGNOREDs from messages queued before the failure — those routed to their handlers; this drains the sync @inbox). Returns once the server has acknowledged the RESET and the connection is quiescent.



641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
 
# File 'lib/neo4j/driver/bolt/connection.rb', line 641

def reset!(propagate: false)
  send_message(Message.reset)
  flush
  messages = drain_quiesced
  # When propagating (verify_connectivity, pool-return) the RESET is a
  # real check: a server FAILURE reply (not just a dead socket) must
  # surface too, so assert success on the drained responses.
  messages.each(&:assert_success!) if propagate
rescue StandardError
  # If RESET itself fails the connection is likely dead. Recovery paths
  # (`propagate: false`, the default) swallow so they don't mask the
  # original error and the caller discovers the break on next use.
  # verify_connectivity passes `propagate: true`: the RESET *is* the
  # probe, so a failure must surface (and the dead connection be
  # discarded) rather than report false success.
  raise if propagate
ensure
  # RESET flushed and drained any pipelined re-auth replies with it.
  @pending_auth_acks = 0
end

#route(database: nil, bookmarks: [], imp_user: nil, routing_context: {}) ⇒ Object

Fetch the cluster routing table. Bolt 4.3+ uses the dedicated ROUTE message; older versions have no ROUTE and call a server-side procedure instead (route_via_procedure). Either way the return is the {ttl:, servers:} map the caller wraps in Routing::RoutingTable.from_response.



457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
 
# File 'lib/neo4j/driver/bolt/connection.rb', line 457

def route(database: nil, bookmarks: [], imp_user: nil, routing_context: {})
  # Enforce impersonation support before touching the wire: a
  # routed session impersonating against a pre-4.4 cluster must
  # fail (ClientException) rather than silently drop imp_user from
  # the discovery call. Raised here — outside the wire-error
  # begin/rescue — so the still-clean connection isn't RESET.
  @protocol.enforce_impersonation_support!(imp_user)

  return route_via_procedure(database, bookmarks, routing_context) if @bolt_version < BoltVersion::V4_3

  begin
    # ROUTE's 3rd field changed at 4.4: 4.3 sends the bare database
    # name (string/null), 4.4+ a `{db, imp_user}` map. The protocol
    # handler owns that shape. The acquisition timeout must encompass
    # discovery, so bound the ROUTE read by the deadline (cleared after).
    @read_deadline = acquisition_deadline
    send_message(@protocol.build_route(routing_context, Array(bookmarks), database, imp_user))
    flush

    fetch_response.assert_success!.metadata[:rt]
  rescue Exceptions::Neo4jException
    # ROUTE failure leaves the server in FAILED state — RESET clears it
    # so the connection can be reused.
    reset!
    raise
  ensure
    @read_deadline = nil
  end
end

#route_via_procedure(database, bookmarks, routing_context) ⇒ Object

Pre-4.3 routing: there is no ROUTE message, so fetch the table by calling the server-side procedure and shaping its single row ([ttl, servers]) into the same map ROUTE would return. Bolt 3.0: CALL dbms.cluster.routing.getRoutingTable($context) on the home database (single-DB protocol). Bolt 4.0-4.2: CALL dbms.routing.getRoutingTable($context, $database) run against the system database.



494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
 
# File 'lib/neo4j/driver/bolt/connection.rb', line 494

def route_via_procedure(database, bookmarks, routing_context)
  if @bolt_version >= BoltVersion::V4_0
    # 4.0-4.2: dbms.routing.getRoutingTable run against `system`.
    # Pass $database only when a target db is named — the home-db
    # case uses the single-arg form (matches the server procedure
    # overloads the stub scripts pin).
    if database
      query = 'CALL dbms.routing.getRoutingTable($context, $database)'
      params = { context: routing_context, database: database }
    else
      query = 'CALL dbms.routing.getRoutingTable($context)'
      params = { context: routing_context }
    end
    extra = { db: 'system', mode: 'r' }
    # 4.0-4.2 runs the procedure against `system`, which accepts
    # bookmarks (causal consistency for a freshly-created database).
    extra[:bookmarks] = Array(bookmarks) unless Array(bookmarks).empty?
  else
    # 3.0: single-database cluster routing procedure, home db. No
    # system db and no bookmark-aware routing (that arrived with the
    # 4.3 ROUTE message), so the discovery RUN carries only `mode`.
    query = 'CALL dbms.cluster.routing.getRoutingTable($context)'
    params = { context: routing_context }
    extra = { mode: 'r' }
  end

  send_message(@protocol.build_run(query, params, extra))
  send_message(@protocol.build_pull(n: -1))
  flush

  summary = fetch_response.assert_success!
  fields = summary.metadata[:fields] || summary.metadata['fields'] || []
  row = nil
  loop do
    response = fetch_response
    case response
    when Message::Success then break # PULL summary — end of stream
    when Message::Record then row ||= fields.zip(response.fields).to_h
    else response.assert_success! # FAILURE / IGNORED — raises
    end
  end

  unless row
    raise Exceptions::ServiceUnavailableException,
          "Routing procedure on #{@address || @uri} returned no rows"
  end

  { ttl: row['ttl'], servers: row['servers'] }
rescue Exceptions::Neo4jException
  reset!
  raise
end

#send_all(*messages) ⇒ Object 



436
437
438
439
 
# File 'lib/neo4j/driver/bolt/connection.rb', line 436

def send_all(*messages)
  messages.each { |msg| send_message(msg) }
  flush
end

#send_message(message, handler = @collector) ⇒ Object

Frame the message into the wire's outbound buffer and count it as in-flight. Nothing hits the socket until #flush — so several send_messages before a flush pipeline naturally (HELLO+LOGON, RUN+PULL), which is the whole point.

A dead/closed connection raises a classified Neo4jException, not a raw IOError: the cleanup and retry paths (Transaction#rollback, reset!, the managed-tx retry) rescue Neo4jException, so a bare IOError would escape them and surface as an unhandled error. Register a request on the wire's FIFO with the handler that will route its reply: @collector (→ @inbox) for sync requests, or a StreamHandler (→ a RecordBuffer) for a streaming PULL. The dedicated reader (started once the connection is READY) delivers it; during the acquisition phase (handshake/hello, before the reader exists) fetch_response drives the reads synchronously.

Raises:



430
431
432
433
434
 
# File 'lib/neo4j/driver/bolt/connection.rb', line 430

def send_message(message, handler = @collector)
  raise Exceptions::ServiceUnavailableException, "Connection to #{@address || @uri} is closed" if closed?

  @wire.enqueue(message, handler)
end

#ssr_enabled? ⇒ Boolean

Whether the server advertised ssr.enabled in its HELLO hints (Bolt 5.8+ server-side routing). Gates the optimistic home-db cache.

Returns:

  • (Boolean) 


279
 
# File 'lib/neo4j/driver/bolt/connection.rb', line 279

def ssr_enabled? = @ssr_enabled == true

#telemetry(api, disabled:) ⇒ Object

Enqueue a TELEMETRY report for the API about to open a tx/query, unless the caller disabled it, the server didn't advertise telemetry, or the negotiated protocol predates it (5.4). Returns whether one was sent so the caller reads its (extra, pipelined) SUCCESS before the op's reply.



445
446
447
448
449
450
 
# File 'lib/neo4j/driver/bolt/connection.rb', line 445

def telemetry(api, disabled:)
  return false if api.nil? || disabled || !@telemetry_enabled || !@protocol.supports_telemetry?

  send_message(@protocol.build_telemetry(api))
  true
end