FIX: Treat corrupt cache as cache miss

Currently when a cache entry is corrupt, we log the event without doing
anything else. It means the cache is still corrupt, and the proper value
isn’t computed again.

Normally, it’s very rare the cache becomes corrupt, but it can happen
when upgrading Rails for example and the cache format changes. This is
normally handled automatically by Rails but since we’re using a custom
cache class, we have to do it ourselves.

This patch takes the same approach the Rails team did, when a cache
entry is corrupt, we treat it as a miss, recomputing the proper value
and caching it in the new format.

lib/cache.rb

@@ -61,7 +61,7 @@ class Cache
   # this removes a bunch of stuff we do not need like instrumentation and versioning
   def read(name)
     key = normalize_key(name)
-    read_entry(key)
+    read_entry(key).tap { |entry| break if entry == :__corrupt_cache__ }
   end
 
   def write(name, value, expires_in: nil)
@@ -73,38 +73,30 @@ class Cache
   end
 
   def fetch(name, expires_in: nil, force: nil, &blk)
-    if block_given?
-      key = normalize_key(name)
-      raw = nil
-
-      raw = redis.get(key) if !force
-
-      if raw
-        begin
-          Marshal.load(raw) # rubocop:disable Security/MarshalLoad
-        rescue => e
-          log_first_exception(e)
-        end
-      else
-        val = blk.call
-        write_entry(key, val, expires_in: expires_in)
-        val
+    if !block_given?
+      if force
+        raise ArgumentError,
+              "Missing block: Calling `Cache#fetch` with `force: true` requires a block."
       end
-    elsif force
-      raise ArgumentError,
-            "Missing block: Calling `Cache#fetch` with `force: true` requires a block."
-    else
-      read(name)
+      return read(name)
     end
+
+    key = normalize_key(name)
+    raw = redis.get(key) if !force
+    entry = read_entry(key) if raw
+    return entry if raw && !(entry == :__corrupt_cache__)
+
+    val = blk.call
+    write_entry(key, val, expires_in: expires_in)
+    val
   end
 
   protected
 
-  def log_first_exception(e)
-    if !defined?(@logged_a_warning)
-      @logged_a_warning = true
-      Discourse.warn_exception(e, "Corrupt cache... skipping entry for key #{key}")
-    end
+  def log_first_exception(e, key)
+    return if defined?(@logged_a_warning)
+    @logged_a_warning = true
+    Discourse.warn_exception(e, message: "Corrupt cache... skipping entry for key #{key}")
   end
 
   def read_entry(key)
@@ -116,7 +108,8 @@ class Cache
     # changes. Log it once so we can tell it is happening.
     # should not happen under any normal circumstances, but we
     # do not want to flood logs
-    log_first_exception(e)
+    log_first_exception(e, key)
+    :__corrupt_cache__
   end
 
   def write_entry(key, value, expires_in: nil)

spec/lib/cache_spec.rb

@@ -3,9 +3,7 @@
 require "cache"
 
 RSpec.describe Cache do
-  let :cache do
-    Cache.new
-  end
+  subject(:cache) { Cache.new }
 
   it "supports exist?" do
     cache.write("testing", 1.1)
@@ -104,4 +102,24 @@ RSpec.describe Cache do
 
     expect(cache.redis.ttl("#{cache.namespace}:foo:bar")).to eq(180)
   end
+
+  describe ".fetch" do
+    subject(:fetch_value) { cache.fetch("my_key") { "bob" } }
+
+    context "when the cache is corrupt" do
+      before do
+        cache.delete("my_key")
+        Discourse.redis.setex(cache.normalize_key("my_key"), described_class::MAX_CACHE_AGE, "")
+      end
+
+      it "runs and return the provided block" do
+        expect(fetch_value).to eq("bob")
+      end
+
+      it "generates a new cache entry" do
+        fetch_value
+        expect(cache.read("my_key")).to eq("bob")
+      end
+    end
+  end
 end