Class: Mongo::Session

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

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

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

attr_reader :client

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

def committing_transaction?
  !!@committing_transaction
end

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

def dirty?
  @server_session&.dirty?
end

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

def ended?
  !!@ended
end

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

def explicit?
  !implicit?
end

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

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

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

def in_transaction?
  within_states?(STARTING_TRANSACTION_STATE, TRANSACTION_IN_PROGRESS_STATE)
end

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

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

  !@server_session.nil?
end

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

attr_reader :operation_time

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

attr_reader :options

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

attr_reader :pinned_connection_global_id

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

attr_reader :pinned_server

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

attr_accessor :recovery_token

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

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

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

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

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

def snapshot?
  !!options[:snapshot]
end

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

attr_accessor :snapshot_timestamp

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

def starting_transaction?
  within_states?(STARTING_TRANSACTION_STATE)
end

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

attr_reader :with_transaction_deadline

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

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

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

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

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

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 974

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 1151

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 1341

def backoff_seconds_for_retry(transaction_attempt)
  exponential = BACKOFF_INITIAL * (1.5 ** (transaction_attempt - 1))
  Random.rand * [exponential, BACKOFF_MAX].min
end

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

def backoff_would_exceed_deadline?(deadline, backoff_seconds)
  return false if deadline.zero?

  Utils.monotonic_time + backoff_seconds >= deadline
end

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

def calculate_with_transaction_deadline(opts)
  calc = -> (timeout) {
    if timeout == 0
      0
    else
      Utils.monotonic_time + (timeout / 1000.0)
    end
  }
  if timeout_ms = opts&.dig(:timeout_ms)
    calc.call(timeout_ms)
  elsif default_timeout_ms = @options[:default_timeout_ms]
    calc.call(default_timeout_ms)
  elsif @client.timeout_ms
    calc.call(@client.timeout_ms)
  end
end

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

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

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

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

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

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 1287

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

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

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

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

def cluster
  @cluster
end

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

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 = Operation::Command.new(spec)
        tracer.trace_operation(operation, context, op_name: 'commitTransaction') do
          operation.execute_with_connection(connection, context: context)
        end
      end
    end
    # Finish the transaction span before changing state
    tracer.finish_transaction_span(self)
  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 1328

def deadline_expired?(deadline)
  if deadline.zero?
    false
  else
    Utils.monotonic_time >= deadline
  end
end

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

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

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

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
      cluster.session_pool.checkin(@server_session)
    end
  end
ensure
  @server_session = nil
  @ended = true
  @client = nil
end

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

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

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

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 1195

def next_txn_num
  if ended?
    raise Error::SessionEnded
  end

  @server_session.next_txn_num
end

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

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 874

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 856

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 1123

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 260

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 1277

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

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

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
  tracer.start_transaction_span(self)

  # 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 1048

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 1211

def txn_num
  if ended?
    raise Error::SessionEnded
  end

  @server_session.txn_num
end

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

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

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

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 233

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 1254

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

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

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

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

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 1084

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 1106

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

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

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 454

def with_transaction(options = nil)
  @with_transaction_deadline = calculate_with_transaction_deadline(options)
  deadline = if @with_transaction_deadline
               # CSOT enabled, so we have a customer defined deadline.
               @with_transaction_deadline
             else
                # CSOT not enabled, so we use the default deadline, 120 seconds.
               Utils.monotonic_time + 120
             end
  transaction_in_progress = false
  transaction_attempt = 0
  last_error = nil

  loop do
    if transaction_attempt > 0
      backoff = backoff_seconds_for_retry(transaction_attempt)
      if backoff_would_exceed_deadline?(deadline, backoff)
        raise(last_error)
      end
      sleep(backoff)
    end

    commit_options = {}
    if options
      commit_options[:write_concern] = options[:write_concern]
    end
    start_transaction(options)
    transaction_in_progress = true
    transaction_attempt += 1

    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 deadline_expired?(deadline)
        transaction_in_progress = false
        raise
      end

      if e.is_a?(Mongo::Error) && e.label?('TransientTransactionError')
        last_error = e
        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 deadline_expired?(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
          last_error = e
          @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 1243

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