Class: Mongo::Session

Inherits:
Object
  • Object
show all
Extended by:
Forwardable
Includes:
Loggable, Retryable
Defined in:
lib/mongo/session.rb,
lib/mongo/session/session_pool.rb,
lib/mongo/session/server_session.rb

Overview

A logical session representing a set of sequential operations executed by an application that are related in some way.

Since:

  • 2.5.0

Defined Under Namespace

Classes: ServerSession, SessionPool

Constant Summary collapse

MISMATCHED_CLUSTER_ERROR_MSG =

Error message indicating that the session was retrieved from a client with a different cluster than that of the client through which it is currently being used.

Since:

  • 2.5.0

'The configuration of the client used to create this session does not match that ' +
'of the client owning this operation. Please only use this session for operations through its parent ' +
'client.'.freeze
SESSION_ENDED_ERROR_MSG =

Error message describing that the session cannot be used because it has already been ended.

Since:

  • 2.5.0

'This session has ended and cannot be used. Please create a new one.'.freeze
SESSIONS_NOT_SUPPORTED =

Error message describing that sessions are not supported by the server version.

Since:

  • 2.5.0

'Sessions are not supported by the connected servers.'.freeze
NO_TRANSACTION_STATE =

The state of a session in which the last operation was not related to any transaction or no operations have yet occurred.

Since:

  • 2.6.0

:no_transaction
STARTING_TRANSACTION_STATE =

The state of a session in which a user has initiated a transaction but no operations within the transactions have occurred yet.

Since:

  • 2.6.0

:starting_transaction
TRANSACTION_IN_PROGRESS_STATE =

The state of a session in which a transaction has been started and at least one operation has occurred, but the transaction has not yet been committed or aborted.

Since:

  • 2.6.0

:transaction_in_progress
TRANSACTION_COMMITTED_STATE =

The state of a session in which the last operation executed was a transaction commit.

Since:

  • 2.6.0

:transaction_committed
TRANSACTION_ABORTED_STATE =

The state of a session in which the last operation executed was a transaction abort.

Since:

  • 2.6.0

:transaction_aborted
UNLABELED_WRITE_CONCERN_CODES =

Since:

  • 2.5.0

[
  79,  # UnknownReplWriteConcern
  100, # CannotSatisfyWriteConcern,
].freeze

Constants included from Loggable

Loggable::PREFIX

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from Loggable

#log_debug, #log_error, #log_fatal, #log_info, #log_warn, #logger

Methods included from Retryable

#read_with_one_retry, #read_with_retry, #write_with_retry

Constructor Details

#initialize(server_session, client, options = {}) ⇒ Session

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Note:

Applications should use Client#start_session to begin a session.

Initialize a Session.

Examples:

Session.new(server_session, client, options)

Parameters:

  • server_session (ServerSession)

    The server session this session is associated with.

  • client (Client)

    The client through which this session is created.

  • options (Hash) (defaults to: {})

    The options for this session.

Options Hash (options):

  • :causal_consistency (true|false)

    Whether to enable causal consistency for this session.

  • :default_transaction_options (Hash)

    Options to pass to start_transaction by default, can contain any of the options that start_transaction accepts.

  • :implicit (true|false)

    For internal driver use only - specifies whether the session is implicit.

  • :read_preference (Hash)

    The read preference options hash, with the following optional keys:

    • :mode – the read preference as a string or symbol; valid values are :primary, :primary_preferred, :secondary, :secondary_preferred and :nearest.

Since:

  • 2.5.0



132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
# File 'lib/mongo/session.rb', line 132

def initialize(server_session, client, options = {})
  @server_session = server_session
  options = options.dup

  # Because the read preference will need to be inserted into a command as a string, we convert
  # it from a symbol immediately upon receiving it.
  if options[:read_preference] && options[:read_preference][:mode]
    options[:read_preference][:mode] = options[:read_preference][:mode].to_s
  end

  @client = client.use(:admin)
  @options = options.freeze
  @cluster_time = nil
  @state = NO_TRANSACTION_STATE
end

Instance Attribute Details

#clientObject (readonly)

Get the client through which this session was created.

Since:

  • 2.5.1



37
38
39
# File 'lib/mongo/session.rb', line 37

def client
  @client
end

#cluster_timeObject (readonly)

The cluster time for this session.

Since:

  • 2.5.0



42
43
44
# File 'lib/mongo/session.rb', line 42

def cluster_time
  @cluster_time
end

#operation_timeObject (readonly)

The latest seen operation time for this session.

Since:

  • 2.5.0



47
48
49
# File 'lib/mongo/session.rb', line 47

def operation_time
  @operation_time
end

#optionsObject (readonly)

Get the options for this session.

Since:

  • 2.5.0



32
33
34
# File 'lib/mongo/session.rb', line 32

def options
  @options
end

#txn_optionsObject (readonly)

The options for the transaction currently being executed on the session.

Since:

  • 2.6.0



52
53
54
# File 'lib/mongo/session.rb', line 52

def txn_options
  @txn_options
end

Instance Method Details

#abort_transactionObject

Abort the currently active transaction without making any changes to the database.

Examples:

Abort the transaction.

session.abort_transaction

Raises:

Since:

  • 2.6.0



668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
# File 'lib/mongo/session.rb', line 668

def abort_transaction
  check_if_ended!
  check_if_no_transaction!

  if within_states?(TRANSACTION_COMMITTED_STATE)
    raise Mongo::Error::InvalidTransactionOperation.new(
      Mongo::Error::InvalidTransactionOperation.cannot_call_after_msg(
        :commitTransaction, :abortTransaction))
  end

  if within_states?(TRANSACTION_ABORTED_STATE)
    raise Mongo::Error::InvalidTransactionOperation.new(
      Mongo::Error::InvalidTransactionOperation.cannot_call_twice_msg(:abortTransaction))
  end

  begin
    unless starting_transaction?
      write_with_retry(self, txn_options[:write_concern], true) do |server, txn_num|
        Operation::Command.new(
          selector: { abortTransaction: 1 },
          db_name: 'admin',
          session: self,
          txn_num: txn_num
        ).execute(server)
      end
    end

    @state = TRANSACTION_ABORTED_STATE
  rescue Mongo::Error::InvalidTransactionOperation
    raise
  rescue Mongo::Error
    @state = TRANSACTION_ABORTED_STATE
  rescue Exception
    @state = TRANSACTION_ABORTED_STATE
    raise
  end
end

#add_autocommit!(command) ⇒ Hash, BSON::Document

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Add the autocommit field to a command document if applicable.

Examples:

session.add_autocommit!(cmd)

Returns:

  • (Hash, BSON::Document)

    The command document.

Since:

  • 2.6.0



203
204
205
206
207
# File 'lib/mongo/session.rb', line 203

def add_autocommit!(command)
  command.tap do |c|
    c[:autocommit] = false if in_transaction?
  end
end

#add_id!(command) ⇒ Hash, BSON::Document

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Add this session's id to a command document.

Examples:

session.add_id!(cmd)

Returns:

  • (Hash, BSON::Document)

    The command document.

Since:

  • 2.5.0



218
219
220
# File 'lib/mongo/session.rb', line 218

def add_id!(command)
  command.merge!(lsid: session_id)
end

#add_start_transaction!(command) ⇒ Hash, BSON::Document

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Add the startTransaction field to a command document if applicable.

Examples:

session.add_start_transaction!(cmd)

Returns:

  • (Hash, BSON::Document)

    The command document.

Since:

  • 2.6.0



231
232
233
234
235
236
237
# File 'lib/mongo/session.rb', line 231

def add_start_transaction!(command)
  command.tap do |c|
    if starting_transaction?
      c[:startTransaction] = true
    end
  end
end

#add_txn_num!(command) ⇒ Hash, BSON::Document

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Add the transaction number to a command document if applicable.

Examples:

session.add_txn_num!(cmd)

Returns:

  • (Hash, BSON::Document)

    The command document.

Since:

  • 2.6.0



248
249
250
251
252
# File 'lib/mongo/session.rb', line 248

def add_txn_num!(command)
  command.tap do |c|
    c[:txnNumber] = BSON::Int64.new(@server_session.txn_num) if in_transaction?
  end
end

#add_txn_opts!(command, read) ⇒ Hash, BSON::Document

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Add the transactions options if applicable.

Examples:

session.add_txn_opts!(cmd)

Returns:

  • (Hash, BSON::Document)

    The command document.

Since:

  • 2.6.0



263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
# File 'lib/mongo/session.rb', line 263

def add_txn_opts!(command, read)
  command.tap do |c|
    # The read preference should be added for all read operations.
    if read && txn_read_pref = txn_read_preference
      Mongo::Lint.validate_underscore_read_preference(txn_read_pref)
      txn_read_pref = txn_read_pref.dup
      txn_read_pref[:mode] = txn_read_pref[:mode].to_s.gsub(/(_\w)/) { |match| match[1].upcase }
      Mongo::Lint.validate_camel_case_read_preference(txn_read_pref)
      c['$readPreference'] = txn_read_pref
    end

    # The read concern should be added to any command that starts a transaction.
    if starting_transaction?
      # https://jira.mongodb.org/browse/SPEC-1161: transaction's
      # read concern overrides collection/database/client read concerns,
      # even if transaction's read concern is not set.
      # Read concern here is the one sent to the server and may
      # include afterClusterTime.
      if rc = c[:readConcern]
        rc = rc.dup
        rc.delete(:level)
      end
      if txn_read_concern
        if rc
          rc.update(txn_read_concern)
        else
          rc = txn_read_concern.dup
        end
      end
      if rc.nil? || rc.empty?
        c.delete(:readConcern)
      else
        c[:readConcern ] = rc
      end
    end

    # We need to send the read concern level as a string rather than a symbol.
    if c[:readConcern] && c[:readConcern][:level]
      c[:readConcern][:level] = c[:readConcern][:level].to_s
    end

    # The write concern should be added to any abortTransaction or commitTransaction command.
    if (c[:abortTransaction] || c[:commitTransaction])
      if @already_committed
        wc = BSON::Document.new(c[:writeConcern] || txn_write_concern || {})
        wc.merge!(w: :majority)
        wc[:wtimeout] ||= 10000
        c[:writeConcern] = wc
      elsif txn_write_concern
        c[:writeConcern] ||= txn_write_concern
      end
    end

    # A non-numeric write concern w value needs to be sent as a string rather than a symbol.
    if c[:writeConcern] && c[:writeConcern][:w] && c[:writeConcern][:w].is_a?(Symbol)
      c[:writeConcern][:w] = c[:writeConcern][:w].to_s
    end
  end
end

#advance_cluster_time(new_cluster_time) ⇒ BSON::Document, Hash

Advance the cached cluster time document for this session.

Examples:

Advance the cluster time.

session.advance_cluster_time(doc)

Parameters:

  • new_cluster_time (BSON::Document, Hash)

    The new cluster time.

Returns:

  • (BSON::Document, Hash)

    The new cluster time.

Since:

  • 2.5.0



420
421
422
423
424
425
426
# File 'lib/mongo/session.rb', line 420

def advance_cluster_time(new_cluster_time)
  if @cluster_time
    @cluster_time = [ @cluster_time, new_cluster_time ].max_by { |doc| doc[Cluster::CLUSTER_TIME] }
  else
    @cluster_time = new_cluster_time
  end
end

#advance_operation_time(new_operation_time) ⇒ BSON::Timestamp

Advance the cached operation time for this session.

Examples:

Advance the operation time.

session.advance_operation_time(timestamp)

Parameters:

  • new_operation_time (BSON::Timestamp)

    The new operation time.

Returns:

  • (BSON::Timestamp)

    The max operation time, considering the current and new times.

Since:

  • 2.5.0



438
439
440
441
442
443
444
# File 'lib/mongo/session.rb', line 438

def advance_operation_time(new_operation_time)
  if @operation_time
    @operation_time = [ @operation_time, new_operation_time ].max
  else
    @operation_time = new_operation_time
  end
end

#clusterObject

Since:

  • 2.5.0



873
874
875
# File 'lib/mongo/session.rb', line 873

def cluster
  @client.cluster
end

#commit_transaction(options = nil) ⇒ Object

Commit the currently active transaction on the session.

Examples:

Commits the transaction.

session.commit_transaction

Parameters:

  • options (Hash) (defaults to: nil)

    a customizable set of options

Options Hash (options):

  • :write_concern (nil | WriteConcern::Base)

    The write concern to use for this operation.

Raises:

Since:

  • 2.6.0



595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
# File 'lib/mongo/session.rb', line 595

def commit_transaction(options=nil)
  check_if_ended!
  check_if_no_transaction!

  if within_states?(TRANSACTION_ABORTED_STATE)
    raise Mongo::Error::InvalidTransactionOperation.new(
      Mongo::Error::InvalidTransactionOperation.cannot_call_after_msg(
        :abortTransaction, :commitTransaction))
  end

  options ||= {}

  begin
    # If commitTransaction is called twice, we need to run the same commit
    # operation again, so we revert the session to the previous state.
    if within_states?(TRANSACTION_COMMITTED_STATE)
      @state = @last_commit_skipped ? STARTING_TRANSACTION_STATE : TRANSACTION_IN_PROGRESS_STATE
      @already_committed = true
    end

    if starting_transaction?
      @last_commit_skipped = true
    else
      @last_commit_skipped = false

      write_concern = options[:write_concern] || txn_options[:write_concern]
      if write_concern && !write_concern.is_a?(WriteConcern::Base)
        write_concern = WriteConcern.get(write_concern)
      end
      write_with_retry(self, write_concern, true) do |server, txn_num, is_retry|
        if is_retry
          if write_concern
            wco = write_concern.options.merge(w: :majority)
            wco[:wtimeout] ||= 10000
            write_concern = WriteConcern.get(wco)
          else
            write_concern = WriteConcern.get(w: :majority, wtimeout: 10000)
          end
        end
        Operation::Command.new(
          selector: { commitTransaction: 1 },
          db_name: 'admin',
          session: self,
          txn_num: txn_num,
          write_concern: write_concern,
        ).execute(server)
      end
    end
  rescue Mongo::Error::NoServerAvailable, Mongo::Error::SocketError => e
    e.send(:add_label, Mongo::Error::UNKNOWN_TRANSACTION_COMMIT_RESULT_LABEL)
    raise e
  rescue Mongo::Error::OperationFailure => e
    err_doc = e.instance_variable_get(:@result).send(:first_document)

    if e.write_retryable? || (err_doc['writeConcernError'] &&
        !UNLABELED_WRITE_CONCERN_CODES.include?(err_doc['writeConcernError']['code']))
      e.send(:add_label, Mongo::Error::UNKNOWN_TRANSACTION_COMMIT_RESULT_LABEL)
    end

    raise e
  ensure
    @state = TRANSACTION_COMMITTED_STATE
  end
end

#end_sessionnil

End this session.

Examples:

session.end_session

Returns:

  • (nil)

    Always nil.

Since:

  • 2.5.0



168
169
170
171
172
173
174
175
176
177
178
179
180
# File 'lib/mongo/session.rb', line 168

def end_session
  if !ended? && @client
    if within_states?(TRANSACTION_IN_PROGRESS_STATE)
      begin
        abort_transaction
      rescue Mongo::Error
      end
    end
    @client.cluster.session_pool.checkin(@server_session)
  end
ensure
  @server_session = nil
end

#ended?true, false

Whether this session has ended.

Examples:

session.ended?

Returns:

  • (true, false)

    Whether the session has ended.

Since:

  • 2.5.0



190
191
192
# File 'lib/mongo/session.rb', line 190

def ended?
  @server_session.nil?
end

#explicit?true, false

Is this session an explicit one (i.e. user-created).

Examples:

Is the session explicit?

session.explicit?

Returns:

  • (true, false)

    Whether this session is explicit.

Since:

  • 2.5.2



531
532
533
# File 'lib/mongo/session.rb', line 531

def explicit?
  @explicit ||= !implicit?
end

#implicit?true, false

Is this session an implicit one (not user-created).

Examples:

Is the session implicit?

session.implicit?

Returns:

  • (true, false)

    Whether this session is implicit.

Since:

  • 2.5.1



519
520
521
# File 'lib/mongo/session.rb', line 519

def implicit?
  @implicit ||= !!(@options.key?(:implicit) && @options[:implicit] == true)
end

#in_transaction?true | false

Whether or not the session is currently in a transaction.

Examples:

Is the session in a transaction?

session.in_transaction?

Returns:

  • (true | false)

    Whether or not the session in a transaction.

Since:

  • 2.6.0



714
715
716
# File 'lib/mongo/session.rb', line 714

def in_transaction?
  within_states?(STARTING_TRANSACTION_STATE, TRANSACTION_IN_PROGRESS_STATE)
end

#inspectString

Get a formatted string for use in inspection.

Examples:

Inspect the session object.

session.inspect

Returns:

  • (String)

    The session inspection.

Since:

  • 2.5.0



156
157
158
# File 'lib/mongo/session.rb', line 156

def inspect
  "#<Mongo::Session:0x#{object_id} session_id=#{session_id} options=#{@options}>"
end

#next_txn_numInteger

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Increment and return the next transaction number.

Examples:

Get the next transaction number.

session.next_txn_num

Returns:

  • (Integer)

    The next transaction number.

Since:

  • 2.5.0



487
488
489
490
491
492
493
# File 'lib/mongo/session.rb', line 487

def next_txn_num
  if ended?
    raise Error::SessionEnded
  end

  @server_session.next_txn_num
end

#process(result) ⇒ Operation::Result

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Process a response from the server that used this session.

Examples:

Process a response from the server.

session.process(result)

Parameters:

Returns:

Since:

  • 2.5.0



401
402
403
404
405
406
407
408
# File 'lib/mongo/session.rb', line 401

def process(result)
  unless implicit?
    set_operation_time(result)
    set_cluster_time(result)
  end
  @server_session.set_last_use!
  result
end

#retry_writes?true, false

Note:

Retryable writes are only available on server versions at least 3.6 and with sharded clusters or replica sets.

Will writes executed with this session be retried.

Examples:

Will writes be retried.

session.retry_writes?

Returns:

  • (true, false)

    If writes will be retried.

Since:

  • 2.5.0



457
458
459
# File 'lib/mongo/session.rb', line 457

def retry_writes?
  !!client.options[:retry_writes] && (cluster.replica_set? || cluster.sharded?)
end

#session_idBSON::Document

Get the server session id of this session, if the session was not ended. If the session was ended, returns nil.

Examples:

Get the session id.

session.session_id

Returns:

  • (BSON::Document)

    The server session id.

Since:

  • 2.5.0



470
471
472
473
474
475
476
# File 'lib/mongo/session.rb', line 470

def session_id
  if ended?
    raise Error::SessionEnded
  end

  @server_session.session_id
end

#start_transaction(options = nil) ⇒ Object

Places subsequent operations in this session into a new transaction.

Note that the transaction will not be started on the server until an operation is performed after start_transaction is called.

Examples:

Start a new transaction

session.start_transaction(options)

Parameters:

  • options (Hash) (defaults to: nil)

    The options for the transaction being started.

Options Hash (options):

  • read_concern (Hash)

    The read concern options hash, with the following optional keys:

    • :level – the read preference level as a symbol; valid values

      are *:local*, *:majority*, and *:snapshot*
      
  • :write_concern (Hash)

    The write concern options. Can be :w => Integer|String, :fsync => Boolean, :j => Boolean.

  • :read (Hash)

    The read preference options. The hash may have the following items:

    • :mode – read preference specified as a symbol; the only valid value is :primary.

Raises:

Since:

  • 2.6.0



560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
# File 'lib/mongo/session.rb', line 560

def start_transaction(options = nil)
  if options
    Lint.validate_read_concern_option(options[:read_concern])
  end

  check_if_ended!

  if within_states?(STARTING_TRANSACTION_STATE, TRANSACTION_IN_PROGRESS_STATE)
    raise Mongo::Error::InvalidTransactionOperation.new(
      Mongo::Error::InvalidTransactionOperation::TRANSACTION_ALREADY_IN_PROGRESS)
  end

  next_txn_num
  @txn_options = options || @options[:default_transaction_options] || {}

  if txn_write_concern && WriteConcern.send(:unacknowledged?, txn_write_concern)
    raise Mongo::Error::InvalidTransactionOperation.new(
      Mongo::Error::InvalidTransactionOperation::UNACKNOWLEDGED_WRITE_CONCERN)
  end

  @state = STARTING_TRANSACTION_STATE
  @already_committed = false
end

#suppress_read_write_concern!(command) ⇒ Hash, BSON::Document

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Remove the read concern and/or write concern from the command if not applicable.

Examples:

session.suppress_read_write_concern!(cmd)

Returns:

  • (Hash, BSON::Document)

    The command document.

Since:

  • 2.6.0



332
333
334
335
336
337
338
339
# File 'lib/mongo/session.rb', line 332

def suppress_read_write_concern!(command)
  command.tap do |c|
    next unless in_transaction?

    c.delete(:readConcern) unless starting_transaction?
    c.delete(:writeConcern) unless c[:commitTransaction] || c[:abortTransaction]
  end
end

#txn_numInteger

Get the current transaction number.

Examples:

Get the current transaction number.

session.txn_num

Returns:

  • (Integer)

    The current transaction number.

Since:

  • 2.6.0



503
504
505
506
507
508
509
# File 'lib/mongo/session.rb', line 503

def txn_num
  if ended?
    raise Error::SessionEnded
  end

  @server_session.txn_num
end

#txn_read_preferenceHash

Get the read preference the session will use in the currently active transaction.

This is a driver style hash with underscore keys.

Examples:

Get the transaction's read preference

session.txn_read_preference

Returns:

  • (Hash)

    The read preference of the transaction.

Since:

  • 2.6.0



866
867
868
869
870
871
# File 'lib/mongo/session.rb', line 866

def txn_read_preference
  rp = txn_options && txn_options[:read_preference] ||
    @client.read_preference
  Mongo::Lint.validate_underscore_read_preference(rp)
  rp
end

#update_state!Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Update the state of the session due to a (non-commit and non-abort) operation being run.

Since:

  • 2.6.0



362
363
364
365
366
367
368
369
# File 'lib/mongo/session.rb', line 362

def update_state!
  case @state
  when STARTING_TRANSACTION_STATE
    @state = TRANSACTION_IN_PROGRESS_STATE
  when TRANSACTION_COMMITTED_STATE, TRANSACTION_ABORTED_STATE
    @state = NO_TRANSACTION_STATE
  end
end

#validate!(cluster) ⇒ nil

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Validate the session.

Examples:

session.validate!(cluster)

Parameters:

  • cluster (Cluster)

    The cluster the session is attempted to be used with.

Returns:

  • (nil)

    nil if the session is valid.

Raises:

Since:

  • 2.5.0



384
385
386
387
388
# File 'lib/mongo/session.rb', line 384

def validate!(cluster)
  check_matching_cluster!(cluster)
  check_if_ended!
  self
end

#validate_read_preference!(command) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Ensure that the read preference of a command primary.

Examples:

session.validate_read_preference!(command)

Raises:

Since:

  • 2.6.0



351
352
353
354
355
356
# File 'lib/mongo/session.rb', line 351

def validate_read_preference!(command)
  return unless in_transaction? && non_primary_read_preference_mode?(command)

  raise Mongo::Error::InvalidTransactionOperation.new(
    Mongo::Error::InvalidTransactionOperation::INVALID_READ_PREFERENCE)
end

#with_transaction(options = nil) ⇒ Object

Note:

with_transaction contains a loop, therefore the if with_transaction itself is placed in a loop, its block should not call next or break to control the outer loop because this will instead affect the loop in with_transaction. The driver will warn and abort the transaction if it detects this situation.

Executes the provided block in a transaction, retrying as necessary.

Returns the return value of the block.

Exact number of retries and when they are performed are implementation details of the driver; the provided block should be idempotent, and should be prepared to be called more than once. The driver may retry the commit command within an active transaction or it may repeat the transaction and invoke the block again, depending on the error encountered if any. Note also that the retries may be executed against different servers.

Transactions cannot be nested - InvalidTransactionOperation will be raised if this method is called when the session already has an active transaction.

Exceptions raised by the block which are not derived from Mongo::Error stop processing, abort the transaction and are propagated out of with_transaction. Exceptions derived from Mongo::Error may be handled by with_transaction, resulting in retries of the process.

Currently, with_transaction will retry commits and block invocations until at least 120 seconds have passed since with_transaction started executing. This timeout is not configurable and may change in a future driver version.

Examples:

Execute a statement in a transaction

session.with_transaction(write_concern: {w: :majority}) do
  collection.update_one({ id: 3 }, { '$set' => { status: 'Inactive'} },
                        session: session)

end

Execute a statement in a transaction, limiting total time consumed

Timeout.timeout(5) do
  session.with_transaction(write_concern: {w: :majority}) do
    collection.update_one({ id: 3 }, { '$set' => { status: 'Inactive'} },
                          session: session)

  end
end

Parameters:

  • options (Hash) (defaults to: nil)

    The options for the transaction being started. These are the same options that start_transaction accepts.

Raises:

Since:

  • 2.7.0



772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
# File 'lib/mongo/session.rb', line 772

def with_transaction(options=nil)
  # Non-configurable 120 second timeout for the entire operation
  deadline = Time.now + 120
  transaction_in_progress = false
  loop do
    commit_options = {}
    if options
      commit_options[:write_concern] = options[:write_concern]
    end
    start_transaction(options)
    transaction_in_progress = true
    begin
      rv = yield self
    rescue Exception => e
      if within_states?(STARTING_TRANSACTION_STATE, TRANSACTION_IN_PROGRESS_STATE)
        abort_transaction
        transaction_in_progress = false
      end

      if Time.now >= deadline
        transaction_in_progress = false
        raise
      end

      if e.is_a?(Mongo::Error) && e.label?(Mongo::Error::TRANSIENT_TRANSACTION_ERROR_LABEL)
        next
      end

      raise
    else
      if within_states?(TRANSACTION_ABORTED_STATE, NO_TRANSACTION_STATE, TRANSACTION_COMMITTED_STATE)
        transaction_in_progress = false
        return rv
      end

      begin
        commit_transaction(commit_options)
        transaction_in_progress = false
        return rv
      rescue Mongo::Error => e
        if e.label?(Mongo::Error::UNKNOWN_TRANSACTION_COMMIT_RESULT_LABEL)
          # WriteConcernFailed
          if e.is_a?(Mongo::Error::OperationFailure) && e.code == 64 && e.wtimeout?
            transaction_in_progress = false
            raise
          end
          if Time.now >= deadline
            transaction_in_progress = false
            raise
          end
          wc_options = case v = commit_options[:write_concern]
            when WriteConcern::Base
              v.options
            when nil
              {}
            else
              v
            end
          commit_options[:write_concern] = wc_options.merge(w: :majority)
          retry
        elsif e.label?(Mongo::Error::TRANSIENT_TRANSACTION_ERROR_LABEL)
          if Time.now >= deadline
            transaction_in_progress = false
            raise
          end
          next
        else
          transaction_in_progress = false
          raise
        end
      end
    end
  end
ensure
  if transaction_in_progress
    log_warn('with_transaction callback altered with_transaction loop, aborting transaction')
    begin
      abort_transaction
    rescue Error::OperationFailure, Error::InvalidTransactionOperation
    end
  end
end