123456789_123456789_123456789_123456789_123456789_

Module: ActiveRecord::ConnectionAdapters::RactorConnectionHandler::Proxy

Do not use. This module is for internal use only.

Overview

The boundary between worker Ractors and the main Ractor, which owns the real connection handler, pools, and connections.

Class Attribute Summary

Class Method Summary

Class Attribute Details

.connections (readonly)

[ GitHub ] 

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy.rb', line 189



attr_reader :connections


  







Class Method Details



  

    .boundary_safe_bind(bind)   (mod_func)
  



  

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

  



  [ GitHub ]


  

  


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



def boundary_safe_bind(bind)
  return bind unless bind.is_a?(ActiveModel::Attribute)
  return bind if bind.value_before_type_cast.is_a?(StatementCache::Substitute)

  safe = Relation::QueryAttribute.new(bind.name, bind.value_for_database, ActiveModel::Type.default_value)
  safe.value_for_database # resolve the memo so a frozen copy never mutates
  safe
end


  








  

    .capture_transport_errors  
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy.rb', line 191



def capture_transport_errors
  yield
rescue => error
  ErrorResponse.new(error)
end


  








  

    .checkin_all_connections   (mod_func)
  



  

    
For supervisors tearing down worker Ractors, which cannot release their own tokens when they die.

  



  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy.rb', line 164



def checkin_all_connections
  main_operation do
    @connections_lock.synchronize do
      @connections.each_value do |connection|
        reclaim(connection)
        connection.pool.checkin(connection) if connection.in_use?
      end
      @connections.clear
    end
    nil
  end
end


  








  

    .checkin_token(connection_token)   (mod_func)
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy.rb', line 134



def checkin_token(connection_token)
  main_operation do
    if connection = take_back_connection(connection_token)
      connection.pool.checkin(connection) if connection.in_use?
    end
    nil
  end
end


  








  

    .connection_pinned?(connection_token)  ⇒ Boolean  (mod_func)
  



  

    
  





  


  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy.rb', line 177



def connection_pinned?(connection_token)
  main_operation do
    connection = @connections_lock.synchronize { @connections[connection_token] }
    !!(connection && connection.in_use?)
  end
end


  








  

    .discard_token(connection_token)   (mod_func)
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy.rb', line 153



def discard_token(connection_token)
  main_operation do
    if connection = take_back_connection(connection_token)
      connection.pool.remove(connection)
      connection.discard!
    end
    nil
  end
end


  








  

    .dump_binds(binds)   (mod_func)
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy.rb', line 94



def dump_binds(binds)
  return nil if binds.nil? || binds.empty?

  dump_object(binds.map { |bind| boundary_safe_bind(bind) }, "bind parameters")
end


  








  

    .dump_column_types(result)   (mod_func)
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy.rb', line 100



def dump_column_types(result)
  types = result.columns.map { |name| result.column_types[name] }
  return nil if types.all?(&:nil?)

  begin
    Marshal.dump(types).freeze
  rescue TypeError
    # Result falls back to Type.default_value for nil entries.
    safe_types = types.map do |type|
      Marshal.dump(type)
      type
    rescue TypeError
      nil
    end
    Marshal.dump(safe_types).freeze
  end
end


  








  

    .dump_object(value, description)   (mod_func)
  

  [ GitHub ]


  

  


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



def dump_object(value, description)
  Marshal.dump(value).freeze
rescue TypeError => error
  raise ActiveRecordError, "Cannot send #{description} across the Ractor boundary: #{error.message}"
end


  








  

    .fetch_connection(connection_token)  
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy.rb', line 219



def fetch_connection(connection_token)
  connection = @connections_lock.synchronize { @connections[connection_token] }
  unless connection
    raise ConnectionNotEstablished, "The Ractor-pinned connection for token #{connection_token.inspect} has been released"
  end
  connection
end


  








  

    .forget_all_connections!  
  



  

    
After a fork, the inherited tokens name discarded connections and must not be checked back in.

  



  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy.rb', line 228



def forget_all_connections!
  @connections_lock.synchronize { @connections.clear }
end


  








  

    .main_connection_handler  
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy.rb', line 197



def main_connection_handler
  ActiveRecord::Base.default_connection_handler
end


  








  

    .main_operation(connection_pool: nil, &block)   (mod_func)
  



  

    
Proxy work to the main ractor. A worker's block may only capture shareable objects.

  



  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy.rb', line 85



def main_operation(connection_pool: nil, &block)
  operation = ActiveSupport::Ractors.shareable_proc(self: Proxy, &block)
  outcome = ActiveSupport::Ractors.on_main do
    Proxy.capture_transport_errors { operation.call }
  end
  raise outcome.exception(connection_pool: connection_pool) if outcome.is_a?(ErrorResponse)
  outcome
end


  








  

    .main_pool(connection_name, role, shard)  
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy.rb', line 201



def main_pool(connection_name, role, shard)
  main_connection_handler.retrieve_connection_pool(
    connection_name,
    role: role,
    shard: shard,
    strict: true,
  )
end


  








  

    .reclaim(connection)   (private)
  



  

    
Token-pinned connections are leased on the dispatch thread

  



  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy.rb', line 241



def reclaim(connection)
  connection.steal! if connection.in_use?
end


  








  

    .register_connection(connection)  
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy.rb', line 210



def register_connection(connection)
  connection.proxied = true
  @connections_lock.synchronize do
    token = (@next_token += 1)
    @connections[token] = connection
    token
  end
end


  








  

    .remove_token(connection_token)   (mod_func)
  

  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/ractor_connection_handler/proxy.rb', line 143



def remove_token(connection_token)
  main_operation do
    if connection = take_back_connection(connection_token)
      connection.pool.remove(connection)
      connection.disconnect!
    end
    nil
  end
end


  








  

    .take_back_connection(connection_token)   (private)
  

  [ GitHub ]


  

  


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



def take_back_connection(connection_token)
  if connection = @connections_lock.synchronize { @connections.delete(connection_token) }
    reclaim(connection)
    connection
  end
end