Class: Mongo::Session

# File 'lib/mongo/session.rb', line 1239

def causal_consistency?
  @causal_consistency ||= (if @options.key?(:causal_consistency)
                             !!@options[:causal_consistency]
                           else
                             true
                           end)
end

# File 'lib/mongo/session.rb', line 116

attr_reader :client

# File 'lib/mongo/session.rb', line 811

def committing_transaction?
  !!@committing_transaction
end

# File 'lib/mongo/session.rb', line 146

def dirty?
  @server_session&.dirty?
end

# File 'lib/mongo/session.rb', line 243

def ended?
  !!@ended
end

# File 'lib/mongo/session.rb', line 178

def explicit?
  !implicit?
end

# File 'lib/mongo/session.rb', line 166

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

# File 'lib/mongo/session.rb', line 803

def in_transaction?
  within_states?(STARTING_TRANSACTION_STATE, TRANSACTION_IN_PROGRESS_STATE)
end

# File 'lib/mongo/session.rb', line 1150

def materialized?
  if ended?
    raise Error::SessionEnded
  end

  !@server_session.nil?
end

# File 'lib/mongo/session.rb', line 131

attr_reader :operation_time

# File 'lib/mongo/session.rb', line 111

attr_reader :options

# File 'lib/mongo/session.rb', line 284

attr_reader :pinned_connection_global_id

# File 'lib/mongo/session.rb', line 278

attr_reader :pinned_server

# File 'lib/mongo/session.rb', line 290

attr_accessor :recovery_token

# File 'lib/mongo/session.rb', line 198

def retry_reads?
  client.options[:retry_reads] != false
end

# File 'lib/mongo/session.rb', line 213

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

# File 'lib/mongo/session.rb', line 124

def snapshot?
  !!options[:snapshot]
end

# File 'lib/mongo/session.rb', line 1192

attr_accessor :snapshot_timestamp

# File 'lib/mongo/session.rb', line 791

def starting_transaction?
  within_states?(STARTING_TRANSACTION_STATE)
end

# File 'lib/mongo/session.rb', line 1194

attr_reader :with_transaction_deadline

# File 'lib/mongo/session.rb', line 899

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

# File 'lib/mongo/session.rb', line 914

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

# File 'lib/mongo/session.rb', line 931

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

# File 'lib/mongo/session.rb', line 946

def add_txn_opts!(command, read, context)
  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 = txn_options[: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

    # Ignore wtimeout if csot
    if context&.csot?
      c[:writeConcern]&.delete(:wtimeout)
    end

    # We must not send an empty (server default) write concern.
    c.delete(:writeConcern) if c[:writeConcern]&.empty?
  end
end

# File 'lib/mongo/session.rb', line 1123

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

# File 'lib/mongo/session.rb', line 1231

def causal_consistency_doc
  if operation_time && causal_consistency?
    {:afterClusterTime => operation_time}
  else
    nil
  end
end

# File 'lib/mongo/session.rb', line 1253

def check_if_ended!
  raise Mongo::Error::InvalidSession.new(SESSION_ENDED_ERROR_MSG) if ended?
end

# File 'lib/mongo/session.rb', line 1217

def check_if_no_transaction!
  return unless within_states?(NO_TRANSACTION_STATE)

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

# File 'lib/mongo/session.rb', line 1257

def check_matching_cluster!(client)
  if @client.cluster != client.cluster
    raise Mongo::Error::InvalidSession.new(MISMATCHED_CLUSTER_ERROR_MSG)
  end
end

# File 'lib/mongo/session.rb', line 1263

def check_transactions_supported!
  raise Mongo::Error::TransactionsNotSupported, "standalone topology" if cluster.single?

  cluster.next_primary.with_connection do |conn|
    if cluster.replica_set? && !conn.features.transactions_enabled?
      raise Mongo::Error::TransactionsNotSupported, "server version is < 4.0"
    end
    if cluster.sharded? && !conn.features.sharded_transactions_enabled?
      raise Mongo::Error::TransactionsNotSupported, "sharded transactions require server version >= 4.2"
    end
  end
end

# File 'lib/mongo/session.rb', line 118

def cluster
  @client.cluster
end

# File 'lib/mongo/session.rb', line 648

def commit_transaction(options=nil)
  QueryCache.clear
  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
      @committing_transaction = true

      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

      context = Operation::Context.new(
        client: @client,
        session: self,
        operation_timeouts: operation_timeouts(options)
      )
      write_with_retry(write_concern, ending_transaction: true,
        context: context,
      ) do |connection, txn_num, context|
        if context.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
        spec = {
          selector: { commitTransaction: 1 },
          db_name: 'admin',
          session: self,
          txn_num: txn_num,
          write_concern: write_concern,
        }
        Operation::Command.new(spec).execute_with_connection(connection, context: context)
      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

# File 'lib/mongo/session.rb', line 138

def dirty!(mark = true)
  @server_session&.dirty!(mark)
end

# File 'lib/mongo/session.rb', line 377

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

# File 'lib/mongo/session.rb', line 355

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

# File 'lib/mongo/session.rb', line 1137

def materialize_if_needed
  if ended?
    raise Error::SessionEnded
  end

  return unless implicit? && !@server_session

  @server_session = cluster.session_pool.checkout

  self
end

# File 'lib/mongo/session.rb', line 1167

def next_txn_num
  if ended?
    raise Error::SessionEnded
  end

  @server_session.next_txn_num
end

# File 'lib/mongo/session.rb', line 1276

def operation_timeouts(opts)
  {
    inherited_timeout_ms: @client.timeout_ms
  }.tap do |result|
    if @with_transaction_deadline.nil?
      if timeout_ms = opts[:timeout_ms]
        result[:operation_timeout_ms] = timeout_ms
      elsif default_timeout_ms = options[:default_timeout_ms]
        result[:operation_timeout_ms] = default_timeout_ms
      end
    end
  end
end

# File 'lib/mongo/session.rb', line 846

def pin_to_connection(connection_global_id)
  if connection_global_id.nil?
    raise ArgumentError, 'Cannot pin to a nil connection id'
  end
  @pinned_connection_global_id = connection_global_id
end

# File 'lib/mongo/session.rb', line 828

def pin_to_server(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

# File 'lib/mongo/session.rb', line 1095

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

# File 'lib/mongo/session.rb', line 255

def session_id
  if ended?
    raise Error::SessionEnded
  end

  # An explicit session will always have a session_id, because during
  # construction a server session must be provided. An implicit session
  # will not have a session_id until materialized, thus calls to
  # session_id might fail. An application should not have an opportunity
  # to experience this failure because an implicit session shouldn't be
  # accessible to applications due to its lifetime being constrained to
  # operation execution, which is done entirely by the driver.
  unless materialized?
    raise Error::SessionNotMaterialized
  end

  @server_session.session_id
end

# File 'lib/mongo/session.rb', line 1247

def set_operation_time(result)
  if result && result.operation_time
    @operation_time = result.operation_time
  end
end

# File 'lib/mongo/session.rb', line 583

def start_transaction(options = nil)
  check_transactions_supported!

  if options
    Lint.validate_read_concern_option(options[: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

  if snapshot?
    raise Mongo::Error::SnapshotSessionTransactionProhibited
  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
  @txn_options = (@options[:default_transaction_options] || {}).merge(options || {})

  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

# File 'lib/mongo/session.rb', line 1020

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

# File 'lib/mongo/session.rb', line 1183

def txn_num
  if ended?
    raise Error::SessionEnded
  end

  @server_session.txn_num
end

# File 'lib/mongo/session.rb', line 154

def txn_options
  @txn_options or raise ArgumentError, "There is no active transaction"
end

# File 'lib/mongo/session.rb', line 1208

def txn_read_concern
  # Read concern is inherited from client but not db or collection.
  txn_options[:read_concern] || @client.read_concern
end

# File 'lib/mongo/session.rb', line 228

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

# File 'lib/mongo/session.rb', line 1224

def txn_write_concern
  txn_options[:write_concern] ||
    (@client.write_concern && @client.write_concern.options)
end

# File 'lib/mongo/session.rb', line 859

def unpin(connection = nil)
  @pinned_server = nil
  @pinned_connection_global_id = nil
  connection.unpin unless connection.nil?
end

# File 'lib/mongo/session.rb', line 876

def unpin_maybe(error, connection = nil)
  if !within_states?(Session::NO_TRANSACTION_STATE) &&
    error.label?('TransientTransactionError')
  then
    unpin(connection)
  end

  if committing_transaction? &&
    error.label?('UnknownTransactionCommitResult')
  then
    unpin(connection)
  end
end

# File 'lib/mongo/session.rb', line 1056

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

# File 'lib/mongo/session.rb', line 1078

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

# File 'lib/mongo/session.rb', line 1039

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

# File 'lib/mongo/session.rb', line 448

def with_transaction(options = nil)
  if timeout_ms = (options || {})[:timeout_ms]
    timeout_sec = timeout_ms / 1_000.0
    deadline = Utils.monotonic_time + timeout_sec
    @with_transaction_deadline = deadline
  elsif default_timeout_ms = @options[:default_timeout_ms]
    timeout_sec = default_timeout_ms / 1_000.0
    deadline = Utils.monotonic_time + timeout_sec
    @with_transaction_deadline = deadline
  elsif @client.timeout_sec
    deadline = Utils.monotonic_time + @client.timeout_sec
    @with_transaction_deadline = deadline
  else
    deadline = Utils.monotonic_time + 120
  end
  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)
        log_warn("Aborting transaction due to #{e.class}: #{e}")
        @with_transaction_deadline = nil
        abort_transaction
        transaction_in_progress = false
      end

      if Utils.monotonic_time >= 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(commit_options)
        transaction_in_progress = false
        return rv
      rescue Mongo::Error => e
        if e.label?('UnknownTransactionCommitResult')
          if Utils.monotonic_time >= deadline ||
            e.is_a?(Error::OperationFailure::Family) && e.max_time_ms_expired?
          then
            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?('TransientTransactionError')
          if Utils.monotonic_time >= deadline
            transaction_in_progress = false
            raise
          end
          @state = NO_TRANSACTION_STATE
          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 broke out of with_transaction loop, aborting transaction')
    begin
      abort_transaction
    rescue Error::OperationFailure::Family, Error::InvalidTransactionOperation
    end
  end
  @with_transaction_deadline = nil
end

# File 'lib/mongo/session.rb', line 1213

def within_states?(*states)
  states.include?(@state)
end