Ensure all connection accesses are wrapped in failsafe

lib/solid_cache/cluster/connection_handling.rb

@@ -64,23 +64,9 @@ def writing_all_shards
         end
       end
 
-      def writing_across_shards(list:, trim: false)
-        across_shards(list:, async: async_writes) do |list|
-          result = yield list
-          trim(list.size) if trim
-          result
-        end
-      end
-
-      def reading_across_shards(list:)
-        across_shards(list:) { |list| yield list }
-      end
-
-      def writing_shard(normalized_key:, trim: false)
+      def writing_shard(normalized_key:)
         with_shard(shard_for_normalized_key(normalized_key), async: async_writes) do
-          result = yield
-          trim(1) if trim
-          result
+          yield
         end
       end
 
@@ -92,24 +78,24 @@ def active_record_instrumentation?
         @active_record_instrumentation
       end
 
-      private
-        attr_reader :consistent_hash
+      def across_shards(list:, async: false)
+        in_shards(list).map do |shard, list|
+          yield shard, list
+        end
+      end
 
-        def with_shard(shard, async: false)
-          if shard
-            Record.connected_to(shard: shard) do
-              configure_for_query(async: async) { yield }
-            end
-          else
+      def with_shard(shard, async: false)
+        if shard
+          Record.connected_to(shard: shard) do
             configure_for_query(async: async) { yield }
           end
+        else
+          configure_for_query(async: async) { yield }
         end
+      end
 
-        def across_shards(list:, async: false)
-          in_shards(list).map do |shard, list|
-            with_shard(shard, async: async) { yield list }
-          end
-        end
+      private
+        attr_reader :consistent_hash
 
         def in_shards(list)
           if shards.count == 1

lib/solid_cache/store.rb

@@ -58,20 +58,16 @@ def delete_matched(matcher, options = {})
     def increment(name, amount = 1, options = nil)
       options = merged_options(options)
       key = normalize_key(name, options)
-      writing_key(key) do
-        failsafe :increment do
-          Entry.increment(key, amount)
-        end
+      writing_key(key, failsafe: :increment) do
+        Entry.increment(key, amount)
       end
     end
 
     def decrement(name, amount = 1, options = nil)
       options = merged_options(options)
       key = normalize_key(name, options)
-      writing_key(key) do
-        failsafe :decrement do
-          Entry.increment(key, -amount)
-        end
+      writing_key(key, failsafe: :decrement) do
+        Entry.increment(key, -amount)
       end
     end
 
@@ -93,8 +89,8 @@ def read_entry(key, **options)
       end
 
       def read_serialized_entry(key, raw: false, **options)
-        primary_cluster.reading_shard(normalized_key: key) do
-          failsafe(:read_entry) do
+        failsafe(:read_entry) do
+          primary_cluster.reading_shard(normalized_key: key) do
             Entry.get(key)
           end
         end
@@ -105,11 +101,10 @@ def write_entry(key, entry, raw: false, **options)
         payload = serialize_entry(entry, raw: raw, **options)
         write_serialized_entry(key, payload, raw: raw, **options)
 
-        writing_key(key, trim: true) do
-          failsafe(:write_entry, returning: false) do
-            Entry.set(key, payload)
-            true
-          end
+        writing_key(key, failsafe: :write_entry, failsafe_returning: false) do |cluster|
+          Entry.set(key, payload)
+          cluster.trim(1)
+          true
         end
       end
 
@@ -118,9 +113,11 @@ def write_serialized_entry(key, payload, raw: false, unless_exist: false, expire
       end
 
       def read_serialized_entries(keys)
-        results = primary_cluster.reading_across_shards(list: keys) do |keys|
+        results = primary_cluster.across_shards(list: keys) do |shard, keys|
           failsafe(:read_multi_mget, returning: {}) do
-            Entry.get_all(keys)
+            primary_cluster.with_shard(shard) do
+              Entry.get_all(keys)
+            end
           end
         end
 
@@ -155,20 +152,21 @@ def write_multi_entries(entries, expires_in: nil, **options)
             write_serialized_entry(entries[:key], entries[:value])
           end
 
-          writing_list(serialized_entries, trim: true) do |serialized_entries|
-            failsafe(:write_multi_entries) do
-              Entry.set_all(serialized_entries)
-              true
+          writing_list(serialized_entries) do |cluster, shard, serialized_entries|
+            failsafe(:write_multi_entries, returning: false) do
+              cluster.with_shard(shard) do
+                Entry.set_all(serialized_entries)
+                cluster.trim(serialized_entries.count)
+                true
+              end
             end
-          end
+          end.all?
         end
       end
 
       def delete_entry(key, **options)
-        writing_key(key) do
-          failsafe(:delete_entry, returning: false) do
-            Entry.delete_by_key(key)
-          end
+        writing_key(key, failsafe: :delete_entry, failsafe_returning: false) do
+          Entry.delete_by_key(key)
         end
       end
 
@@ -220,18 +218,20 @@ def failsafe(method, returning: nil)
         returning
       end
 
-      def writing_key(key, trim: false)
+      def writing_key(key, failsafe:, failsafe_returning: nil)
         writing_clusters do |cluster|
-          cluster.writing_shard(normalized_key: key, trim: trim) do
-            yield
+          failsafe(failsafe, returning: failsafe_returning) do
+            cluster.writing_shard(normalized_key: key) do
+              yield cluster
+            end
           end
         end
       end
 
-      def writing_list(list, trim: false)
+      def writing_list(list)
         writing_clusters do |cluster|
-          cluster.writing_across_shards(list: list, trim: trim) do |list|
-            yield list
+          cluster.across_shards(list: list) do |shard, list|
+            yield cluster, shard, list
           end
         end
       end

test/unit/connection_handling_test.rb

@@ -29,4 +29,32 @@ def test_active_record_instrumention
       end
     end
   end
+
+  def test_no_connections_uninstrumented
+    ActiveRecord::ConnectionAdapters::ConnectionPool.any_instance.stubs(:connection).raises(ActiveRecord::StatementInvalid)
+
+    cache = lookup_store(expires_in: 60, cluster: { shards: [:primary_shard_one, :primary_shard_two] }, active_record_instrumentation: false)
+
+    assert_equal false, cache.write("1", "fsjhgkjfg")
+    assert_nil cache.read("1")
+    assert_nil cache.increment("1")
+    assert_nil cache.decrement("1")
+    assert_equal false, cache.delete("1")
+    assert_equal({}, cache.read_multi("1", "2", "3"))
+    assert_equal false, cache.write_multi("1" => "a", "2" => "b", "3" => "c")
+  end
+
+  def test_no_connections_instrumented
+    ActiveRecord::ConnectionAdapters::ConnectionPool.any_instance.stubs(:connection).raises(ActiveRecord::StatementInvalid)
+
+    cache = lookup_store(expires_in: 60, cluster: { shards: [:primary_shard_one, :primary_shard_two] })
+
+    assert_equal false, cache.write("1", "fsjhgkjfg")
+    assert_nil cache.read("1")
+    assert_nil cache.increment("1")
+    assert_nil cache.decrement("1")
+    assert_equal false, cache.delete("1")
+    assert_equal({}, cache.read_multi("1", "2", "3"))
+    assert_equal false, cache.write_multi("1" => "a", "2" => "b", "3" => "c")
+  end
 end