aks/sync query cache changes

fresh_connection.gemspec

@@ -33,4 +33,5 @@ Gem::Specification.new do |spec|
   spec.add_development_dependency "minitest"
   spec.add_development_dependency "minitest-reporters"
   spec.add_development_dependency "benchmark-ips"
+  #spec.add_development_dependency "pry-byebug"  # debugging only
 end

gemfiles/rails42.gemfile

@@ -6,4 +6,4 @@ gem "activerecord", "~> 4.2.0"
 gem "activesupport", "~> 4.2.0"
 gem "mysql2", "~> 0.3.13"
 
-gemspec :path => "../"
+gemspec path: "../"

gemfiles/rails50.gemfile

@@ -6,4 +6,4 @@ gem "activerecord", "~> 5.0.0"
 gem "activesupport", "~> 5.0.0"
 gem "mysql2", ">= 0.3.18", "< 0.5"
 
-gemspec :path => "../"
+gemspec path: "../"

lib/fresh_connection/abstract_connection_manager.rb

@@ -12,7 +12,7 @@ def method_added(name)
       end
     end
 
-    attr_reader :replica_group
+    attr_reader   :replica_group
 
     def initialize(replica_group = "replica")
       replica_group = "replica" if replica_group.to_s == "slave"
@@ -36,6 +36,10 @@ def clear_all_connections!
     end
     undef_method :clear_all_connections!
 
+    def clear_replica_query_caches!
+    end
+    undef_method :clear_replica_query_caches!
+
     def put_aside!
     end
     undef_method :put_aside!

lib/fresh_connection/connection_factory.rb

@@ -11,8 +11,10 @@ def initialize(group, modify_spec = nil)
       @spec = nil
     end
 
-    def new_connection
-      ActiveRecord::Base.__send__(adapter_method, spec)
+    def new_connection(pool = nil)
+      ActiveRecord::Base.__send__(adapter_method, spec).tap do |conn|
+        conn.connection_pool = pool
+      end
     end
 
     private

lib/fresh_connection/connection_manager.rb

@@ -1,33 +1,61 @@
+require 'active_record'
 require 'concurrent'
 require 'fresh_connection/abstract_connection_manager'
 require 'fresh_connection/connection_factory'
 
 module FreshConnection
   class ConnectionManager < AbstractConnectionManager
+
+    if ActiveRecord::VERSION::MAJOR == 5
+      eval 'include ::ActiveRecord::ConnectionAdapters::QueryCache::ConnectionPoolConfiguration'
+    end
+
     def initialize(*args)
       super
       @connections = Concurrent::Map.new
+      if ActiveRecord::VERSION::MAJOR == 4
+        @query_cache_enabled = Concurrent::Map.new { false }
+      end
     end
 
+
     def replica_connection
       @connections.fetch_or_store(current_thread_id) do |_|
-        connection_factory.new_connection
+        connection_factory.new_connection(self)
       end
     end
 
     def put_aside!
-      conn = @connections.delete(current_thread_id)
-      return unless conn
-      conn && conn.disconnect! rescue nil
+      if (conn = @connections.delete(current_thread_id))
+        conn.disconnect! rescue nil
+      end
     end
 
     def clear_all_connections!
-      @connections.each_value do |conn|
+      for_all_replica_connections do |conn|
         conn.disconnect! rescue nil
       end
       @connections.clear
     end
 
+    def clear_replica_query_caches!
+      for_all_replica_connections do |conn|
+        conn.clear_query_cache
+      end
+    end
+
+    def enable_query_cache!
+      for_all_replica_connections do |conn|
+        conn.enable_query_cache!
+      end
+    end
+
+    def disable_query_cache!
+      for_all_replica_connections do |conn|
+        conn.disable_query_cache!
+      end
+    end
+
     def recovery?
       return false if replica_connection.active?
       put_aside!
@@ -36,6 +64,12 @@ def recovery?
 
     private
 
+    def for_all_replica_connections
+      @connections.each_value do |connection|
+        yield(connection)
+      end
+    end
+
     def connection_factory
       @connection_factory ||= ConnectionFactory.new(@replica_group)
     end

lib/fresh_connection/extend/ar_abstract_adapter.rb

@@ -2,13 +2,44 @@ module FreshConnection
   module Extend
     module ArAbstractAdapter
       def self.prepended(base)
-        base.send :attr_writer, :replica_group
+        base.send :attr_accessor, :replica_group
+        base.send :attr_accessor, :connection_pool
+      end
+
+      def initialize(connection, logger = nil, config = {})
+        super
+        @connection_pool = nil
       end
 
       def log(*args)
         args[1] = "[#{@replica_group}] #{args[1]}" if defined?(@replica_group)
         super
       end
+
+      # This is normally called only on master connections, so we need
+      # to clear the replica connection caches, too.  But, don't recurse.
+      def clear_query_cache
+        if FreshConnection::ReplicaConnectionHandler.replica_query_cache_sync
+
+          # This call is interesting.  Here, in the FreshConnection
+          # extension to the AR AbstractAdapter, we are either on a :master
+          # connection or a :replica connection.  Unfortunately, there is no direct
+          # linkage between them.  So, from a :master connection, we don't know
+          # which :replica connection pool to use, or even which :replica connection
+          # manager to use, because those are associated with AR objects, not
+          # AR adapters.
+
+          # So, we need to "cross over" from the master connection side to the
+          # replica connection side, via a top-level AR::Base call, but we need
+          # to avoid accidental recursions, too.  The "replica_group" test
+          # should be non-nil for replica connections.
+
+          if replica_group.nil? || replica_group == :master
+            ActiveRecord::Base.clear_replica_query_caches!
+          end
+        end
+        super
+      end
     end
   end
 end

lib/fresh_connection/extend/ar_base.rb

@@ -29,6 +29,10 @@ def establish_fresh_connection(replica_group = "replica")
         replica_connection_handler.establish_connection(name, replica_group)
       end
 
+      def master_connection
+        superclass.connection
+      end
+
       def replica_connection
         replica_connection_handler.connection(self)
       end
@@ -53,6 +57,22 @@ def clear_all_slave_connections!
         clear_all_replica_connections!
       end
 
+      def clear_replica_query_caches!
+        replica_connection_handler.clear_replica_query_caches!
+      end
+
+      def enable_replica_query_cache_sync!
+        FreshConnection::ReplicaConnectionHandler.enable_query_cache_sync!
+      end
+
+      def disable_replica_query_cache_sync!
+        FreshConnection::ReplicaConnectionHandler.disable_query_cache_sync!
+      end
+
+      def replica_query_cache_sync
+        FreshConnection::ReplicaConnectionHandler.replica_query_cache_sync
+      end
+
       def master_db_only!
         @_fresh_connection_master_only = true
       end

lib/fresh_connection/replica_connection_handler.rb

@@ -1,7 +1,19 @@
 require 'concurrent'
 
 module FreshConnection
+
   class ReplicaConnectionHandler
+
+    cattr_accessor :replica_query_cache_sync
+
+    def self.enable_query_cache_sync!
+      @@replica_query_cache_sync = true
+    end
+
+    def self.disable_query_cache_sync!
+      @@replica_query_cache_sync = false
+    end
+
     def initialize
       @replica_group_to_pool = Concurrent::Map.new
       @class_to_pool = Concurrent::Map.new
@@ -26,6 +38,12 @@ def clear_all_connections!
       end
     end
 
+    def clear_replica_query_caches!
+      all_connection_managers do |connection_manager|
+        connection_manager.clear_replica_query_caches!
+      end
+    end
+
     def recovery?(klass)
       detect_connection_manager(klass).recovery?
     end

lib/fresh_connection/version.rb

@@ -1,4 +1,4 @@
 module FreshConnection
-  VERSION = "2.3.1"
+  VERSION = "2.3.1.rc6"
 end
 

test/ar_base/ar_abstract_adapter_test.rb

@@ -0,0 +1,76 @@
+require "test_helper"
+#require 'pry-byebug'
+
+class AbstractAdapterTest < Minitest::Test
+
+  class Tel < ActiveRecord::Base
+    establish_fresh_connection :fake_replica
+    enable_replica_query_cache_sync!
+
+    belongs_to :user
+  end
+
+  def setup
+    ActiveRecord::Base.connection.clear_query_cache
+  end
+
+  test "cache_query is incorrect after master update with replica cache syncing disabled" do
+
+    Tel.cache do
+
+      Tel.replica_connection.enable_query_cache!
+      Tel.disable_replica_query_cache_sync!
+
+      tel = Tel.find(1)
+      assert_match(/master/ , tel.number)
+      rconn = Tel.replica_connection
+      mconn = Tel.master_connection
+
+      refute rconn.query_cache.empty?
+      assert mconn.query_cache.empty?
+
+      orig_number = tel.number
+      tel.number = tel.number.sub(/master/,'fake_replica')
+      tel.save!
+      assert mconn.query_cache.empty?
+      refute rconn.query_cache.empty?       # the replica cache should still have some contents after an update
+
+      tel2 = Tel.find(1)
+      refute_equal tel2.number, tel.number  # the replica cache contents should be stale
+
+      tel.number = orig_number
+      tel.save!
+    end
+
+  end
+
+  test "cache_query is correct after master update with replica cache syncing enabled" do
+
+    Tel.cache do
+
+      Tel.replica_connection.enable_query_cache!
+      Tel.enable_replica_query_cache_sync!
+
+      tel = Tel.find(1)
+      assert_match(/master/ , tel.number)
+      rconn = Tel.replica_connection
+      mconn = Tel.master_connection
+
+      refute rconn.query_cache.empty?     # the replica cache should have some contents
+      assert mconn.query_cache.empty?     # the master connection cache is always empty
+
+      orig_number = tel.number
+      tel.number = tel.number.sub(/master/,'fake_replica')
+      tel.save!
+
+      tel2 = Tel.find(1)
+      assert_equal tel2.number, tel.number  # the replica should have pulled fresh, correct data
+
+      tel.number = orig_number
+      tel.save!
+    end
+
+  end
+
+end
+

test/config/database_mysql2.yml

@@ -12,3 +12,5 @@ test:
   replica2:
     database: fresh_connection_test_replica2
 
+  fake_replica:
+    database: fresh_connection_test_master

test/config/database_postgresql.yml

@@ -10,3 +10,5 @@ test:
   replica2:
     database: fresh_connection_test_replica2
 
+  fake_replica:
+    database: fresh_connection_test_master