Fix /sync caching transient errors for the sync_response_cache_duration. (#19845)

Fixes: #19844

Concretely, this changes `ResponseCache` to unset cache entries once
they resolve to a `Failure`.

---------

Signed-off-by: Olivier 'reivilibre <oliverw@matrix.org>
Co-authored-by: Eric Eastwood <erice@element.io>

Author: reivilibre

changelog.d/19845.bugfix

@@ -0,0 +1 @@
+Fix `/sync` caching transient errors for the `sync_response_cache_duration`.

synapse/util/async_helpers.py

@@ -98,6 +98,8 @@ class ObservableDeferred(Generic[_T], AbstractObservableDeferred[_T]):
     deferred.
 
     If consumeErrors is true errors will be captured from the origin deferred.
+    In that case, errors will NOT be propagated onto any errbacks added to this
+    `ObservableDeferred`; instead the success callbacks will be called with `None`.
 
     Cancelling or otherwise resolving an observer will not affect the original
     ObservableDeferred.

synapse/util/caches/response_cache.py

@@ -32,6 +32,7 @@
 import attr
 
 from twisted.internet import defer
+from twisted.python.failure import Failure
 
 from synapse.logging.context import make_deferred_yieldable, run_in_background
 from synapse.logging.opentracing import (
@@ -226,14 +227,19 @@ def _set(
         Returns:
             The cache entry object.
         """
-        result = ObservableDeferred(deferred, consumeErrors=True)
+        result = ObservableDeferred(
+            deferred,
+            # We set `consumeErrors=False` as we want to handle errors ourselves (`on_fail`) instead of
+            # replacing them with a `None` successful result that would go to `on_succeed`
+            consumeErrors=False,
+        )
         key = context.cache_key
         entry = ResponseCacheEntry(
             result, opentracing_span_context, cancellable=cancellable
         )
         self._result_cache[key] = entry
 
-        def on_complete(r: RV) -> RV:
+        def on_succeed(r: RV) -> RV:
             # if this cache has a non-zero timeout, and the callback has not cleared
             # the should_cache bit, we leave it in the cache for now and schedule
             # its removal later.
@@ -254,10 +260,29 @@ def on_complete(r: RV) -> RV:
                 self.unset(key)
             return r
 
-        # make sure we do this *after* adding the entry to result_cache,
-        # in case the result is already complete (in which case flipping the order would
-        # leave us with a stuck entry in the cache).
-        result.addBoth(on_complete)
+        def on_fail(failure: Failure) -> None:
+            """
+            If the deferred fails, unset the cache entry.
+            """
+            self.unset(key)
+
+            # Consider the Failure handled so they don't get thrown by the reactor
+            return None
+
+        # Two ordering constraints to be aware of for registering the callback and errback:
+        # 1. Both of them must be registered _after_ adding the entry to the result_cache,
+        #    otherwise it is possible for them to trigger immediately (before the entry
+        #    is added to the result_cache), the net effect of which is that it would leave
+        #    us with a stuck entry in the cache.
+        # 2. We must register the errback after callback.
+        #    If the errback was registered first, `on_fail` returning `None` would
+        #    cause `on_succeed` to be called with that `None` as argument.
+        #    This would start a timer to remove a cache entry, even though there isn't a
+        #    valid cache entry yet.
+        #    (This could prematurely remove a future cache entry with the same key.)
+        result.addCallback(on_succeed)
+        result.addErrback(on_fail)
+
         return entry
 
     def unset(self, key: KV) -> None:

tests/util/caches/test_response_cache.py

@@ -3,6 +3,7 @@
 #
 # Copyright 2021 The Matrix.org Foundation C.I.C.
 # Copyright (C) 2023 New Vector, Ltd
+# Copyright (C) 2026 Element Creations Ltd.
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License as
@@ -15,7 +16,8 @@
 # Originally licensed under the Apache License, Version 2.0:
 # <http://www.apache.org/licenses/LICENSE-2.0>.
 #
-# [This file includes modifications made by New Vector Limited]
+# [This file includes modifications made by New Vector Limited
+# and Element Creations Ltd]
 #
 #
 
@@ -234,9 +236,9 @@ async def non_caching(o: str, cache_context: ResponseCacheContext[int]) -> str:
                 [], cache.keys(), "cache should not have the result now"
             )
 
-    def test_cache_func_errors(self) -> None:
+    def test_errors_raised_to_all_waiters(self) -> None:
         """If the callback raises an error, the error should be raised to all
-        callers and the result should not be cached"""
+        concurrent callers that were waiting on the same in-flight result."""
         cache = self.with_cache("error_cache", ms=3000)
 
         expected_error = Exception("oh no")
@@ -259,6 +261,60 @@ async def erring(o: str) -> str:
         self.assertFailure(wrap_d, Exception)
         self.assertFailure(wrap2_d, Exception)
 
+    def test_errors_are_not_cached(self) -> None:
+        """If the callback raises an error, the error is not cached and
+        served to any subsequent requests.
+        """
+        cache = self.with_cache("error_not_cached", ms=3000)
+
+        return_error = True
+
+        REQUEST_FAKE_WORK_SLEEP_TIME = Duration(seconds=1)
+
+        async def erring_then_fine(_: str) -> str:
+            """
+            This function raises an error the first time it is called,
+            then is fine the next time it is called.
+            """
+            nonlocal return_error
+
+            # pretend to do some work
+            await self.clock.sleep(REQUEST_FAKE_WORK_SLEEP_TIME)
+
+            if return_error:
+                return_error = False
+                raise RuntimeError("this is a temporary error!")
+            return "fine"
+
+        # First call: should get an error
+        wrap_d = defer.ensureDeferred(cache.wrap(0, erring_then_fine, "ignored"))
+
+        # Should be pending
+        self.assertNoResult(wrap_d)
+
+        # Wait for the time it takes for the request to resolve to an error
+        self.reactor.advance(REQUEST_FAKE_WORK_SLEEP_TIME.as_secs())
+
+        # Check that the Deferred resolved to an error of the correct type
+        self.failureResultOf(wrap_d, RuntimeError)
+
+        # Second call: the error shouldn't be replayed
+        # and the next call should succeed, so we should get a successful response
+        wrap2_d = defer.ensureDeferred(cache.wrap(0, erring_then_fine, "ignored"))
+
+        # Since this is a new request (not coalesced with the previous failed one),
+        # this should still be pending
+        self.assertNoResult(wrap2_d)
+
+        # Wait for the time it takes for the request to resolve
+        self.reactor.advance(REQUEST_FAKE_WORK_SLEEP_TIME.as_secs())
+
+        self.assertEqual(
+            self.successResultOf(wrap2_d),
+            "fine",
+            "should get the fresh result, not the cached error",
+        )
+
     def test_cache_cancel_first_wait(self) -> None:
         """Test that cancellation of the deferred returned by wrap() on the
         first call does not immediately cause a cancellation error to be raised