Harden ResponseCache (and fix #9507)

changelog.d/9522.bugfix

@@ -0,0 +1 @@
+Fix #9507 by hardening ResponseCache against errors in execution of to-be-cached functions.

synapse/util/caches/response_cache.py

@@ -16,6 +16,7 @@
 from typing import TYPE_CHECKING, Any, Callable, Dict, Generic, Optional, Set, TypeVar
 
 from twisted.internet import defer
+from twisted.python.failure import Failure
 
 from synapse.logging.context import make_deferred_yieldable, run_in_background
 from synapse.util.async_helpers import ObservableDeferred
@@ -80,6 +81,42 @@ def get(self, key: T) -> Optional[defer.Deferred]:
             self._metrics.inc_misses()
             return None
 
+    @staticmethod
+    def _safe_call(func: Callable[[Any], bool], ret_val: Any, key: T) -> bool:
+        """
+        Wraps a conditional checking function with a try-catch, returns False
+        if the conditional function threw an exception during evaluation.
+        """
+        try:
+            return func(ret_val)
+        except Exception as e:
+            logger.warning(
+                "conditional %r threw exception on %s, ignoring and invalidating cache...",
+                func,
+                key,
+                exc_info=e,
+            )
+            return False
+
+    def _remove_cache(self, key: T):
+        self.pending_result_cache.pop(key, None)
+
+    def _errback_remove_cache(self, f: Failure, key: T):
+        self._remove_cache(key)
+        return f
+
+    def _test_cache(self, r: Any, key: T):
+        should_cache = all(
+            self._safe_call(func, r, key=key)
+            for func in self.pending_conditionals.pop(key, [])
+        )
+
+        if self.timeout_sec and should_cache:
+            self.clock.call_later(self.timeout_sec, self._remove_cache, key)
+        else:
+            self._remove_cache(key)
+        return r
+
     def set(self, key: T, deferred: defer.Deferred) -> defer.Deferred:
         """Set the entry for the given key to the given deferred.
 
@@ -101,20 +138,8 @@ def set(self, key: T, deferred: defer.Deferred) -> defer.Deferred:
         result = ObservableDeferred(deferred, consumeErrors=True)
         self.pending_result_cache[key] = result
 
-        def remove(r):
-            should_cache = all(
-                func(r) for func in self.pending_conditionals.pop(key, [])
-            )
-
-            if self.timeout_sec and should_cache:
-                self.clock.call_later(
-                    self.timeout_sec, self.pending_result_cache.pop, key, None
-                )
-            else:
-                self.pending_result_cache.pop(key, None)
-            return r
-
-        result.addBoth(remove)
+        result.addCallback(self._test_cache, key)
+        result.addErrback(self._errback_remove_cache, key)
         return result.observe()
 
     def add_conditional(self, key: T, conditional: Callable[[Any], bool]):
@@ -130,8 +155,12 @@ def wrap_conditional(
     ) -> defer.Deferred:
         """The same as wrap(), but adds a conditional to the final execution.
 
-        When the final execution completes, *all* conditionals need to return True for it to properly cache,
-        else it'll not be cached in a timed fashion.
+        When the final execution completes, *all* conditionals need to return
+        True for it to properly cache (apart from it also not having thrown an
+        uncaught exception), else the cache will instantly expire.
+
+        (Note that also no conditional must throw an uncaught exception for it to
+        cache)
         """
 
         # See if there's already a result on this key that hasn't yet completed. Due to the single-threaded nature of
@@ -145,7 +174,9 @@ def wrap_conditional(
     def wrap(
         self, key: T, callback: "Callable[..., Any]", *args: Any, **kwargs: Any
     ) -> defer.Deferred:
-        """Wrap together a *get* and *set* call, taking care of logcontexts
+        """Wrap together a *get* and *set* call, taking care of logcontexts.
+
+        Does not cache if the wrapped function throws an uncaught exception.
 
         First looks up the key in the cache, and if it is present makes it
         follow the synapse logcontext rules and returns it.