Force connected_to to load the records if it's a Relation

Fixes #38332

If the `connected_to` block returns a relation and does not inspect, or
load that relation before returning it, the block will exit before the
database is queried. This causes the wrong database connection to be
queried.

The consequences of this are getting records from the primary instead of
the replica, and potentially having database performance impact.

Relations lazily query the database. If you return the relation from the
block like:

```
posts = ActiveRecord::Base.connected_to(role: :reading) { Post.where(id: 1) }
```

`posts.first` will be queried from the `writing` connection instead
because it's lazy and performed outside the block. Any query that loads
the relation (ie `to_a`) inside the block would eagerly load the
relation's records and not exhibit this bug.

`connected_to` now checks if the return value is a `Relation` and if so
calls `load`.
Co-authored-by: NAaron Patterson <aaron.patterson@gmail.com>

activerecord/lib/active_record/connection_handling.rb

@@ -257,7 +257,9 @@ def resolve_config_for_connection(config_or_env)
 
       def swap_connection_handler(handler, &blk) # :nodoc:
         old_handler, ActiveRecord::Base.connection_handler = ActiveRecord::Base.connection_handler, handler
-        yield
+        return_value = yield
+        return_value.load if return_value.is_a? ActiveRecord::Relation
+        return_value
       ensure
         ActiveRecord::Base.connection_handler = old_handler
       end

activerecord/test/cases/connection_adapters/connection_handlers_multi_db_test.rb

@@ -35,12 +35,12 @@ def test_multiple_connection_handlers_works_in_a_threaded_environment
         # and won't be able to write to the second connection.
         MultiConnectionTestModel.connects_to database: { writing: { database: tf_writing.path, adapter: "sqlite3" }, secondary: { database: tf_reading.path, adapter: "sqlite3" } }
 
-        MultiConnectionTestModel.connection.execute("CREATE TABLE `test_1` (connection_role VARCHAR (255))")
-        MultiConnectionTestModel.connection.execute("INSERT INTO test_1 VALUES ('writing')")
+        MultiConnectionTestModel.connection.execute("CREATE TABLE `multi_connection_test_models` (connection_role VARCHAR (255))")
+        MultiConnectionTestModel.connection.execute("INSERT INTO multi_connection_test_models VALUES ('writing')")
 
         ActiveRecord::Base.connected_to(role: :secondary) do
-          MultiConnectionTestModel.connection.execute("CREATE TABLE `test_1` (connection_role VARCHAR (255))")
-          MultiConnectionTestModel.connection.execute("INSERT INTO test_1 VALUES ('reading')")
+          MultiConnectionTestModel.connection.execute("CREATE TABLE `multi_connection_test_models` (connection_role VARCHAR (255))")
+          MultiConnectionTestModel.connection.execute("INSERT INTO multi_connection_test_models VALUES ('reading')")
         end
 
         read_latch = Concurrent::CountDownLatch.new
@@ -52,13 +52,13 @@ def test_multiple_connection_handlers_works_in_a_threaded_environment
           MultiConnectionTestModel.connection
 
           write_latch.wait
-          assert_equal "writing", MultiConnectionTestModel.connection.select_value("SELECT connection_role from test_1")
+          assert_equal "writing", MultiConnectionTestModel.connection.select_value("SELECT connection_role from multi_connection_test_models")
           read_latch.count_down
         end
 
         ActiveRecord::Base.connected_to(role: :secondary) do
           write_latch.count_down
-          assert_equal "reading", MultiConnectionTestModel.connection.select_value("SELECT connection_role from test_1")
+          assert_equal "reading", MultiConnectionTestModel.connection.select_value("SELECT connection_role from multi_connection_test_models")
           read_latch.wait
         end
 
@@ -70,6 +70,20 @@ def test_multiple_connection_handlers_works_in_a_threaded_environment
         tf_writing.unlink
       end
 
+      def test_loading_relations_with_multi_db_connection_handlers
+        # We need to use a role for reading not named reading, otherwise we'll prevent writes
+        # and won't be able to write to the second connection.
+        MultiConnectionTestModel.connects_to database: { writing: { database: ":memory:", adapter: "sqlite3" }, secondary: { database: ":memory:", adapter: "sqlite3" } }
+
+        relation = ActiveRecord::Base.connected_to(role: :secondary) do
+          MultiConnectionTestModel.connection.execute("CREATE TABLE `multi_connection_test_models` (connection_role VARCHAR (255))")
+          MultiConnectionTestModel.create!(connection_role: "reading")
+          MultiConnectionTestModel.where(connection_role: "reading")
+        end
+
+        assert_equal "reading", relation.first.connection_role
+      end
+
       unless in_memory_db?
         def test_establish_connection_using_3_levels_config
           previous_env, ENV["RAILS_ENV"] = ENV["RAILS_ENV"], "default_env"