Class: Mongo::ServerSelector::Base

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

attr_reader :max_staleness

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

attr_reader :options

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

attr_reader :tag_sets

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

def candidates(cluster)
  servers = cluster.servers
  servers.each do |server|
    validate_max_staleness_support!(server)
  end
  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 572

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|
      validate_max_staleness_support!(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|
      validate_max_staleness_support!(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 469

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 122

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 105

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

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

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

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

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 523

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 487

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

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

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 183

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 200

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

  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?
      unless cluster.sharded?
        raise Error::LintError, "Session has a pinned server in a non-sharded topology: #{topology}"
      end
    end

    if !session.in_transaction?
      session.unpin
    end

    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

  if cluster.replica_set?
    validate_max_staleness_value_early!
  end

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

=begin Add this check in version 3.0.0
  unless cluster.connected?
    msg = 'Cluster is disconnected'
    raise Error::NoServerAvailable.new(self, cluster, msg)
  end
=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
      unless cluster.topology.compatible?
        raise Error::UnsupportedFeatures, cluster.topology.compatibility_error.to_s
      end

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

      return server
    end

    cluster.scan!(false)

    time_remaining = deadline - Utils.monotonic_time
    if 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.
    else
      break
    end
  end

  msg = "No #{name} server"
  if is_a?(ServerSelector::Secondary) && !tag_sets.empty?
    msg += " with tag sets: #{tag_sets}"
  end
  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
  if session && session.committing_transaction?
    e.add_label('UnknownTransactionCommitResult')
  end
  raise e
end

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

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

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

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

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

def suitable_server(servers, deprioritized)
  preferred = servers - deprioritized
  if preferred.empty?
    servers.first
  else
    preferred.first
  end
end

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

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

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

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)
    else
      # 3. Otherwise we fallback to primary for replica set.
      [cluster.servers.detect(&:primary?)]
    end
  else
    suitable_servers(cluster)
  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, deprioritized)

  if server
    if Lint.enabled?
      # 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.
      if server.average_round_trip_time.nil?
        raise Error::LintError, "Server #{server.address} has nil average rtt"
      end
    end
  end

  server
end

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

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

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

    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
end

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

def validate_max_staleness_support!(server)
  if @max_staleness && !server.features.max_staleness_enabled?
    raise Error::InvalidServerPreference.new(Error::InvalidServerPreference::NO_MAX_STALENESS_WITH_LEGACY_SERVER)
  end
end

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

def validate_max_staleness_value!(cluster)
  if @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
end

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

def validate_max_staleness_value_early!
  if @max_staleness
    unless @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
  end
end

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

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