class ActiveRecord::ConnectionAdapters::ConnectionPool

def acquire_connection: (Float checkout_timeout) -> ActiveRecord::ConnectionAdapters::PostgreSQLAdapter

def acquire_connection(checkout_timeout)
  # NOTE: we rely on <tt>@available.poll</tt> and +try_to_checkout_new_connection+ to
  # +conn.lease+ the returned connection (and to do this in a +synchronized+
  # section). This is not the cleanest implementation, as ideally we would
  # <tt>synchronize { conn.lease }</tt> in this method, but by leaving it to <tt>@available.poll</tt>
  # and +try_to_checkout_new_connection+ we can piggyback on +synchronize+ sections
  # of the said methods and avoid an additional +synchronize+ overhead.
  if conn = @available.poll || try_to_checkout_new_connection
    conn
  else
    reap
    @available.poll(checkout_timeout)
  end
end

def active_connection?: () -> ActiveRecord::ConnectionAdapters::PostgreSQLAdapter?

def active_connection?
  @thread_cached_conns[connection_cache_key(current_thread)]
end

def adopt_connection(conn)
  conn.pool = self
  @connections << conn
end

def attempt_to_checkout_all_existing_connections(raise_on_acquisition_timeout = true)
  collected_conns = synchronize do
    # account for our own connections
    @connections.select { |conn| conn.owner == Thread.current }
  end
  newly_checked_out = []
  timeout_time      = Process.clock_gettime(Process::CLOCK_MONOTONIC) + (@checkout_timeout * 2)
  @available.with_a_bias_for(Thread.current) do
    loop do
      synchronize do
        return if collected_conns.size == @connections.size && @now_connecting == 0
        remaining_timeout = timeout_time - Process.clock_gettime(Process::CLOCK_MONOTONIC)
        remaining_timeout = 0 if remaining_timeout < 0
        conn = checkout_for_exclusive_access(remaining_timeout)
        collected_conns   << conn
        newly_checked_out << conn
      end
    end
  end
rescue ExclusiveConnectionTimeoutError
  # <tt>raise_on_acquisition_timeout == false</tt> means we are directed to ignore any
  # timeouts and are expected to just give up: we've obtained as many connections
  # as possible, note that in a case like that we don't return any of the
  # +newly_checked_out+ connections.
  if raise_on_acquisition_timeout
    release_newly_checked_out = true
    raise
  end
rescue Exception # if something else went wrong
  # this can't be a "naked" rescue, because we have should return conns
  # even for non-StandardErrors
  release_newly_checked_out = true
  raise
ensure
  if release_newly_checked_out && newly_checked_out
    # releasing only those conns that were checked out in this method, conns
    # checked outside this method (before it was called) are not for us to release
    newly_checked_out.each { |conn| checkin(conn) }
  end
end

def build_async_executor: () -> nil

def build_async_executor
  case ActiveRecord.async_query_executor
  when :multi_thread_pool
    if @db_config.max_threads > 0
      Concurrent::ThreadPoolExecutor.new(
        min_threads: @db_config.min_threads,
        max_threads: @db_config.max_threads,
        max_queue: @db_config.max_queue,
        fallback_policy: :caller_runs
      )
    end
  when :global_thread_pool
    ActiveRecord.global_thread_pool_async_query_executor
  end
end

def bulk_make_new_connections(num_new_conns_needed)
  num_new_conns_needed.times do
    # try_to_checkout_new_connection will not exceed pool's @size limit
    if new_conn = try_to_checkout_new_connection
      # make the new_conn available to the starving threads stuck @available Queue
      checkin(new_conn)
    end
  end
end

def checkin: (ActiveRecord::ConnectionAdapters::PostgreSQLAdapter conn) -> Thread::ConditionVariable

def checkin(conn)
  conn.lock.synchronize do
    synchronize do
      remove_connection_from_thread_cache conn
      conn._run_checkin_callbacks do
        conn.expire
      end
      @available.add conn
    end
  end
end

def checkout: (?Float checkout_timeout) -> ActiveRecord::ConnectionAdapters::PostgreSQLAdapter

def checkout(checkout_timeout = @checkout_timeout)
  checkout_and_verify(acquire_connection(checkout_timeout))
end

def checkout_and_verify: (ActiveRecord::ConnectionAdapters::PostgreSQLAdapter c) -> ActiveRecord::ConnectionAdapters::PostgreSQLAdapter

def checkout_and_verify(c)
  c._run_checkout_callbacks do
    c.verify!
  end
  c
rescue
  remove c
  c.disconnect!
  raise
end

def checkout_for_exclusive_access(checkout_timeout)
  checkout(checkout_timeout)
rescue ConnectionTimeoutError
  # this block can't be easily moved into attempt_to_checkout_all_existing_connections's
  # rescue block, because doing so would put it outside of synchronize section, without
  # being in a critical section thread_report might become inaccurate
  msg = +"could not obtain ownership of all database connections in #{checkout_timeout} seconds"
  thread_report = []
  @connections.each do |conn|
    unless conn.owner == Thread.current
      thread_report << "#{conn} is owned by #{conn.owner}"
    end
  end
  msg << " (#{thread_report.join(', ')})" if thread_report.any?
  raise ExclusiveConnectionTimeoutError, msg
end

def checkout_new_connection
  raise ConnectionNotEstablished unless @automatic_reconnect
  new_connection
end

def clear_reloadable_connections(raise_on_acquisition_timeout = true)
  with_exclusively_acquired_all_connections(raise_on_acquisition_timeout) do
    synchronize do
      @connections.each do |conn|
        if conn.in_use?
          conn.steal!
          checkin conn
        end
        conn.disconnect! if conn.requires_reloading?
      end
      @connections.delete_if(&:requires_reloading?)
      @available.clear
    end
  end
end

def clear_reloadable_connections!
  clear_reloadable_connections(false)
end

def connected?
  synchronize { @connections.any? }
end

def connection: () -> (ActiveRecord::ConnectionAdapters::PostgreSQLAdapter | ActiveRecord::ConnectionAdapters::SQLite3Adapter)

def connection
  @thread_cached_conns[connection_cache_key(current_thread)] ||= checkout
end

def connection_cache_key: (Thread thread) -> Thread

def connection_cache_key(thread)
  thread
end

def connections
  synchronize { @connections.dup }
end

def current_thread: () -> Thread

def current_thread
  @lock_thread || Thread.current
end

def discard! # :nodoc:
  synchronize do
    return if self.discarded?
    @connections.each do |conn|
      conn.discard!
    end
    @connections = @available = @thread_cached_conns = nil
  end
end

def discarded? # :nodoc:
  @connections.nil?
end

def disconnect(raise_on_acquisition_timeout = true)
  with_exclusively_acquired_all_connections(raise_on_acquisition_timeout) do
    synchronize do
      @connections.each do |conn|
        if conn.in_use?
          conn.steal!
          checkin conn
        end
        conn.disconnect!
      end
      @connections = []
      @available.clear
    end
  end
end

def disconnect!
  disconnect(false)
end

def flush(minimum_idle = @idle_timeout)
  return if minimum_idle.nil?
  idle_connections = synchronize do
    return if self.discarded?
    @connections.select do |conn|
      !conn.in_use? && conn.seconds_idle >= minimum_idle
    end.each do |conn|
      conn.lease
      @available.delete conn
      @connections.delete conn
    end
  end
  idle_connections.each do |conn|
    conn.disconnect!
  end
end

def flush!
  reap
  flush(-1)
end

def initialize: (*unused *) -> void

def initialize(pool_config)
  super()
  @pool_config = pool_config
  @db_config = pool_config.db_config
  @connection_class = pool_config.connection_class
  @role = pool_config.role
  @shard = pool_config.shard
  @checkout_timeout = db_config.checkout_timeout
  @idle_timeout = db_config.idle_timeout
  @size = db_config.pool
  # This variable tracks the cache of threads mapped to reserved connections, with the
  # sole purpose of speeding up the +connection+ method. It is not the authoritative
  # registry of which thread owns which connection. Connection ownership is tracked by
  # the +connection.owner+ attr on each +connection+ instance.
  # The invariant works like this: if there is mapping of <tt>thread => conn</tt>,
  # then that +thread+ does indeed own that +conn+. However, an absence of such
  # mapping does not mean that the +thread+ doesn't own the said connection. In
  # that case +conn.owner+ attr should be consulted.
  # Access and modification of <tt>@thread_cached_conns</tt> does not require
  # synchronization.
  @thread_cached_conns = Concurrent::Map.new(initial_capacity: @size)
  @connections         = []
  @automatic_reconnect = true
  # Connection pool allows for concurrent (outside the main +synchronize+ section)
  # establishment of new connections. This variable tracks the number of threads
  # currently in the process of independently establishing connections to the DB.
  @now_connecting = 0
  @threads_blocking_new_connections = 0
  @available = ConnectionLeasingQueue.new self
  @lock_thread = false
  @async_executor = build_async_executor
  lazily_set_schema_cache
  @reaper = Reaper.new(self, db_config.reaping_frequency)
  @reaper.run
end

def lock_thread=: (bool lock_thread) -> Thread?

def lock_thread=(lock_thread)
  if lock_thread
    @lock_thread = Thread.current
  else
    @lock_thread = nil
  end
end

def new_connection: () -> ActiveRecord::ConnectionAdapters::SQLite3Adapter

def new_connection
  Base.public_send(db_config.adapter_method, db_config.configuration_hash).tap do |conn|
    conn.check_version
  end
end

def num_waiting_in_queue # :nodoc:
  @available.num_waiting
end

def reap
  stale_connections = synchronize do
    return if self.discarded?
    @connections.select do |conn|
      conn.in_use? && !conn.owner.alive?
    end.each do |conn|
      conn.steal!
    end
  end
  stale_connections.each do |conn|
    if conn.active?
      conn.reset!
      checkin conn
    else
      remove conn
    end
  end
end

def release_connection: (?Thread owner_thread) -> Thread::ConditionVariable

def release_connection(owner_thread = Thread.current)
  if conn = @thread_cached_conns.delete(connection_cache_key(owner_thread))
    checkin conn
  end
end

def remove(conn)
  needs_new_connection = false
  synchronize do
    remove_connection_from_thread_cache conn
    @connections.delete conn
    @available.delete conn
    # @available.any_waiting? => true means that prior to removing this
    # conn, the pool was at its max size (@connections.size == @size).
    # This would mean that any threads stuck waiting in the queue wouldn't
    # know they could checkout_new_connection, so let's do it for them.
    # Because condition-wait loop is encapsulated in the Queue class
    # (that in turn is oblivious to ConnectionPool implementation), threads
    # that are "stuck" there are helpless. They have no way of creating
    # new connections and are completely reliant on us feeding available
    # connections into the Queue.
    needs_new_connection = @available.any_waiting?
  end
  # This is intentionally done outside of the synchronized section as we
  # would like not to hold the main mutex while checking out new connections.
  # Thus there is some chance that needs_new_connection information is now
  # stale, we can live with that (bulk_make_new_connections will make
  # sure not to exceed the pool's @size limit).
  bulk_make_new_connections(1) if needs_new_connection
end

def remove_connection_from_thread_cache: (ActiveRecord::ConnectionAdapters::PostgreSQLAdapter conn, ?Thread owner_thread) -> false

def remove_connection_from_thread_cache(conn, owner_thread = conn.owner)
  @thread_cached_conns.delete_pair(connection_cache_key(owner_thread), conn)
end

def schedule_query(future_result) # :nodoc:
  @async_executor.post { future_result.execute_or_skip }
  Thread.pass
end

def stat
  synchronize do
    {
      size: size,
      connections: @connections.size,
      busy: @connections.count { |c| c.in_use? && c.owner.alive? },
      dead: @connections.count { |c| c.in_use? && !c.owner.alive? },
      idle: @connections.count { |c| !c.in_use? },
      waiting: num_waiting_in_queue,
      checkout_timeout: checkout_timeout
    }
  end
end

def try_to_checkout_new_connection: () -> ActiveRecord::ConnectionAdapters::SQLite3Adapter

def try_to_checkout_new_connection
  # first in synchronized section check if establishing new conns is allowed
  # and increment @now_connecting, to prevent overstepping this pool's @size
  # constraint
  do_checkout = synchronize do
    if @threads_blocking_new_connections.zero? && (@connections.size + @now_connecting) < @size
      @now_connecting += 1
    end
  end
  if do_checkout
    begin
      # if successfully incremented @now_connecting establish new connection
      # outside of synchronized section
      conn = checkout_new_connection
    ensure
      synchronize do
        if conn
          adopt_connection(conn)
          # returned conn needs to be already leased
          conn.lease
        end
        @now_connecting -= 1
      end
    end
  end
end

def with_connection
  unless conn = @thread_cached_conns[connection_cache_key(Thread.current)]
    conn = connection
    fresh_connection = true
  end
  yield conn
ensure
  release_connection if fresh_connection
end

def with_exclusively_acquired_all_connections(raise_on_acquisition_timeout = true)
  with_new_connections_blocked do
    attempt_to_checkout_all_existing_connections(raise_on_acquisition_timeout)
    yield
  end
end

def with_new_connections_blocked
  synchronize do
    @threads_blocking_new_connections += 1
  end
  yield
ensure
  num_new_conns_required = 0
  synchronize do
    @threads_blocking_new_connections -= 1
    if @threads_blocking_new_connections.zero?
      @available.clear
      num_new_conns_required = num_waiting_in_queue
      @connections.each do |conn|
        next if conn.in_use?
        @available.add conn
        num_new_conns_required -= 1
      end
    end
  end
  bulk_make_new_connections(num_new_conns_required) if num_new_conns_required > 0
end