Class: Neo4j::Driver::Session

Inherits:
Object
  • Object
show all
Includes:
Internal::DurationNormalizer
Defined in:
lib/neo4j/driver/session.rb

Overview

Per-operation connection acquisition: each run / begin_transaction / execute_read / execute_write acquires a fresh connection from the driver with that operation's access mode and database, and releases it back to the pool when the operation completes (Result fully consumed, or transaction committed/rolled back). The session itself does not hold a connection — only its in-flight operation does.

Instance Method Summary collapse

Methods included from Internal::DurationNormalizer

#timeout_to_milliseconds

Constructor Details

#initialize(connection_provider, options = {}, clock: Internal::Clock.new) ⇒ Session

Returns a new instance of Session.



13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
 
# File 'lib/neo4j/driver/session.rb', line 13

def initialize(connection_provider, options = {}, clock: Internal::Clock.new)
  @connection_provider = connection_provider
  @options = options
  @clock = clock
  @transaction = nil
  @open = true
  @last_bookmarks = Set.new
  @current_result = nil
  # The home database this session pinned after its first resolution
  # (Optimization:HomeDatabaseCache); nil until then / for explicit-db sessions.
  @pinned_database = nil
  @used_guess = false  # whether an operation optimistically guessed the home db
  @bookmark_manager = options[:bookmark_manager]
  # The bookmark snapshot we sent on the most recent BEGIN — used
  # as `previous_bookmarks` when forwarding to the manager so it
  # set-difference-drops what it knew before. Computed at BEGIN
  # time (not update time) because the manager may have grown
  # between the two via another session.
  @bookmarks_used_on_begin = nil

  # Initialize with provided bookmarks if any
  if options[:bookmarks]
    initial_bookmarks = options[:bookmarks]
    initial_bookmarks = [initial_bookmarks] unless initial_bookmarks.is_a?(Enumerable)
    initial_bookmarks.each do |bookmark|
      @last_bookmarks << (bookmark.is_a?(Bookmark) ? bookmark : Bookmark.from(bookmark))
    end
  end
end

Instance Method Details

#begin_transaction(timeout: nil, metadata: nil, &block) ⇒ Object

Raises:



170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
 
# File 'lib/neo4j/driver/session.rb', line 170

def begin_transaction(timeout: nil, metadata: nil, &block)
  raise Exceptions::ClientException, 'Session is closed' unless @open
  if @transaction&.open?
    raise Exceptions::ClientException,
          "You cannot begin a transaction on a session with an open transaction; either run from within the transaction or use a different session."
  end

  drain_current_result
  @current_result = nil

  tx_options = @options.merge(
    # The resolved home database is added in open_transaction, where the
    # BEGIN bookmark snapshot is computed (so resolution and BEGIN share
    # one snapshot).
    timeout: timeout_to_milliseconds(timeout),
    metadata:,
    # BEGIN carries `mode: "r"` for read sessions (write is the
    # server default and omits it), matching the auto-commit path.
    # The Transaction reads tx_options[:access_mode]; without this an
    # explicit read transaction dropped the field (session stores the
    # mode under :default_access_mode, not :access_mode).
    access_mode: (session_access_mode == :read ? 'r' : nil)
  ).compact

  # TELEMETRY 1 = unmanaged (explicit) transaction.
  @transaction = open_transaction(session_access_mode, tx_options, telemetry_api: 1)

  if block_given?
    begin
      result = yield @transaction
      # Explicit-block transactions default to rollback; user must call
      # tx.commit to persist changes (matches Java driver semantics).
      @transaction.rollback if @transaction.open?
      result
    rescue => e
      @transaction.rollback if @transaction.open?
      raise e
    ensure
      @transaction = nil
    end
  else
    @transaction
  end
end

#close ⇒ Object

Close any pending result by asking the server to abandon remaining records (DISCARD) rather than pulling them just to throw them away client-side. Matches Java's behaviour and is required by testkit's session_run.test_discard_on_session_close_* scripts which lie about has_more=true to verify the driver eventually sends DISCARD. With pagination, draining via buffer would loop forever on those scripts.



230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
 
# File 'lib/neo4j/driver/session.rb', line 230

def close
  return unless @open

  pending_error = nil
  begin
    @transaction&.close  # tx releases its own connection
    if @current_result
      connection = @current_result.connection
      begin
        @current_result.consume
      rescue Exceptions::Neo4jException => e
        pending_error = e
      end
      # Any failure during the consume leaves the connection in
      # the server's FAILED state; RESET so the next borrower starts
      # from a clean READY state.
      connection.reset! if @current_result.failed?
      @current_result.discard!  # idempotent; releases the connection
    end
  ensure
    @open = false
  end

  raise pending_error if pending_error
end

#execute_read(timeout: nil, metadata: nil, &block) ⇒ Object 



215
216
217
 
# File 'lib/neo4j/driver/session.rb', line 215

def execute_read(timeout: nil, metadata: nil, &block)
  execute_transaction(AccessMode::READ, timeout:, metadata:, &block)
end

#execute_write(timeout: nil, metadata: nil, &block) ⇒ Object 



219
220
221
 
# File 'lib/neo4j/driver/session.rb', line 219

def execute_write(timeout: nil, metadata: nil, &block)
  execute_transaction(AccessMode::WRITE, timeout:, metadata:, &block)
end

#last_bookmarks ⇒ Object 



260
261
262
 
# File 'lib/neo4j/driver/session.rb', line 260

def last_bookmarks
  @last_bookmarks
end

#open? ⇒ Boolean

Returns:

  • (Boolean) 


256
257
258
 
# File 'lib/neo4j/driver/session.rb', line 256

def open?
  @open
end

#run(query, parameters = {}, config = {}) ⇒ Object

Raises:



43
44
45
46
47
48
49
50
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
117
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
 
# File 'lib/neo4j/driver/session.rb', line 43

def run(query, parameters = {}, config = {})
  Internal::Validator.require_query_text!(query)
  parameters ||= {}

  unless parameters.is_a?(Hash)
    raise ArgumentError,
          "The parameters should be provided as Map type. Unsupported parameters type: #{parameters.class}"
  end

  raise Exceptions::ClientException, 'Session is closed' unless @open
  raise Exceptions::ClientException, 'You cannot run a query directly on a session while a transaction is open' if @transaction&.open?

  drain_current_result

  timeout = config.delete(:timeout)
  tx_metadata = config.delete(:metadata)

  # `mode` is sent for read sessions (matches routing-server
  # expectations); writers are the default and omit the field.
  # `bookmarks` is the current bookmark snapshot so server-side
  # causal consistency works across auto-commit runs without an
  # explicit transaction.
  # `imp_user` (Bolt 4.4+) makes the server run the query as if
  # the named user had issued it — auth as the session's user,
  # authz as the impersonated one.
  bookmarks = current_bookmarks_for_extra
  # Acquire (with its own bookmark snapshot) so the home-db cache can pick
  # the routing table: on a cache guess the RUN sends db=nil (server
  # re-resolves), otherwise it pins the discovered name. run_db carries
  # whichever applies; run_extra keeps this operation's bookmark snapshot.
  connection, run_db = acquire_for_operation(session_access_mode)
  run_extra = {
    db: run_db,
    mode: (session_access_mode == :read ? 'r' : nil),
    tx_timeout: timeout_to_milliseconds(timeout),
    tx_metadata:,
    imp_user: @options[:impersonated_user],
    bookmarks: bookmarks
  }
  run_extra.reject!(&Internal::Extras::BLANK)

  fetch_size = effective_fetch_size
  # Feature:IdempotentRetries — an auto-commit RUN whose failure the server
  # flags `_idempotent` may be retried once: RESET clears the FAILED state
  # (and drains the pipelined PULL's IGNORED), then RUN+PULL are re-sent
  # (TELEMETRY is not). Only the RUN reply is eligible — a telemetry failure
  # or a later stream (PULL) failure is raised as usual.
  retries_left = auto_commit_retries_enabled? ? 1 : 0

  buffer = handler = run_response = nil
  telemetry_sent = false    # TELEMETRY 2 = auto-commit; sent once, never resent on a retry
  telemetry_pending = false # its (opted-in) SUCCESS is still owed
  loop do
    # A fresh stream per attempt — a prior attempt's buffer saw the IGNORED.
    buffer = Bolt::RecordBuffer.new(fetch_size: fetch_size)
    handler = Bolt::StreamHandler.new(buffer)
    retry_run = false
    # `stage` tells the rescue which reply raised: only a :run failure is
    # idempotent-retryable (not :telemetry, nor a send/flush transport error).
    stage = nil
    run_response =
      begin
        # TELEMETRY is pipelined ahead of the first RUN when the server opted
        # in; kept inside this block so a send failure still releases the lease.
        unless telemetry_sent
          telemetry_sent = true
          telemetry_pending = connection.telemetry(2, disabled: @options[:telemetry_disabled])
        end
        # Auto-commit RUN carries this session's NotificationsConfig (5.2+);
        # the tx path puts it on BEGIN instead. nil / non-5.2 => absent.
        connection.send_message(connection.protocol.build_run(query, parameters, run_extra,
                                                              notification_config: @options[:notification_config]))
        connection.send_message(connection.protocol.build_pull(n: fetch_size), handler)
        connection.flush
        if telemetry_pending
          telemetry_pending = false
          stage = :telemetry
          connection.fetch_response.assert_success!
        end
        stage = :run
        connection.fetch_response.assert_success!
      rescue Exceptions::Neo4jException => e
        # Classify first: this notifies the auth-token manager (and, for a
        # security failure, sets auth_failed so the guards below hold) and,
        # for routed connections, fires on_write_failure / deactivate and
        # swaps NotALeader for SessionExpired. assert_success! raises
        # *outside* RoutedConnection's wrapper, so this is the only place the
        # classifier sees FAILURE responses to RUN.
        classified = connection.classify_failure(e)
        if stage == :run && retries_left.positive? && idempotent_error?(e) && !connection.auth_failed
          retries_left -= 1
          # RESET clears FAILED and drains the abandoned RUN/PULL replies.
          connection.reset!
          retry_run = true
          nil
        else
          # Server is in FAILED state; RESET so the connection is immediately
          # reusable — but not if it's being discarded (auth failure: the
          # server closes it, RESET would just error).
          connection.reset! unless connection.auth_failed
          @connection_provider.release(connection)
          raise classified
        end
      rescue StandardError
        # Transport-level failure (IO/socket) — the connection is likely
        # dead. Return it to the pool either way so this lease doesn't leak;
        # the next user will rediscover the breakage.
        @connection_provider.release(connection)
        raise
      end
    break unless retry_run
  end

  # A home-db RUN that sent db=nil comes back with the server's resolved
  # home database — cache it so the next same-identity session can guess.
  cache_home_db_from(run_response)

  keys = (run_response.metadata[:fields] || run_response.metadata['fields'] || []).map(&:to_sym)
  @current_result = Result.new(
    connection, keys, buffer: buffer, handler: handler,
    query_text: query, parameters: parameters, run_metadata: run_response.metadata,
    fetch_size: fetch_size,
    on_summary: method(:harvest_auto_commit_bookmark),
    on_release: -> { @connection_provider.release(connection) }
  )
end

#update_bookmarks(bookmarks) ⇒ Object 



264
265
266
267
268
269
270
271
272
273
274
 
# File 'lib/neo4j/driver/session.rb', line 264

def update_bookmarks(bookmarks)
  # Replace bookmarks (don't accumulate) — matches Java driver behavior.
  # Each committed transaction generates a new bookmark that replaces the previous one.
  new_set = Set.new(Array(bookmarks).map(&Bookmark.method(:from)))
  # Forward to the BookmarkManager (if configured) so cross-session
  # causal consistency works. Pass the bookmarks we *sent* in BEGIN
  # as `previous` — Java's manager uses set-difference, so passing
  # the session's own bookmarks alongside the manager's is safe.
  @bookmark_manager&.update_bookmarks(@bookmarks_used_on_begin || @last_bookmarks, new_set)
  @last_bookmarks = new_set
end