Make Active Support Cache treat deserialization errors like cache misses

This help treating caches entries as expandable.

Because Marshal will hapily serialize almost everything, It's not uncommon to
inadvertently cache a class that's not particularly stable, and cause deserialization
errors on rollout when the implementation changes.

E.g. sporkmonger/addressable#508

With this change, in case of such event, the hit rate will suffer for a
bit, but the application won't return 500s responses.

activesupport/lib/active_support/cache.rb

@@ -30,6 +30,10 @@ module Cache
       expires_in: [:expire_in, :expired_in]
     }.freeze
 
+    # Raised by coders when the cache entry can't be deserialized.
+    # This error is treated as a cache miss.
+    DeserializationError = Class.new(StandardError)
+
     module Strategy
       autoload :LocalCache, "active_support/cache/strategy/local_cache"
     end
@@ -732,6 +736,8 @@ def serialize_entry(entry, **options)
 
         def deserialize_entry(payload)
           payload.nil? ? nil : @coder.load(payload)
+        rescue DeserializationError
+          nil
         end
 
         # Reads multiple entries from the cache implementation. Subclasses MAY

activesupport/lib/active_support/cache/serializer_with_fallback.rb

@@ -59,6 +59,12 @@ def load(dumped)
         BARE_STRING_VERSION_LENGTH_TEMPLATE = "@#{[0].pack(BARE_STRING_EXPIRES_AT_TEMPLATE).bytesize}l<"
         BARE_STRING_VERSION_INDEX = [0].pack(BARE_STRING_VERSION_LENGTH_TEMPLATE).bytesize
 
+        def marshal_load(payload)
+          Marshal.load(payload)
+        rescue ArgumentError => error
+          raise Cache::DeserializationError, error.message
+        end
+
         def try_dump_bare_string(entry)
           value = entry.value
           return if !value.instance_of?(String)
@@ -144,9 +150,8 @@ def dump_compressed(entry, threshold)
             Marshal.dump(entry.compressed(threshold))
           end
 
-          def _load(dumped)
-            Marshal.load(dumped)
-          end
+          alias_method :_load, :marshal_load
+          public :_load
 
           def dumped?(dumped)
             dumped.start_with?(MARSHAL_SIGNATURE)
@@ -184,7 +189,7 @@ def dump_compressed(entry, threshold)
           def _load(marked)
             dumped = marked.byteslice(1..-1)
             dumped = decompress(dumped) if marked.start_with?(MARK_COMPRESSED)
-            try_load_bare_string(dumped) || Cache::Entry.unpack(Marshal.load(dumped))
+            try_load_bare_string(dumped) || Cache::Entry.unpack(marshal_load(dumped))
           end
 
           def dumped?(dumped)

activesupport/test/cache/behaviors/cache_store_format_version_behavior.rb

@@ -61,6 +61,20 @@ module CacheStoreFormatVersionBehavior
 
         assert_operator serialized, :start_with?, compressed_signature
       end
+
+      test "Marshal undefined class/module deserialization error with #{format_version} format" do
+        key = "marshal-#{rand}"
+        self.class.const_set(:Foo, Class.new)
+        @store = with_format(format_version) { lookup_store }
+        @store.write(key, self.class::Foo.new)
+        assert_instance_of self.class::Foo, @store.read(key)
+
+        self.class.send(:remove_const, :Foo)
+        assert_nil @store.read(key)
+        assert_equal false, @store.exist?(key)
+      ensure
+        self.class.send(:remove_const, :Foo) rescue nil
+      end
     end
 
     FORMAT_VERSIONS.product(FORMAT_VERSIONS) do |read_version, write_version|