Handle outdated Marshal payloads in Cache::Entry with 6.1 cache_format

Ref: https://github.com/rails/rails/issues/48611
Followup: https://github.com/rails/rails/pull/48663

It's the same logic than https://github.com/rails/rails/pull/48663
but now works for the 6.1 cache format.

activesupport/lib/active_support/cache.rb

@@ -459,7 +459,17 @@ module ActiveSupport
             instrument(:read, name, options) do |payload|
               cached_entry = read_entry(key, **options, event: payload)
               entry = handle_expired_entry(cached_entry, key, options)
-              entry = nil if entry && entry.mismatched?(normalize_version(name, options))
+              if entry
+                if entry.mismatched?(normalize_version(name, options))
+                  entry = nil
+                else
+                  begin
+                    entry.value
+                  rescue DeserializationError
+                    entry = nil
+                  end
+                end
+              end
               payload[:super_operation] = :fetch if payload
               payload[:hit] = !!entry if payload
             end
@@ -511,7 +521,12 @@ module ActiveSupport
               nil
             else
               payload[:hit] = true if payload
-              entry.value
+              begin
+                entry.value
+              rescue DeserializationError
+                payload[:hit] = false
+                nil
+              end
             end
           else
             payload[:hit] = false if payload

activesupport/lib/active_support/cache/entry.rb

@@ -121,7 +121,13 @@ module ActiveSupport
 
       private
         def uncompress(value)
-          Marshal.load(Zlib::Inflate.inflate(value))
+          marshal_load(Zlib::Inflate.inflate(value))
+        end
+
+        def marshal_load(payload)
+          Marshal.load(payload)
+        rescue ArgumentError => error
+          raise Cache::DeserializationError, error.message
         end
     end
   end

activesupport/lib/active_support/cache/mem_cache_store.rb

@@ -270,14 +270,22 @@ module ActiveSupport
         def read_multi_entries(names, **options)
           keys_to_names = names.index_by { |name| normalize_key(name, options) }
 
-          raw_values = @data.with { |c| c.get_multi(keys_to_names.keys) }
+          raw_values = begin
+            @data.with { |c| c.get_multi(keys_to_names.keys) }
+          rescue Dalli::UnmarshalError
+            {}
+          end
+
           values = {}
 
           raw_values.each do |key, value|
             entry = deserialize_entry(value, raw: options[:raw])
 
             unless entry.nil? || entry.expired? || entry.mismatched?(normalize_version(keys_to_names[key], options))
-              values[keys_to_names[key]] = entry.value
+              begin
+                values[keys_to_names[key]] = entry.value
+              rescue DeserializationError
+              end
             end
           end
 

activesupport/lib/active_support/cache/redis_cache_store.rb

@@ -332,7 +332,10 @@ module ActiveSupport
             if value
               entry = deserialize_entry(value, raw: raw)
               unless entry.nil? || entry.expired? || entry.mismatched?(normalize_version(name, options))
-                results[name] = entry.value
+                begin
+                  results[name] = entry.value
+                rescue DeserializationError
+                end
               end
             end
           end

activesupport/test/cache/behaviors/cache_store_format_version_behavior.rb

@@ -64,16 +64,31 @@ module CacheStoreFormatVersionBehavior
 
       test "Marshal undefined class/module deserialization error with #{format_version} format" do
         key = "marshal-#{rand}"
-        self.class.const_set(:Foo, Class.new)
+        self.class.const_set(:RemovedConstant, 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)
+        @store.write(key, self.class::RemovedConstant.new)
+        assert_instance_of self.class::RemovedConstant, @store.read(key)
 
-        self.class.send(:remove_const, :Foo)
+        self.class.send(:remove_const, :RemovedConstant)
         assert_nil @store.read(key)
         assert_equal false, @store.exist?(key)
       ensure
-        self.class.send(:remove_const, :Foo) rescue nil
+        self.class.send(:remove_const, :RemovedConstant) rescue nil
+      end
+
+      test "Compressed Marshal undefined class/module deserialization error with #{format_version} format" do
+        key = "marshal-#{rand}"
+        self.class.const_set(:RemovedConstant, Class.new)
+        @store = with_format(format_version) { lookup_store }
+        @store.write(key, self.class::RemovedConstant.new, compress: true, compress_threshold: 1)
+        assert_instance_of self.class::RemovedConstant, @store.read(key)
+
+        self.class.send(:remove_const, :RemovedConstant)
+        assert_nil @store.read(key)
+        assert_equal({}, @store.read_multi(key))
+        assert_equal("new-value", @store.fetch(key) { "new-value" })
+      ensure
+        self.class.send(:remove_const, :RemovedConstant) rescue nil
       end
     end