Refactor LocalCache to avoid calling Marshal.dump as much (take 2)

The local cache need to protect against mutations of both
the initial received value, and the value returned.

See:

  - rails#36656
  - rails#37587

Because of this, the overhead of the local cache end up being
quite significant. On a single read, the value will be deep duped
twice. So unless the value is one of the type benefiting from a
fast path, we'll have to do two `Marshal.load(Marshal.dump(value))`
roundtrips, which for large values can be very expensive.

By using a specialized `LocalEntry` type instead, we can store
the `Marshal` payload rather than the original value. This way
the overhead is reduced to a single `Marshal.dump` on writes
and a single `Marshal.load` on reads.

activesupport/lib/active_support/cache.rb

@@ -527,6 +527,10 @@ def exist?(name, options = nil)
         end
       end
 
+      def new_entry(value, options = nil) # :nodoc:
+        Entry.new(value, **merged_options(options))
+      end
+
       # Deletes all entries with keys matching the pattern.
       #
       # Options are passed to the underlying cache implementation.
@@ -858,6 +862,14 @@ def bytesize
         end
       end
 
+      def compressed? # :nodoc:
+        defined?(@compressed)
+      end
+
+      def local?
+        false
+      end
+
       # Duplicates the value in a class. This is used by cache implementations that don't natively
       # serialize entries to protect against accidental cache modifications.
       def dup_value!
@@ -893,10 +905,6 @@ def compress!(compress_threshold)
           end
         end
 
-        def compressed?
-          defined?(@compressed)
-        end
-
         def uncompress(value)
           Marshal.load(Zlib::Inflate.inflate(value))
         end

activesupport/lib/active_support/cache/strategy/local_cache.rb

@@ -35,6 +35,78 @@ def self.cache_for(l); instance.cache_for l; end
         # Simple memory backed cache. This cache is not thread safe and is intended only
         # for serving as a temporary memory cache for a single thread.
         class LocalStore < Store
+          class Entry # :nodoc:
+            class << self
+              def build(cache_entry)
+                return if cache_entry.nil?
+                return cache_entry if cache_entry.compressed?
+
+                value = cache_entry.value
+                if value.is_a?(String)
+                  DupableEntry.new(cache_entry)
+                elsif !value || value == true || value.is_a?(Numeric)
+                  new(cache_entry)
+                else
+                  MutableEntry.new(cache_entry)
+                end
+              end
+            end
+
+            attr_reader :value, :version
+            attr_accessor :expires_at
+
+            def initialize(cache_entry)
+              @value = cache_entry.value
+              @expires_at = cache_entry.expires_at
+              @version = cache_entry.version
+            end
+
+            def local?
+              true
+            end
+
+            def compressed?
+              false
+            end
+
+            def mismatched?(version)
+              @version && version && @version != version
+            end
+
+            def expired?
+              expires_at && expires_at <= Time.now.to_f
+            end
+
+            def marshal_dump
+              raise NotImplementedError, "LocalStore::Entry should never be serialized"
+            end
+          end
+
+          class DupableEntry < Entry # :nodoc:
+            def initialize(_cache_entry)
+              super
+              unless @value.frozen?
+                @value = @value.dup.freeze
+              end
+            end
+
+            def value
+              @value.dup
+            end
+          end
+
+          class MutableEntry < Entry # :nodoc:
+            def initialize(cache_entry)
+              @payload = Marshal.dump(cache_entry.value)
+              @expires_at = cache_entry.expires_at
+              @version = cache_entry.version
+            end
+
+            def value
+              Marshal.load(@payload)
+            end
+          end
+
           def initialize
             super
             @data = {}
@@ -65,8 +137,7 @@ def read_multi_entries(keys, **options)
           end
 
           def write_entry(key, entry, **options)
-            entry.dup_value!
-            @data[key] = entry
+            @data[key] = Entry.build(entry)
             true
           end
 
@@ -75,10 +146,7 @@ def delete_entry(key, **options)
           end
 
           def fetch_entry(key, options = nil) # :nodoc:
-            entry = @data.fetch(key) { @data[key] = yield }
-            dup_entry = entry.dup
-            dup_entry&.dup_value!
-            dup_entry
+            @data.fetch(key) { @data[key] = Entry.build(yield) }
           end
         end
 
@@ -131,12 +199,12 @@ def decrement(name, amount = 1, **options) # :nodoc:
           def read_entry(key, **options)
             if cache = local_cache
               hit = true
-              value = cache.fetch_entry(key) do
+              entry = cache.fetch_entry(key) do
                 hit = false
                 super
               end
               options[:event][:store] = cache.class.name if hit && options[:event]
-              value
+              entry
             else
               super
             end
@@ -162,7 +230,12 @@ def write_entry(key, entry, **options)
               local_cache.write_entry(key, entry, **options) if local_cache
             end
 
-            super
+
+            if entry.local?
+              super(key, new_entry(entry.value, options), **options)
+            else
+              super
+            end
           end
 
           def delete_entry(key, **options)