Class: Neo4j::Driver::Transaction

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

Overview

Represents an explicit transaction

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(connection, session, bookmarks = [], options = {}, telemetry_api: nil, telemetry_ack: nil, pipelined: false, on_begin: nil, on_release: nil) ⇒ Transaction

Returns a new instance of Transaction.



9
10
11
12
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
42
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
 
# File 'lib/neo4j/driver/transaction.rb', line 9

def initialize(connection, session, bookmarks = [], options = {}, telemetry_api: nil, telemetry_ack: nil,
               pipelined: false, on_begin: nil, on_release: nil)
  @connection = connection
  @session = session
  @options = options
  # executeQuery pipelines BEGIN + RUN + PULL (Optimization:ExecuteQueryPipelining):
  # BEGIN's reply is read only after the first RUN+PULL are flushed, not eagerly.
  @pipelined = pipelined
  @on_begin = on_begin # called with the BEGIN reply (home-db cache update)
  @on_release = on_release # called once when the connection is no longer needed
  @open = true
  @committed = false
  @rolled_back = false
  @failed = false
  @terminating_error = nil # the classified error that terminated this tx (a RUN/commit failure)
  @current_result = nil
  # Every result opened in this tx, in order. Bolt lets multiple stay open
  # and streaming concurrently (qid multiplexing); a new RUN no longer
  # force-buffers the previous one, so we track them all here to discard
  # any still-open at commit/rollback and to spot a mid-stream failure.
  @open_results = []

  # Begin the transaction
  # Drop blank values so the serialised BEGIN map matches what
  # testkit's stub scripts expect (e.g. `BEGIN {"db": "adb"}`,
  # not `BEGIN {"db": "adb", "tx_metadata": {}}`).
  begin_extra = {
    bookmarks: bookmarks,
    db: options[:database],
    mode: options[:access_mode],
    tx_timeout: options[:timeout],
    tx_metadata: options[:metadata],
    imp_user: options[:impersonated_user]
  }
  begin_extra.reject!(&Internal::Extras::BLANK)

  # TELEMETRY (api = 0 managed / 1 explicit / 3 executeQuery) pipelined
  # ahead of BEGIN when the server opted in and the driver didn't disable
  # it; its SUCCESS is read first (see #ack_begin!).
  @telemetry_sent = @connection.telemetry(telemetry_api, disabled: options[:telemetry_disabled])
  @telemetry_ack = telemetry_ack
  # Session-level NotificationsConfig rides on BEGIN (5.2+); the tx's own
  # RUNs carry none. nil / pre-5.2 => no notification keys on the wire.
  @connection.send_message(@connection.protocol.build_begin(begin_extra,
                                                            notification_config: options[:notification_config]))
  @connection.flush

  # Non-pipelined: read BEGIN's reply now, a plain round-trip (unchanged).
  # Pipelined (executeQuery): defer it so the first #run can flush RUN+PULL
  # before we block — a pipelining server withholds BEGIN's SUCCESS until it
  # has all three. #ack_begin! drains it after that flush.
  @begin_acked = false
  ack_begin! unless @pipelined
rescue Exceptions::Neo4jException => e
  # Classify first so the auth-token manager is notified and the
  # connection is flagged for discard on an auth failure (the server
  # closes it). BEGIN failed → server is in FAILED state; RESET to
  # make it reusable, unless it's being discarded (a security
  # failure: the server closes it and RESET would just error).
  classified = @connection.classify_failure(e)
  @connection.reset! unless @connection.auth_failed
  @open = false
  release_connection
  raise classified
rescue StandardError
  # Transport-level failure (IO/socket). RESET will likely fail
  # too on a dead connection, but release the lease so it doesn't
  # leak; pool reuse will surface the breakage to the next caller.
  @open = false
  release_connection
  raise
end

Instance Attribute Details

#connection ⇒ Object (readonly)

Returns the value of attribute connection.



7
8
9
 
# File 'lib/neo4j/driver/transaction.rb', line 7

def connection
  @connection
end

Instance Method Details

#close ⇒ Object 



248
249
250
 
# File 'lib/neo4j/driver/transaction.rb', line 248

def close
  rollback if @open && !@committed
end

#commit ⇒ Object

Raises:



151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
 
# File 'lib/neo4j/driver/transaction.rb', line 151

def commit
  # Java/JRuby-aligned messages for the already-closed states.
  raise Exceptions::ClientException, 'Can\'t commit, transaction has been committed' if @committed
  raise Exceptions::ClientException, 'Can\'t commit, transaction has been rolled back' if @rolled_back
  raise Exceptions::ClientException, 'Transaction is already closed' unless @open

  if terminated?
    rollback_via_reset
    raise Exceptions::TransactionTerminatedException,
          "Transaction can't be committed. It has been rolled back"
  end

  begin
    discard_open_results
  rescue Exceptions::Neo4jException
    rollback_via_reset
    raise
  end

  # send/flush inside the begin block — see Transaction#run for
  # the rationale on JRuby vs MRI socket-write timing.
  response =
    begin
      @connection.send_message(Bolt::Message.commit)
      @connection.flush
      @connection.fetch_response.assert_success!
    rescue Exceptions::Neo4jException => e
      @failed = true
      # Classify first so a security failure flags the connection
      # for discard before rollback_via_reset releases it.
      classified = @connection.classify_failure(e)
      rollback_via_reset
      raise classified
    end

  @committed = true
  @open = false

  bookmarks = response.metadata[:bookmark]
  @session.update_bookmarks(bookmarks) if bookmarks
  release_connection
end

#failed? ⇒ Boolean

Returns:

  • (Boolean) 


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

def failed?
  @failed
end

#open? ⇒ Boolean

Returns:

  • (Boolean) 


252
253
254
 
# File 'lib/neo4j/driver/transaction.rb', line 252

def open?
  @open
end

#rollback ⇒ Object

Raises:



194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
 
# File 'lib/neo4j/driver/transaction.rb', line 194

def rollback
  raise Exceptions::ClientException, 'Transaction is already closed' unless @open

  # A pipelined executeQuery tx whose query never ran (e.g. local validation
  # failed before RUN/PULL were sent) left BEGIN's reply unread — and a
  # pipelining server may withhold it until RUN/PULL arrive, which now never
  # will. RESET rolls the tx back and drains any pending reply without a
  # blocking read that could deadlock; there are no open results to discard.
  return rollback_via_reset unless @begin_acked

  # A terminated tx left the connection FAILED: don't drain open results
  # (that would send PULL/DISCARD the server rejects) — just RESET.
  if terminated?
    rollback_via_reset
    return
  end

  begin
    discard_open_results
  rescue Exceptions::Neo4jException
    # Failures surfaced while draining a pending result are expected
    # during rollback — the tx is being discarded anyway. @failed is
    # set; RESET path below will clean up the connection.
  end

  if @failed
    rollback_via_reset
    return
  end

  begin
    @connection.send_message(Bolt::Message.rollback)
    @connection.flush
    @connection.fetch_response.assert_success!
  rescue Exceptions::ServiceUnavailableException, Exceptions::SessionExpiredException
    # Rolling back on a broken/dead connection is a no-op — the
    # server discards the tx when the link dies. Swallow these so
    # session.close's rollback path stays clean.
  rescue Exceptions::Neo4jException => e
    # A server FAILURE on ROLLBACK (e.g. DatabaseUnavailable) is a
    # real error: the connection is now in FAILED state. RESET it
    # back to READY, then surface the failure through the routing
    # classifier — same as commit/run — so routing side effects
    # (e.g. deactivate on DatabaseUnavailable) fire and the surfaced
    # type is consistent. No-op for direct connections.
    @connection.reset!
    raise @connection.classify_failure(e)
  ensure
    @rolled_back = true
    @open = false
    release_connection
  end
end

#run(query, **parameters) ⇒ Object 



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
 
# File 'lib/neo4j/driver/transaction.rb', line 82

def run(query, **parameters)
  Internal::Validator.require_query_text!(query)
  # A server failure terminated this transaction; further work is
  # rejected locally (no wire traffic) as a TransactionTerminatedException
  # — a ClientException subclass, matching the Java driver. Covers both a
  # failed RUN (sets @failed) and a result that failed mid-stream during
  # the user's own iteration (@current_result.failed?). Message wording
  # stays "rolled back" to match the Java flavor (a shared integration spec
  # asserts it on both impls); only the exception class becomes specific.
  if terminated?
    raise Exceptions::TransactionTerminatedException,
          'Cannot run more queries in this transaction, it has been rolled back'
  end
  unless @open
    # Mirror the Java/JRuby messages so the closed-state reason
    # (committed vs rolled back) is reported the same on both impls.
    raise Exceptions::ClientException,
          "Cannot run more queries in this transaction, it has been #{@committed ? 'committed' : 'rolled back'}"
  end

  # A new query becomes current; the previous result must now name its qid
  # explicitly on further PULL/DISCARD (the server defaults them to the last
  # opened query). We keep it open and streaming rather than buffering it —
  # that's the qid multiplexing the nested-result tests exercise.
  @current_result&.demote!

  fetch_size = effective_fetch_size

  # send/flush are inside the begin block so a transport-level
  # failure surfacing on fetch_response still sets @failed and
  # goes through classify_failure. Connection#send_message and
  # #flush both defer peer-closed errors so a buffered server
  # FAILURE is read before fetch_response can raise its own
  # EOF-driven ServiceUnavailableException — JRuby surfaces
  # EPIPE eagerly, MRI tends to defer it naturally. Without
  # this rescue placement, session.close's subsequent rollback
  # path takes the ROLLBACK-message branch (rather than
  # rollback_via_reset) and re-fails on the dead connection.
  buffer = Bolt::RecordBuffer.new(fetch_size: fetch_size)
  handler = Bolt::StreamHandler.new(buffer)
  run_response =
    begin
      @connection.send_message(@connection.protocol.build_run(query, parameters, {}))
      @connection.send_message(@connection.protocol.build_pull(n: fetch_size), handler)
      @connection.flush
      # Drain the pipelined BEGIN (+telemetry) reply now that RUN+PULL are on
      # the wire — no-op unless this is the pipelined first run. A BEGIN that
      # failed surfaces here and is handled as a run failure below; the tx
      # then rolls back on its way out, resetting the connection.
      ack_begin!
      @connection.fetch_response.assert_success!
    rescue Exceptions::Neo4jException => e
      @failed = true
      # Remember the terminating error: sibling results still open must
      # raise it (not pull) once this RUN fails — the connection is FAILED.
      raise(@terminating_error = @connection.classify_failure(e))
    end

  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,
                                                  qid: run_response.metadata[:qid],
                                                  terminated_error: method(:terminating_error))
  @open_results << @current_result
  @current_result
end