Merge pull request #31173 from matthewd/connection-fork-safety

Improve AR connection fork safety

activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb

@@ -447,6 +447,21 @@ def disconnect!
         disconnect(false)
       end
 
+      # Discards all connections in the pool (even if they're currently
+      # leased!), along with the pool itself. Any further interaction with the
+      # pool (except #spec and #schema_cache) is undefined.
+      #
+      # See AbstractAdapter#discard!
+      def discard! # :nodoc:
+        synchronize do
+          return if @connections.nil? # already discarded
+          @connections.each do |conn|
+            conn.discard!
+          end
+          @connections = @available = @thread_cached_conns = nil
+        end
+      end
+
       # Clears the cache which maps classes and re-connects connections that
       # require reloading.
       #
@@ -863,11 +878,31 @@ def checkout_and_verify(c)
     # about the model. The model needs to pass a specification name to the handler,
     # in order to look up the correct connection pool.
     class ConnectionHandler
+      def self.unowned_pool_finalizer(pid_map) # :nodoc:
+        lambda do |_|
+          discard_unowned_pools(pid_map)
+        end
+      end
+
+      def self.discard_unowned_pools(pid_map) # :nodoc:
+        pid_map.each do |pid, pools|
+          pools.values.compact.each(&:discard!) unless pid == Process.pid
+        end
+      end
+
       def initialize
         # These caches are keyed by spec.name (ConnectionSpecification#name).
         @owner_to_pool = Concurrent::Map.new(initial_capacity: 2) do |h, k|
+          # Discard the parent's connection pools immediately; we have no need
+          # of them
+          ConnectionHandler.discard_unowned_pools(h)
+
           h[k] = Concurrent::Map.new(initial_capacity: 2)
         end
+
+        # Backup finalizer: if the forked child never needed a pool, the above
+        # early discard has not occurred
+        ObjectSpace.define_finalizer self, ConnectionHandler.unowned_pool_finalizer(@owner_to_pool)
       end
 
       def connection_pool_list

activerecord/lib/active_record/connection_adapters/abstract_adapter.rb

@@ -367,6 +367,19 @@ def disconnect!
         reset_transaction
       end
 
+      # Immediately forget this connection ever existed. Unlike disconnect!,
+      # this will not communicate with the server.
+      #
+      # After calling this method, the behavior of all other methods becomes
+      # undefined. This is called internally just before a forked process gets
+      # rid of a connection that belonged to its parent.
+      def discard!
+        # This should be overridden by concrete adapters.
+        #
+        # Prevent @connection's finalizer from touching the socket, or
+        # otherwise communicating with its server, when it is collected.
+      end
+
       # Reset the state of this connection, directing the DBMS to clear
       # transactions and other connection-related server-side state. Usually a
       # database-dependent operation.

activerecord/lib/active_record/connection_adapters/mysql2_adapter.rb

@@ -105,6 +105,11 @@ def disconnect!
         @connection.close
       end
 
+      def discard! # :nodoc:
+        @connection.automatic_close = false
+        @connection = nil
+      end
+
       private
 
         def connect

activerecord/lib/active_record/connection_adapters/postgresql_adapter.rb

@@ -273,6 +273,11 @@ def disconnect!
         end
       end
 
+      def discard! # :nodoc:
+        @connection.socket_io.reopen(IO::NULL)
+        @connection = nil
+      end
+
       def native_database_types #:nodoc:
         NATIVE_DATABASE_TYPES
       end

activerecord/test/cases/connection_adapters/connection_handler_test.rb

 # frozen_string_literal: true
 
 require "cases/helper"
+require "models/person"
 
 module ActiveRecord
   module ConnectionAdapters
     class ConnectionHandlerTest < ActiveRecord::TestCase
+      self.use_transactional_tests = false
+
+      fixtures :people
+
       def setup
         @handler = ConnectionHandler.new
         @spec_name = "primary"
@@ -139,6 +144,33 @@ def test_connection_pool_per_pid
           rd.close
         end
 
+        def test_forked_child_doesnt_mangle_parent_connection
+          object_id = ActiveRecord::Base.connection.object_id
+          assert ActiveRecord::Base.connection.active?
+
+          rd, wr = IO.pipe
+          rd.binmode
+          wr.binmode
+
+          pid = fork {
+            rd.close
+            if ActiveRecord::Base.connection.active?
+              wr.write Marshal.dump ActiveRecord::Base.connection.object_id
+            end
+            wr.close
+
+            exit # allow finalizers to run
+          }
+
+          wr.close
+
+          Process.waitpid pid
+          assert_not_equal object_id, Marshal.load(rd.read)
+          rd.close
+
+          assert_equal 3, ActiveRecord::Base.connection.select_value("SELECT COUNT(*) FROM people")
+        end
+
         def test_retrieve_connection_pool_copies_schema_cache_from_ancestor_pool
           @pool.schema_cache = @pool.connection.schema_cache
           @pool.schema_cache.add("posts")