FIX: Add a lock to ensure only a single thread is running each time.

lib/active_record/connection_adapters/postgresql_fallback_adapter.rb

@@ -11,11 +11,14 @@ class PostgreSQLFallbackHandler
   def initialize
     @master = true
     @running = false
+    @mutex = Mutex.new
   end
 
   def verify_master
-    return if @running && recently_checked?
-    @running = true
+    @mutex.synchronize do
+      return if @running || recently_checked?
+      @running = true
+    end
 
     Thread.new do
       begin
@@ -23,6 +26,7 @@ class PostgreSQLFallbackHandler
         connection = ActiveRecord::Base.postgresql_connection(config)
 
         if connection.active?
+          connection.disconnect!
           logger.info "#{self.class}: Master server is active. Reconnecting..."
           ActiveRecord::Base.establish_connection(config)
           Discourse.disable_readonly_mode
@@ -53,7 +57,7 @@ class PostgreSQLFallbackHandler
 
   def recently_checked?
     if @last_check
-      Time.zone.now <= @last_check + 5.seconds
+      Time.zone.now <= (@last_check + 5.seconds)
     else
       false
     end
@@ -66,7 +70,7 @@ module ActiveRecord
       fallback_handler = ::PostgreSQLFallbackHandler.instance
       config = config.symbolize_keys
 
-      if !fallback_handler.master
+      if !fallback_handler.master && !fallback_handler.running
         connection = postgresql_connection(config.dup.merge({
           host: config[:replica_host], port: config[:replica_port]
         }))

spec/components/active_record/connection_adapters/postgresql_fallback_adapter_spec.rb

@@ -2,11 +2,14 @@ require 'rails_helper'
 require_dependency 'active_record/connection_adapters/postgresql_fallback_adapter'
 
 describe ActiveRecord::ConnectionHandling do
+  let(:replica_host) { "1.1.1.1" }
+  let(:replica_port) { "6432" }
+
   let(:config) do
     ActiveRecord::Base.configurations["test"].merge({
       "adapter" => "postgresql_fallback",
-      "replica_host" => "localhost",
-      "replica_port" => "6432"
+      "replica_host" => replica_host,
+      "replica_port" => replica_port
     }).symbolize_keys!
   end
 
@@ -31,7 +34,7 @@ describe ActiveRecord::ConnectionHandling do
         ActiveRecord::Base.expects(:verify_replica).with(@replica_connection)
 
         ActiveRecord::Base.expects(:postgresql_connection).with(config.merge({
-          host: "localhost", port: "6432"
+          host: replica_host, port: replica_port
         })).returns(@replica_connection)
 
         expect { ActiveRecord::Base.postgresql_fallback_connection(config) }
@@ -47,6 +50,14 @@ describe ActiveRecord::ConnectionHandling do
         expect{ ActiveRecord::Base.connection_pool.checkout }
           .to change{ Thread.list.size }.by(1)
 
+        # Ensure that we don't try to connect back to the replica when a thread
+        # is running
+        begin
+          ActiveRecord::Base.postgresql_fallback_connection(config)
+        rescue PG::ConnectionBad => e
+          # This is expected if the thread finishes before the above is called.
+        end
+
         # Wait for the thread to finish execution
         threads = (Thread.list - current_threads).each(&:join)