123456789_123456789_123456789_123456789_123456789_

Class: ActiveRecord::ConnectionAdapters::RactorConnectionHandler::ProxyConnectionPool

Do not use. This class is for internal use only.
Relationships & Source Files
Namespace Children
Classes:
Lease, SchemaCacheProxy, State
Super Chains via Extension / Inclusion / Inheritance
Instance Chain:
self, Proxy
Inherits: Object
Defined in: activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb

Class Method Summary

Instance Attribute Summary

Instance Method Summary

Proxy - Included

#boundary_safe_bind

Attribute types may close over procs, so the database value is resolved locally.
#checkin_all_connections

For supervisors tearing down worker Ractors, which cannot release their own tokens when they die.
#checkin_token, #connection_pinned?, #discard_token, #dump_binds, #dump_column_types, #dump_object,
#main_operation

Proxy work to the main ractor.
#remove_token

Constructor Details

.new(spec) ⇒ ProxyConnectionPool

[ GitHub ] 

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 39



def initialize(spec)
  @db_config = spec.fetch(:db_config)
  @connection_name = spec.fetch(:connection_name).to_s.freeze
  @role = spec.fetch(:role)
  @shard = spec.fetch(:shard)
  @pool_token = spec.fetch(:pool_token)
  @key = [@connection_name, @role, @shard, @pool_token].freeze
end


  







Class Method Details



  

    .for_spec(spec)  
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 33



def self.for_spec(spec)
  pools = (ActiveSupport::Ractors[:active_record_ractor_pools] ||= Concurrent::Map.new)
  key = [spec.fetch(:connection_name), spec.fetch(:role), spec.fetch(:shard), spec.fetch(:pool_token)]
  pools.compute_if_absent(key) { new(spec) }
end


  







Instance Attribute Details



  

    #active_connection   (readonly)
  



  

    
Alias for #active_connection?.

  



  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 125



alias :active_connection :active_connection?


  








  

    #connected?  ⇒ Boolean  (readonly)
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 225



def connected?
  main_pool_value(:connected?)
end


  








  

    #db_config   (readonly)
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 31



attr_reader :db_config, :role, :shard, :key


  








  

    #key   (readonly)
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 31



attr_reader :db_config, :role, :shard, :key


  








  

    #permanent_lease?  ⇒ Boolean  (readonly)
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 118



def permanent_lease?
  connection_lease.sticky.nil?
end


  








  

    #pool_transaction_isolation_level   (rw)
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 283



def pool_transaction_isolation_level
  ActiveSupport::IsolatedExecutionState[pool_transaction_isolation_level_key]
end


  








  

    #pool_transaction_isolation_level=(isolation_level)   (rw)
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 287



def pool_transaction_isolation_level=(isolation_level)
  ActiveSupport::IsolatedExecutionState[pool_transaction_isolation_level_key] = isolation_level
end


  








  

    #role   (readonly)
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 31



attr_reader :db_config, :role, :shard, :key


  








  

    #shard   (readonly)
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 31



attr_reader :db_config, :role, :shard, :key


  







Instance Method Details



  

    #==(other)  
    Also known as: #eql?
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 52



def ==(other)
  self.class == other.class && key == other.key
end


  








  

    #active_connection?  ⇒ Boolean  (readonly)
    Also known as: #active_connection
  



  

    
  





  


  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 122



def active_connection?
  connection_lease.connection
end


  








  

    #async_executor  
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 309



def async_executor
  state.async_executor(self)
end


  








  

    #build_async_executor  
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 313



def build_async_executor # :nodoc:
  case ActiveRecord.async_query_executor
  when :multi_thread_pool
    if db_config.max_threads > 0
      Concurrent::ThreadPoolExecutor.new(
        name: "ActiveRecord-#{db_config.name}-ractor-async-query-executor",
        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


  








  

    #checkin(connection)  
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 173



def checkin(connection)
  release_lease(connection)
  return if state.pinned_connection.equal?(connection)

  connection.expire if connection.in_use?
  connection.release_connection
end


  








  

    #checkout(_checkout_timeout = nil)  
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 161



def checkout(_checkout_timeout = nil)
  if pinned = state.pinned_connection
    return pinned
  end

  connection_token, profile = checkout_main_connection
  connection = profile[:proxy_class].new(self, connection_token, profile, db_config.configuration_hash)
  connection.lease
  connection.query_cache = query_cache
  connection
end


  








  

    #checkout_main_connection   (private)
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 402



def checkout_main_connection
  connection_name, role, shard = @connection_name, @role, @shard
  main_operation do
    connection = main_pool(connection_name, role, shard).checkout
    begin
      # A proxied connection is never verified by the query pipeline, so it must be usable up front.
      connection.connect!
      profile = connection.ractor_connection_profile
      token = register_connection(connection)
    rescue Exception
      connection.pool.checkin(connection)
      raise
    end
    ActiveSupport::Ractors.make_shareable([token, profile], copy: true)
  end
end


  








  

    #clear_query_cache  
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 273



def clear_query_cache
  state.query_cache_version.increment if state.pinned_connection
  query_cache.clear
end


  








  

    #clear_reloadable_connections!  
  



  

    
Alias for #disconnect!.

  



  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 233



alias :clear_reloadable_connections! :disconnect!


  








  

    #connection_descriptor  
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 61



def connection_descriptor
  main_pool_value(:connection_descriptor)
end


  








  

    #connection_lease   (private)
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 380



def connection_lease
  leases = (ActiveSupport::IsolatedExecutionState[:active_record_ractor_connection_leases] ||= {})
  lease = leases[key] ||= Lease.new(nil, nil)
  if (connection = lease.connection) && !connection.holds_main_connection?
    lease.connection = nil
    lease.sticky = nil
  end
  lease
end


  








  

    #dirties_query_cache  
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 269



def dirties_query_cache
  query_cache.dirties
end


  








  

    #disable_query_cache(dirties: true)  
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 235



def disable_query_cache(dirties: true)
  cache = query_cache
  old_enabled, cache.enabled, old_dirties, cache.dirties = cache.enabled, false, cache.dirties, dirties
  begin
    yield
  ensure
    cache.enabled, cache.dirties = old_enabled, old_dirties
  end
end


  








  

    #disable_query_cache!  
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 260



def disable_query_cache!
  query_cache.enabled = false
  query_cache.dirties = true
end


  








  

    #disconnect!  
    Also known as: #flush!, #clear_reloadable_connections!
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 229



def disconnect!
  release_connection
end


  








  

    #dispatch_to_main_schema_cache(method_name, args, kwargs)  
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 73



def dispatch_to_main_schema_cache(method_name, args, kwargs) # :nodoc:
  connection_name, role, shard = @connection_name, @role, @shard
  connection_token = connection_lease.connection&.connection_token
  main_args = ActiveSupport::Ractors.make_shareable(args)
  main_kwargs = ActiveSupport::Ractors.make_shareable(kwargs)

  main_operation(connection_pool: self) do
    pool = main_pool(connection_name, role, shard)
    schema_cache =
      if connection_token
        # Bind to the pinned connection so the worker observes its own uncommitted DDL.
        connection = fetch_connection(connection_token)
        # Proxied connections don't connect on demand, so connect explicitly.
        connection.connect! unless connection.connected?
        BoundSchemaReflection.for_lone_connection(pool.schema_reflection, connection)
      else
        pool.schema_cache
      end
    schema_cache.public_send(method_name, *main_args, **main_kwargs)
  end
end


  








  

    #enable_query_cache  
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 245



def enable_query_cache
  cache = query_cache
  old_enabled, cache.enabled, old_dirties, cache.dirties = cache.enabled, true, cache.dirties, true
  begin
    yield
  ensure
    cache.enabled, cache.dirties = old_enabled, old_dirties
  end
end


  








  

    #enable_query_cache!  
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 255



def enable_query_cache!
  query_cache.enabled = true
  query_cache.dirties = true
end


  








  

    #eql?(other)  
  



  

    
Alias for #==.

  



  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 55



alias :eql? :==


  








  

    #flush!  
  



  

    
Alias for #disconnect!.

  



  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 232



alias :flush! :disconnect!


  








  

    #hash  
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 57



def hash
  [self.class, key].hash
end


  








  

    #inspect  
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 48



def inspect
  "#<#{self.class.name} env_name=#{db_config.env_name.inspect} name=#{db_config.name.inspect} role=#{role.inspect} shard=#{shard.inspect}>"
end


  








  

    #internal_metadata  
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 107



def internal_metadata
  InternalMetadata.new(self)
end


  








  

    #lease_connection  
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 111



def lease_connection
  lease = connection_lease
  lease.connection ||= checkout
  lease.sticky = true
  lease.connection
end


  








  

    #main_pool_value(method_name)   (private)
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 440



def main_pool_value(method_name)
  connection_name, role, shard = @connection_name, @role, @shard
  main_operation(connection_pool: self) do
    main_pool(connection_name, role, shard).public_send(method_name)
  end
end


  








  

    #migration_context  
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 95



def migration_context
  MigrationContext.new(migrations_paths, schema_migration, internal_metadata)
end


  








  

    #migrations_paths  
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 99



def migrations_paths
  db_config.migrations_paths || Migrator.migrations_paths
end


  








  

    #pin_connection!(lock_thread)  
  



  

    
Mirrors ConnectionPool#pin_connection!: both pools are pinned so every checkout
yields the same connection; worker-side transactions nest as savepoints.

  



  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 183



def pin_connection!(lock_thread)
  pin_main_connection!(lock_thread)
  state = self.state
  begin
    state.pinned_connection ||= (connection_lease.connection || checkout)
    state.pinned_depth += 1

    connection = state.pinned_connection
    connection.lock_thread = ActiveSupport::IsolatedExecutionState.context if lock_thread
    connection.pinned = true
  rescue Exception
    unpin_main_connection!
    raise
  end
end


  








  

    #pin_main_connection!(lock_thread)   (private)
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 419



def pin_main_connection!(lock_thread)
  connection_name, role, shard = @connection_name, @role, @shard
  pin_lock_thread = !!lock_thread
  main_operation(connection_pool: self) do
    main_pool(connection_name, role, shard).pin_connection!(pin_lock_thread)
    nil
  end
end


  








  

    #pool_transaction_isolation_level_key   (private)
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 398



def pool_transaction_isolation_level_key
  "activerecord_pool_transaction_isolation_level_#{@db_config.name}"
end


  








  

    #query_cache  
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 278



def query_cache
  caches = (ActiveSupport::IsolatedExecutionState[:active_record_ractor_query_caches] ||= {})
  caches[key] ||= QueryCache::Store.new(state.query_cache_version, query_cache_max_size)
end


  








  

    #query_cache_enabled  
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 265



def query_cache_enabled
  query_cache.enabled
end


  








  

    #query_cache_max_size  
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 368



def query_cache_max_size # :nodoc:
  case size = db_config&.query_cache
  when 0, false
    nil
  when Integer
    size
  when nil
    QueryCache::DEFAULT_SIZE
  end
end


  








  

    #release_connection(existing_lease = nil)  
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 127



def release_connection(existing_lease = nil)
  lease = existing_lease || connection_lease
  if connection = lease.connection
    lease.connection = nil
    lease.sticky = nil
    checkin(connection)
    true
  else
    false
  end
end


  








  

    #release_lease(connection)   (private)
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 390



def release_lease(connection)
  lease = connection_lease
  if lease.connection.equal?(connection)
    lease.connection = nil
    lease.sticky = nil
  end
end


  








  

    #remove(connection)  
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 220



def remove(connection)
  release_lease(connection)
  connection.remove_connection
end


  








  

    #schedule_query(future_result)  
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 330



def schedule_query(future_result)
  if executor = async_executor
    executor.post { future_result.execute_or_skip }
    Thread.pass
  else
    future_result.execute_or_skip
  end
end


  








  

    #schema_cache  
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 69



def schema_cache
  state.schema_cache ||= SchemaCacheProxy.new(self)
end


  








  

    #schema_migration  
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 103



def schema_migration
  SchemaMigration.new(self)
end


  








  

    #schema_reflection  
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 65



def schema_reflection
  main_pool_value(:schema_reflection)
end


  








  

    #state  
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 363



def state # :nodoc:
  states = (ActiveSupport::Ractors[:active_record_ractor_pool_states] ||= Concurrent::Map.new)
  states.compute_if_absent(key) { State.new }
end


  








  

    #unpin_connection!  
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 199



def unpin_connection!
  state = self.state
  raise "There isn't a pinned connection #{object_id}" unless state.pinned_connection

  begin
    state.pinned_depth -= 1

    if state.pinned_depth.zero?
      connection = state.pinned_connection
      state.pinned_connection = nil
      connection.pinned = false
      connection.lock_thread = nil
      release_lease(connection)
      connection.expire if connection.in_use?
      connection.release_connection
    end
  ensure
    unpin_main_connection!
  end
end


  








  

    #unpin_main_connection!   (private)
  



  

    
Nothing to unpin when the named pool was replaced or removed (pool token mismatch).

  



  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 429



def unpin_main_connection!
  connection_name, role, shard, pool_token = @connection_name, @role, @shard, @pool_token
  main_operation(connection_pool: self) do
    pool = main_connection_handler.retrieve_connection_pool(
      connection_name, role: role, shard: shard, strict: false
    )
    pool.unpin_connection! if pool && pool.pool_config.pool_token == pool_token
    nil
  end
end


  








  

    #with_connection(prevent_permanent_checkout: false)  
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 139



def with_connection(prevent_permanent_checkout: false)
  lease = connection_lease
  sticky_was = lease.sticky
  lease.sticky = false if prevent_permanent_checkout

  if lease.connection
    begin
      yield lease.connection
    ensure
      lease.sticky = sticky_was if prevent_permanent_checkout && !sticky_was
    end
  else
    begin
      lease.connection = checkout
      yield lease.connection
    ensure
      lease.sticky = sticky_was if prevent_permanent_checkout && !sticky_was
      release_connection(lease) unless lease.sticky
    end
  end
end


  








  

    #with_pool_transaction_isolation_level(isolation_level, transaction_open)  
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy_connection_pool.rb', line 291



def with_pool_transaction_isolation_level(isolation_level, transaction_open)
  if !ActiveRecord.default_transaction_isolation_level.nil?
    begin
      if transaction_open && pool_transaction_isolation_level != ActiveRecord.default_transaction_isolation_level
        raise ActiveRecord::TransactionIsolationError, "cannot set default isolation level while transaction is open"
      end

      old_level = pool_transaction_isolation_level
      self.pool_transaction_isolation_level = isolation_level
      yield
    ensure
      self.pool_transaction_isolation_level = old_level
    end
  else
    yield
  end
end