Class: Mongo::ServerSelector::Base

# File 'lib/mongo/server_selector/base.rb', line 66

attr_reader :max_staleness

# File 'lib/mongo/server_selector/base.rb', line 57

attr_reader :options

# File 'lib/mongo/server_selector/base.rb', line 60

attr_reader :tag_sets

# File 'lib/mongo/server_selector/base.rb', line 379

def candidates(cluster, deprioritized = [])
  servers = cluster.servers.reject { |s| deprioritized.include?(s) }
  if cluster.single?
    servers
  elsif cluster.sharded?
    servers
  elsif cluster.replica_set?
    select_in_replica_set(servers)
  else
    # Unknown cluster - no servers
    []
  end
end

# File 'lib/mongo/server_selector/base.rb', line 556

def filter_stale_servers(candidates, primary = nil)
  return candidates unless @max_staleness

  # last_scan is filled out by the Monitor, and can be nil if a server
  # had its description manually set rather than being normally updated
  # via the SDAM flow. We don't handle the possibility of a nil
  # last_scan here.
  if primary
    candidates.select do |server|
      staleness = (server.last_scan - server.last_write_date) -
                  (primary.last_scan - primary.last_write_date) +
                  server.cluster.heartbeat_interval
      staleness <= @max_staleness
    end
  else
    max_write_date = candidates.collect(&:last_write_date).max
    candidates.select do |server|
      staleness = max_write_date - server.last_write_date + server.cluster.heartbeat_interval
      staleness <= @max_staleness
    end
  end
end

# File 'lib/mongo/server_selector/base.rb', line 453

def full_doc
  @full_doc ||= begin
    preference = { mode: self.class.const_get(:SERVER_FORMATTED_NAME) }
    preference.update(tags: tag_sets) unless tag_sets.empty?
    preference.update(maxStalenessSeconds: max_staleness) if max_staleness
    preference.update(hedge: hedge) if hedge
    preference
  end
end

# File 'lib/mongo/server_selector/base.rb', line 116

def inspect
  "#<#{self.class.name}:0x#{object_id} tag_sets=#{tag_sets.inspect} max_staleness=#{max_staleness.inspect} hedge=#{hedge}>"
end

# File 'lib/mongo/server_selector/base.rb', line 99

def local_threshold
  @local_threshold ||= options[:local_threshold] || ServerSelector::LOCAL_THRESHOLD
end

# File 'lib/mongo/server_selector/base.rb', line 104

def local_threshold_with_cluster(cluster)
  options[:local_threshold] || cluster.options[:local_threshold] || LOCAL_THRESHOLD
end

# File 'lib/mongo/server_selector/base.rb', line 547

def match_tag_sets(candidates)
  matches = []
  tag_sets.find do |tag_set|
    matches = candidates.select { |server| server.matches_tag_set?(tag_set) }
    !matches.empty?
  end
  matches || []
end

# File 'lib/mongo/server_selector/base.rb', line 507

def near_servers(candidates = [], local_threshold = nil)
  return candidates if candidates.empty?

  # Average RTT on any server may change at any time by the server
  # monitor's background thread. ARTT may also become nil if the
  # server is marked unknown. Take a snapshot of ARTTs for the duration
  # of this method.

  candidates = candidates.map do |server|
    { server: server, artt: server.average_round_trip_time }
  end.reject do |candidate|
    candidate[:artt].nil?
  end

  return candidates if candidates.empty?

  nearest_candidate = candidates.min_by do |candidate|
    candidate[:artt]
  end

  # Default for legacy signarure
  local_threshold ||= self.local_threshold

  threshold = nearest_candidate[:artt] + local_threshold

  candidates.select do |candidate|
    candidate[:artt] <= threshold
  end.map do |candidate|
    candidate[:server]
  end.shuffle!
end

# File 'lib/mongo/server_selector/base.rb', line 471

def primary(candidates)
  candidates.select do |server|
    server.primary?
  end
end

# File 'lib/mongo/server_selector/base.rb', line 485

def secondaries(candidates)
  matching_servers = candidates.select(&:secondary?)
  matching_servers = filter_stale_servers(matching_servers, primary(candidates).first)
  matching_servers = match_tag_sets(matching_servers) unless tag_sets.empty?
  # Per server selection spec the server selected MUST be a random
  # one matching staleness and latency requirements.
  # Selectors always pass the output of #secondaries to #nearest
  # which shuffles the server list, fulfilling this requirement.
  matching_servers
end

# File 'lib/mongo/server_selector/base.rb', line 176

def select_server(
  cluster,
  ping = nil,
  session = nil,
  write_aggregation: false,
  deprioritized: [],
  timeout: nil
)
  select_server_impl(cluster, ping, session, write_aggregation, deprioritized, timeout).tap do |server|
    if Lint.enabled? && !server.pool.ready?
      raise Error::LintError, 'Server selector returning a server with a pool which is not ready'
    end
  end
end

# File 'lib/mongo/server_selector/base.rb', line 193

private def select_server_impl(cluster, _ping, session, write_aggregation, deprioritized, csot_timeout)
  return cluster.servers.first if cluster.topology.is_a?(Cluster::Topology::LoadBalanced)

  timeout = cluster.options[:server_selection_timeout] || SERVER_SELECTION_TIMEOUT

  server_selection_timeout = if csot_timeout && csot_timeout > 0
                               [ timeout, csot_timeout ].min
                             else
                               timeout
                             end

  # Special handling for zero timeout: if we have to select a server,
  # and the timeout is zero, fail immediately (since server selection
  # will take some non-zero amount of time in any case).
  if server_selection_timeout == 0
    msg = 'Failing server selection due to zero timeout. ' +
          " Requested #{name} in cluster: #{cluster.summary}"
    raise Error::NoServerAvailable.new(self, cluster, msg)
  end

  deadline = Utils.monotonic_time + server_selection_timeout

  if session && session.pinned_server
    if Mongo::Lint.enabled? && !cluster.sharded?
      raise Error::LintError, "Session has a pinned server in a non-sharded topology: #{topology}"
    end

    session.unpin unless session.in_transaction?

    if server = session.pinned_server
      # Here we assume that a mongos stays in the topology indefinitely.
      # This will no longer be the case once SRV polling is implemented.

      unless server.mongos?
        while (time_remaining = deadline - Utils.monotonic_time) > 0
          wait_for_server_selection(cluster, time_remaining)
        end

        unless server.mongos?
          msg = "The session being used is pinned to the server which is not a mongos: #{server.summary} " +
                "(after #{server_selection_timeout} seconds)"
          raise Error::NoServerAvailable.new(self, cluster, msg)
        end
      end

      return server
    end
  end

  validate_max_staleness_value_early! if cluster.replica_set?

  if cluster.addresses.empty?
    if Lint.enabled? && !cluster.servers.empty?
      raise Error::LintError,
            "Cluster has no addresses but has servers: #{cluster.servers.map(&:inspect).join(', ')}"
    end
    msg = 'Cluster has no addresses, and therefore will never have a server'
    raise Error::NoServerAvailable.new(self, cluster, msg)
  end

  # Add this check in version 3.0.0
  #         unless cluster.connected?
  #           msg = 'Cluster is disconnected'
  #           raise Error::NoServerAvailable.new(self, cluster, msg)
  #         end

  loop do
    if Lint.enabled?
      cluster.servers.each do |server|
        # TODO: Add this back in RUBY-3174.
        # if !server.unknown? && !server.connected?
        #   raise Error::LintError, "Server #{server.summary} is known but is not connected"
        # end
        if !server.unknown? && !server.pool.ready?
          raise Error::LintError, "Server #{server.summary} is known but has non-ready pool"
        end
      end
    end

    server = try_select_server(cluster, write_aggregation: write_aggregation, deprioritized: deprioritized)
    if server.nil? && deprioritized.any?
      server = try_select_server(cluster, write_aggregation: write_aggregation, deprioritized: [])
    end

    if server
      unless cluster.topology.compatible?
        raise Error::UnsupportedFeatures, cluster.topology.compatibility_error.to_s
      end

      session.pin_to_server(server) if session && session.starting_transaction? && cluster.sharded?

      return server
    end

    cluster.scan!(false)

    time_remaining = deadline - Utils.monotonic_time
    break unless time_remaining > 0

    wait_for_server_selection(cluster, time_remaining)

    # If we wait for server selection, perform another round of
    # attempting to locate a suitable server. Otherwise server selection
    # can raise NoServerAvailable message when the diagnostics
    # reports an available server of the requested type.
  end

  msg = "No #{name} server"
  msg += " with tag sets: #{tag_sets}" if is_a?(ServerSelector::Secondary) && !tag_sets.empty?
  msg += " is available in cluster: #{cluster.summary} " +
         "with timeout=#{server_selection_timeout}, " +
         "LT=#{local_threshold_with_cluster(cluster)}"
  msg += server_selection_diagnostic_message(cluster)
  raise Error::NoServerAvailable.new(self, cluster, msg)
rescue Error::NoServerAvailable => e
  if session && session.in_transaction? && !session.committing_transaction?
    e.add_label('TransientTransactionError')
  end
  e.add_label('UnknownTransactionCommitResult') if session && session.committing_transaction?
  raise e
end

# File 'lib/mongo/server_selector/base.rb', line 668

def server_selection_diagnostic_message(cluster)
  msg = ''
  dead_monitors = []
  cluster.servers_list.each do |server|
    thread = server.monitor.instance_variable_get(:@thread)
    dead_monitors << server if thread.nil? || !thread.alive?
  end
  if dead_monitors.any?
    msg += ". The following servers have dead monitor threads: #{dead_monitors.map(&:summary).join(', ')}"
  end
  msg += '. The cluster is disconnected (client may have been closed)' unless cluster.connected?
  msg
end

# File 'lib/mongo/server_selector/base.rb', line 83

def server_selection_timeout
  @server_selection_timeout ||=
    options[:server_selection_timeout] || ServerSelector::SERVER_SELECTION_TIMEOUT
end

# File 'lib/mongo/server_selector/base.rb', line 443

def suitable_server(servers)
  servers.first
end

# File 'lib/mongo/server_selector/base.rb', line 402

def suitable_servers(cluster, deprioritized = [])
  result = suitable_servers_impl(cluster, deprioritized)
  result = suitable_servers_impl(cluster, []) if result.empty? && deprioritized.any?
  result
end

# File 'lib/mongo/server_selector/base.rb', line 420

def suitable_servers_impl(cluster, deprioritized)
  if cluster.single?
    candidates(cluster, deprioritized)
  elsif cluster.sharded?
    local_threshold = local_threshold_with_cluster(cluster)
    servers = candidates(cluster, deprioritized)
    near_servers(servers, local_threshold)
  elsif cluster.replica_set?
    validate_max_staleness_value!(cluster)
    candidates(cluster, deprioritized)
  else
    # Unknown cluster - no servers
    []
  end
end

# File 'lib/mongo/server_selector/base.rb', line 330

def try_select_server(cluster, write_aggregation: false, deprioritized: [])
  servers = if write_aggregation && cluster.replica_set?
              # 1. Check if ALL servers in cluster support secondary writes.
              is_write_supported = cluster.servers.reduce(true) do |res, server|
                res && server.features.merge_out_on_secondary_enabled?
              end

              if is_write_supported
                # 2. If all servers support secondary writes, we respect read preference.
                suitable_servers(cluster, deprioritized)
              else
                # 3. Otherwise we fallback to primary for replica set.
                [ cluster.servers.detect(&:primary?) ]
              end
            else
              suitable_servers(cluster, deprioritized)
            end

  # This list of servers may be ordered in a specific way
  # by the selector (e.g. for secondary preferred, the first
  # server may be a secondary and the second server may be primary)
  # and we should take the first server here respecting the order
  server = suitable_server(servers)

  if server && Lint.enabled? && server.average_round_trip_time.nil?
    # It is possible for a server to have a nil average RTT here
    # because the ARTT comes from description which may be updated
    # by a background thread while server selection is running.
    # Currently lint mode is not a public feature, if/when this
    # changes (https://jira.mongodb.org/browse/RUBY-1576) the
    # requirement for ARTT to be not nil would need to be removed.
    raise Error::LintError, "Server #{server.address} has nil average rtt"
  end

  server
end

# File 'lib/mongo/server_selector/base.rb', line 579

def validate!
  if !@tag_sets.all? { |set| set.empty? } && !tags_allowed?
    raise Error::InvalidServerPreference.new(Error::InvalidServerPreference::NO_TAG_SUPPORT)
  elsif @max_staleness && !max_staleness_allowed?
    raise Error::InvalidServerPreference.new(Error::InvalidServerPreference::NO_MAX_STALENESS_SUPPORT)
  end

  return unless @hedge
  raise Error::InvalidServerPreference.new(Error::InvalidServerPreference::NO_HEDGE_SUPPORT) unless hedge_allowed?

  unless @hedge.is_a?(Hash) && @hedge.key?(:enabled) &&
         [ true, false ].include?(@hedge[:enabled])
    raise Error::InvalidServerPreference.new(
      "`hedge` value (#{hedge}) is invalid - hedge must be a Hash in the " \
      'format { enabled: true }'
    )
  end
end

# File 'lib/mongo/server_selector/base.rb', line 607

def validate_max_staleness_value!(cluster)
  return unless @max_staleness

  heartbeat_interval = cluster.heartbeat_interval
  unless @max_staleness >= [
    SMALLEST_MAX_STALENESS_SECONDS,
    min_cluster_staleness = heartbeat_interval + Cluster::IDLE_WRITE_PERIOD_SECONDS,
  ].max
    msg = "`max_staleness` value (#{@max_staleness}) is too small - it must be at least " +
          "`Mongo::ServerSelector::SMALLEST_MAX_STALENESS_SECONDS` (#{ServerSelector::SMALLEST_MAX_STALENESS_SECONDS}) and (the cluster's heartbeat_frequency " +
          "setting + `Mongo::Cluster::IDLE_WRITE_PERIOD_SECONDS`) (#{min_cluster_staleness})"
    raise Error::InvalidServerPreference.new(msg)
  end
end

# File 'lib/mongo/server_selector/base.rb', line 598

def validate_max_staleness_value_early!
  return unless @max_staleness
  return if @max_staleness >= SMALLEST_MAX_STALENESS_SECONDS

  msg = "`max_staleness` value (#{@max_staleness}) is too small - it must be at least " +
        "`Mongo::ServerSelector::SMALLEST_MAX_STALENESS_SECONDS` (#{ServerSelector::SMALLEST_MAX_STALENESS_SECONDS})"
  raise Error::InvalidServerPreference.new(msg)
end

# File 'lib/mongo/server_selector/base.rb', line 638

def wait_for_server_selection(cluster, time_remaining)
  if cluster.server_selection_semaphore
    # Since the semaphore may have been signaled between us checking
    # the servers list earlier and the wait call below, we should not
    # wait for the full remaining time - wait for up to 0.5 second, then
    # recheck the state.
    cluster.server_selection_semaphore.wait([ time_remaining, 0.5 ].min)
  else
    if Lint.enabled?
      raise Error::LintError, 'Waiting for server selection without having a server selection semaphore'
    end

    sleep [ time_remaining, 0.25 ].min
  end
end