Class: Mongo::Session
- Inherits:
-
Object
- Object
- Mongo::Session
- Extended by:
- Forwardable
- Includes:
- ClusterTime::Consumer, Loggable, Retryable
- Defined in:
- lib/mongo/session.rb,
lib/mongo/session/session_pool.rb,
lib/mongo/session/server_session.rb
Overview
Session objects are not thread-safe. An application may use a session from only one thread or process at a time.
A logical session representing a set of sequential operations executed by an application that are related in some way.
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.
'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.
'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.
'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.
: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.
: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.
:transaction_in_progress- TRANSACTION_COMMITTED_STATE =
The state of a session in which the last operation executed was a transaction commit.
:transaction_committed- TRANSACTION_ABORTED_STATE =
The state of a session in which the last operation executed was a transaction abort.
:transaction_aborted- UNLABELED_WRITE_CONCERN_CODES =
This constant is part of a private API. You should avoid using this constant if possible, as it may be removed or be changed in the future.
[ 79, # UnknownReplWriteConcern 100, # CannotSatisfyWriteConcern, ].freeze
Constants included from Loggable
Instance Attribute Summary collapse
-
#client ⇒ Client
readonly
The client through which this session was created.
-
#operation_time ⇒ BSON::Timestamp
readonly
The latest seen operation time for this session.
-
#options ⇒ Hash
readonly
The options for this session.
-
#pinned_server ⇒ Server | nil
readonly
private
The server (which should be a mongos) that this session is pinned to, if any.
-
#recovery_token ⇒ BSON::Document | nil
private
Recovery token for the sharded transaction being executed on this session, if any.
Attributes included from ClusterTime::Consumer
Instance Method Summary collapse
-
#abort_transaction ⇒ Object
Abort the currently active transaction without making any changes to the database.
-
#add_autocommit!(command) ⇒ Hash, BSON::Document
private
Add the autocommit field to a command document if applicable.
-
#add_start_transaction!(command) ⇒ Hash, BSON::Document
private
Add the startTransaction field to a command document if applicable.
-
#add_txn_num!(command) ⇒ Hash, BSON::Document
private
Add the transaction number to a command document if applicable.
-
#add_txn_opts!(command, read) ⇒ Hash, BSON::Document
private
Add the transactions options if applicable.
-
#advance_operation_time(new_operation_time) ⇒ BSON::Timestamp
Advance the cached operation time for this session.
- #cluster ⇒ Object
-
#commit_transaction(options = nil) ⇒ Object
Commit the currently active transaction on the session.
-
#committing_transaction? ⇒ true | false
private
Whether the session is currently committing a transaction.
-
#end_session ⇒ nil
End this session.
-
#ended? ⇒ true, false
Whether this session has ended.
-
#explicit? ⇒ true, false
Is this session an explicit one (i.e. user-created).
-
#implicit? ⇒ true, false
Is this session an implicit one (not user-created).
-
#in_transaction? ⇒ true | false
Whether or not the session is currently in a transaction.
-
#initialize(server_session, client, options = {}) ⇒ Session
constructor
private
Initialize a Session.
-
#inspect ⇒ String
Get a formatted string for use in inspection.
-
#next_txn_num ⇒ Integer
private
Increment and return the next transaction number.
-
#pin(server) ⇒ Object
private
Pins this session to the specified server, which should be a mongos.
-
#process(result) ⇒ Operation::Result
private
Process a response from the server that used this session.
-
#retry_reads? ⇒ Boolean
private
Whether reads executed with this session can be retried according to the modern retryable reads specification.
-
#retry_writes? ⇒ true, false
Will writes executed with this session be retried.
-
#session_id ⇒ BSON::Document
Get the server session id of this session, if the session was not ended.
-
#start_transaction(options = nil) ⇒ Object
Places subsequent operations in this session into a new transaction.
- #starting_transaction? ⇒ Boolean private
-
#suppress_read_write_concern!(command) ⇒ Hash, BSON::Document
private
Remove the read concern and/or write concern from the command if not applicable.
-
#txn_num ⇒ Integer
Get the current transaction number.
-
#txn_options ⇒ Hash
on this session.
-
#txn_read_preference ⇒ Hash
Get the read preference the session will use in the currently active transaction.
-
#unpin ⇒ Object
private
Unpins this session from the pinned server, if the session was pinned.
-
#unpin_maybe(error) ⇒ Object
private
Unpins this session from the pinned server, if the session was pinned and the specified exception instance and the session’s transaction state require it to be unpinned.
-
#update_state! ⇒ Object
private
Update the state of the session due to a (non-commit and non-abort) operation being run.
-
#validate!(client) ⇒ Session
private
Validate the session for use by the specified client.
-
#validate_read_preference!(command) ⇒ Object
private
Ensure that the read preference of a command primary.
-
#with_transaction(options = nil) ⇒ Object
Executes the provided block in a transaction, retrying as necessary.
Methods included from ClusterTime::Consumer
Methods included from Loggable
#log_debug, #log_error, #log_fatal, #log_info, #log_warn, #logger
Methods included from Retryable
#legacy_write_with_retry, #nro_write_with_retry, #read_with_one_retry, #read_with_retry, #read_with_retry_cursor, #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.
Applications should use Client#start_session to begin a session.
Initialize a Session.
59 60 61 62 63 64 65 66 67 |
# File 'lib/mongo/session.rb', line 59 def initialize(server_session, client, = {}) @server_session = server_session = .dup @client = client.use(:admin) = .freeze @cluster_time = nil @state = NO_TRANSACTION_STATE end |
Instance Attribute Details
#client ⇒ Client (readonly)
Returns The client through which this session was created.
77 78 79 |
# File 'lib/mongo/session.rb', line 77 def client @client end |
#operation_time ⇒ BSON::Timestamp (readonly)
Returns The latest seen operation time for this session.
86 87 88 |
# File 'lib/mongo/session.rb', line 86 def operation_time @operation_time end |
#options ⇒ Hash (readonly)
Returns The options for this session.
72 73 74 |
# File 'lib/mongo/session.rb', line 72 def end |
#pinned_server ⇒ Server | nil (readonly)
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.
Returns The server (which should be a mongos) that this session is pinned to, if any.
206 207 208 |
# File 'lib/mongo/session.rb', line 206 def pinned_server @pinned_server end |
#recovery_token ⇒ BSON::Document | 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.
Returns Recovery token for the sharded transaction being executed on this session, if any.
212 213 214 |
# File 'lib/mongo/session.rb', line 212 def recovery_token @recovery_token end |
Instance Method Details
#abort_transaction ⇒ Object
Abort the currently active transaction without making any changes to the database.
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 |
# File 'lib/mongo/session.rb', line 602 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, [:write_concern], true) do |server, txn_num| Operation::Command.new( selector: { abortTransaction: 1 }, db_name: 'admin', session: self, txn_num: txn_num ).execute(server, client: @client) 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 # No official return value, but return true so that in interactive # use the method hints that it succeeded. true 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.
726 727 728 729 730 |
# File 'lib/mongo/session.rb', line 726 def add_autocommit!(command) command.tap do |c| c[:autocommit] = false if in_transaction? end 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.
741 742 743 744 745 746 747 |
# File 'lib/mongo/session.rb', line 741 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.
758 759 760 761 762 |
# File 'lib/mongo/session.rb', line 758 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.
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 |
# File 'lib/mongo/session.rb', line 773 def add_txn_opts!(command, read) command.tap do |c| # 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 ] = Options::Mapper.transform_values_to_strings(rc) end end # We need to send the read concern level as a string rather than a symbol. if c[:readConcern] c[:readConcern] = Options::Mapper.transform_values_to_strings(c[:readConcern]) end if c[:commitTransaction] if max_time_ms = [:max_commit_time_ms] c[:maxTimeMS] = max_time_ms end 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_operation_time(new_operation_time) ⇒ BSON::Timestamp
Advance the cached operation time for this session.
942 943 944 945 946 947 948 |
# File 'lib/mongo/session.rb', line 942 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 |
#cluster ⇒ Object
79 80 81 |
# File 'lib/mongo/session.rb', line 79 def cluster @client.cluster end |
#commit_transaction(options = nil) ⇒ Object
Commit the currently active transaction on the session.
534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 |
# File 'lib/mongo/session.rb', line 534 def commit_transaction(=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 ||= {} 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 @committing_transaction = true write_concern = [:write_concern] || [: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..merge(w: :majority) wco[:wtimeout] ||= 10000 write_concern = WriteConcern.get(wco) else write_concern = WriteConcern.get(w: :majority, wtimeout: 10000) end end spec = { selector: { commitTransaction: 1 }, db_name: 'admin', session: self, txn_num: txn_num, write_concern: write_concern, } Operation::Command.new(spec).execute(server, client: @client) end end ensure @state = TRANSACTION_COMMITTED_STATE @committing_transaction = false end # No official return value, but return true so that in interactive # use the method hints that it succeeded. true end |
#committing_transaction? ⇒ true | false
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.
Returns Whether the session is currently committing a transaction.
665 666 667 |
# File 'lib/mongo/session.rb', line 665 def committing_transaction? !!@committing_transaction end |
#end_session ⇒ nil
End this session.
If there is an in-progress transaction on this session, the transaction is aborted. The server session associated with this session is returned to the server session pool. Finally, this session is marked ended and is no longer usable.
If this session is already ended, this method does nothing.
Note that this method does not directly issue an endSessions command to this server, contrary to what its name might suggest.
297 298 299 300 301 302 303 304 305 306 307 308 309 |
# File 'lib/mongo/session.rb', line 297 def end_session if !ended? && @client if within_states?(TRANSACTION_IN_PROGRESS_STATE) begin abort_transaction rescue Mongo::Error, Error::AuthError end end @client.cluster.session_pool.checkin(@server_session) end ensure @server_session = nil end |
#ended? ⇒ true, false
Whether this session has ended.
181 182 183 |
# File 'lib/mongo/session.rb', line 181 def ended? @server_session.nil? end |
#explicit? ⇒ true, false
Is this session an explicit one (i.e. user-created).
116 117 118 |
# File 'lib/mongo/session.rb', line 116 def explicit? !implicit? end |
#implicit? ⇒ true, false
Is this session an implicit one (not user-created).
104 105 106 |
# File 'lib/mongo/session.rb', line 104 def implicit? @implicit ||= !!(.key?(:implicit) && [:implicit] == true) end |
#in_transaction? ⇒ true | false
Whether or not the session is currently in a transaction.
657 658 659 |
# File 'lib/mongo/session.rb', line 657 def in_transaction? within_states?(STARTING_TRANSACTION_STATE, TRANSACTION_IN_PROGRESS_STATE) end |
#inspect ⇒ String
Get a formatted string for use in inspection.
275 276 277 |
# File 'lib/mongo/session.rb', line 275 def inspect "#<Mongo::Session:0x#{object_id} session_id=#{session_id} options=#{@options}>" end |
#next_txn_num ⇒ Integer
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.
959 960 961 962 963 964 965 |
# File 'lib/mongo/session.rb', line 959 def next_txn_num if ended? raise Error::SessionEnded end @server_session.next_txn_num end |
#pin(server) ⇒ 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.
Pins this session to the specified server, which should be a mongos.
674 675 676 677 678 679 680 681 682 683 684 |
# File 'lib/mongo/session.rb', line 674 def pin(server) if server.nil? raise ArgumentError, 'Cannot pin to a nil server' end if Lint.enabled? unless server.mongos? raise Error::LintError, "Attempted to pin the session to server #{server.summary} which is not a mongos" end end @pinned_server = server 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.
914 915 916 917 918 919 920 921 922 923 924 925 926 927 928 929 930 |
# File 'lib/mongo/session.rb', line 914 def process(result) unless implicit? set_operation_time(result) if cluster_time_doc = result.cluster_time advance_cluster_time(cluster_time_doc) end end @server_session.set_last_use! if doc = result.reply && result.reply.documents.first if doc[:recoveryToken] self.recovery_token = doc[:recoveryToken] end end result end |
#retry_reads? ⇒ Boolean
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.
Whether reads executed with this session can be retried according to the modern retryable reads specification.
If this method returns true, the modern retryable reads have been requested by the application. If the server selected for a read operation supports modern retryable reads, they will be used for that particular operation. If the server selected for a read operation does not support modern retryable reads, the read will not be retried.
If this method returns false, legacy retryable reads have been requested by the application. Legacy retryable read logic will be used regardless of server version of the server(s) that the client is connected to. The number of read retries is given by :max_read_retries client option, which is 1 by default and can be set to 0 to disable legacy read retries.
136 137 138 |
# File 'lib/mongo/session.rb', line 136 def retry_reads? client.[:retry_reads] != false end |
#retry_writes? ⇒ true, false
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.
151 152 153 |
# File 'lib/mongo/session.rb', line 151 def retry_writes? !!client.[:retry_writes] && (cluster.replica_set? || cluster.sharded?) end |
#session_id ⇒ BSON::Document
Get the server session id of this session, if the session was not ended. If the session was ended, returns nil.
194 195 196 197 198 199 200 |
# File 'lib/mongo/session.rb', line 194 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.
479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 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 |
# File 'lib/mongo/session.rb', line 479 def start_transaction( = nil) if Lint.validate_read_concern_option([:read_concern]) =begin # It would be handy to detect invalid read preferences here, but # some of the spec tests require later detection of invalid read prefs. # Maybe we can do this when lint mode is on. mode = options[:read] && options[:read][:mode].to_s if mode && mode != 'primary' raise Mongo::Error::InvalidTransactionOperation.new( "read preference in a transaction must be primary (requested: #{mode})" ) end =end 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 unpin next_txn_num = ([:default_transaction_options] || {}).merge( || {}) if txn_write_concern && !WriteConcern.get(txn_write_concern).acknowledged? raise Mongo::Error::InvalidTransactionOperation.new( Mongo::Error::InvalidTransactionOperation::UNACKNOWLEDGED_WRITE_CONCERN) end @state = STARTING_TRANSACTION_STATE @already_committed = false # This method has no explicit return value. # We could return nil here but true indicates to the user that the # operation succeeded. This is intended for interactive use. # Note that the return value is not documented. true end |
#starting_transaction? ⇒ Boolean
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.
645 646 647 |
# File 'lib/mongo/session.rb', line 645 def starting_transaction? within_states?(STARTING_TRANSACTION_STATE) 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.
839 840 841 842 843 844 845 846 |
# File 'lib/mongo/session.rb', line 839 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_num ⇒ Integer
Get the current transaction number.
975 976 977 978 979 980 981 |
# File 'lib/mongo/session.rb', line 975 def txn_num if ended? raise Error::SessionEnded end @server_session.txn_num end |
#txn_options ⇒ Hash
on this session.
92 93 94 |
# File 'lib/mongo/session.rb', line 92 def or raise ArgumentError, "There is no active transaction" end |
#txn_read_preference ⇒ Hash
Get the read preference the session will use in the currently active transaction.
This is a driver style hash with underscore keys.
166 167 168 169 170 171 |
# File 'lib/mongo/session.rb', line 166 def txn_read_preference rp = [:read] || @client.read_preference Mongo::Lint.validate_underscore_read_preference(rp) rp end |
#unpin ⇒ 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.
Unpins this session from the pinned server, if the session was pinned.
689 690 691 |
# File 'lib/mongo/session.rb', line 689 def unpin @pinned_server = nil end |
#unpin_maybe(error) ⇒ 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.
Unpins this session from the pinned server, if the session was pinned and the specified exception instance and the session’s transaction state require it to be unpinned.
The exception instance should already have all of the labels set on it (both client- and server-side generated ones).
703 704 705 706 707 708 709 710 711 712 713 714 715 |
# File 'lib/mongo/session.rb', line 703 def unpin_maybe(error) if !within_states?(Session::NO_TRANSACTION_STATE) && error.label?('TransientTransactionError') then unpin end if committing_transaction? && error.label?('UnknownTransactionCommitResult') then unpin end 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.
875 876 877 878 879 880 881 882 |
# File 'lib/mongo/session.rb', line 875 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!(client) ⇒ 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.
Validate the session for use by the specified client.
The session must not be ended and must have been created by a client with the same cluster as the client that the session is to be used with.
897 898 899 900 901 |
# File 'lib/mongo/session.rb', line 897 def validate!(client) check_matching_cluster!(client) 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.
858 859 860 861 862 863 864 865 866 867 868 869 |
# File 'lib/mongo/session.rb', line 858 def validate_read_preference!(command) return unless in_transaction? return unless command['$readPreference'] mode = command['$readPreference']['mode'] || command['$readPreference'][:mode] if mode && mode != 'primary' raise Mongo::Error::InvalidTransactionOperation.new( "read preference in a transaction must be primary (requested: #{mode})" ) end end |
#with_transaction(options = nil) ⇒ Object
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.
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 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 |
# File 'lib/mongo/session.rb', line 365 def with_transaction(=nil) # Non-configurable 120 second timeout for the entire operation deadline = Time.now + 120 transaction_in_progress = false loop do = {} if [:write_concern] = [:write_concern] end start_transaction() 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?('TransientTransactionError') 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() transaction_in_progress = false return rv rescue Mongo::Error => e if e.label?('UnknownTransactionCommitResult') if Time.now >= deadline || e.is_a?(Error::OperationFailure) && e.max_time_ms_expired? then transaction_in_progress = false raise end = case v = [:write_concern] when WriteConcern::Base v. when nil {} else v end [:write_concern] = .merge(w: :majority) retry elsif e.label?('TransientTransactionError') if Time.now >= deadline transaction_in_progress = false raise end next else transaction_in_progress = false raise end rescue Error::AuthError transaction_in_progress = false raise end end end # No official return value, but return true so that in interactive # use the method hints that it succeeded. true 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 |