Allow long lived syncs to be cancelled if client has gone away (#19499)

erikjohnston · web-flow · commit 2c73e8daef1c · 2026-02-26T21:41:06.000Z
diff --git a/changelog.d/19499.misc b/changelog.d/19499.misc
@@ -0,0 +1 @@
+Cancel long-running sync requests if the client has gone away.
diff --git a/pyproject.toml b/pyproject.toml
@@ -411,6 +411,12 @@ indent-style = "space"
 skip-magic-trailing-comma = false
 line-ending = "auto"
 
+[tool.ruff.lint.flake8-bugbear]
+extend-immutable-calls = [
+    # Durations are immutable
+    "synapse.util.duration.Duration",
+]
+
 [tool.maturin]
 manifest-path = "rust/Cargo.toml"
 module-name = "synapse.synapse_rust"
diff --git a/synapse/api/auth/mas.py b/synapse/api/auth/mas.py
@@ -45,6 +45,7 @@
 from synapse.types import JsonDict, Requester, UserID, create_requester
 from synapse.util.caches.cached_call import RetryOnExceptionCachedCall
 from synapse.util.caches.response_cache import ResponseCache, ResponseCacheContext
+from synapse.util.duration import Duration
 from synapse.util.json import json_decoder
 
 from . import introspection_response_timer
@@ -139,7 +140,7 @@ def __init__(self, hs: "HomeServer"):
             clock=self._clock,
             name="mas_token_introspection",
             server_name=self.server_name,
-            timeout_ms=120_000,
+            timeout=Duration(minutes=2),
             # don't log because the keys are access tokens
             enable_logging=False,
         )
diff --git a/synapse/api/auth/msc3861_delegated.py b/synapse/api/auth/msc3861_delegated.py
@@ -49,6 +49,7 @@
 from synapse.types import Requester, UserID, create_requester
 from synapse.util.caches.cached_call import RetryOnExceptionCachedCall
 from synapse.util.caches.response_cache import ResponseCache, ResponseCacheContext
+from synapse.util.duration import Duration
 from synapse.util.json import json_decoder
 
 from . import introspection_response_timer
@@ -205,7 +206,7 @@ def __init__(self, hs: "HomeServer"):
             clock=self._clock,
             name="token_introspection",
             server_name=self.server_name,
-            timeout_ms=120_000,
+            timeout=Duration(minutes=2),
             # don't log because the keys are access tokens
             enable_logging=False,
         )
diff --git a/synapse/appservice/api.py b/synapse/appservice/api.py
@@ -46,6 +46,7 @@
 from synapse.metrics import SERVER_NAME_LABEL
 from synapse.types import DeviceListUpdates, JsonDict, JsonMapping, ThirdPartyInstanceID
 from synapse.util.caches.response_cache import ResponseCache
+from synapse.util.duration import Duration
 
 if TYPE_CHECKING:
     from synapse.server import HomeServer
@@ -132,7 +133,7 @@ def __init__(self, hs: "HomeServer"):
             clock=hs.get_clock(),
             name="as_protocol_meta",
             server_name=self.server_name,
-            timeout_ms=HOUR_IN_MS,
+            timeout=Duration(hours=1),
         )
 
     def _get_headers(self, service: "ApplicationService") -> dict[bytes, list[bytes]]:
diff --git a/synapse/config/cache.py b/synapse/config/cache.py
@@ -29,6 +29,7 @@
 
 from synapse.types import JsonDict
 from synapse.util.check_dependencies import check_requirements
+from synapse.util.duration import Duration
 
 from ._base import Config, ConfigError
 
@@ -108,7 +109,7 @@ class CacheConfig(Config):
     global_factor: float
     track_memory_usage: bool
     expiry_time_msec: int | None
-    sync_response_cache_duration: int
+    sync_response_cache_duration: Duration
 
     @staticmethod
     def reset() -> None:
@@ -207,10 +208,14 @@ def read_config(self, config: JsonDict, **kwargs: Any) -> None:
             min_cache_ttl = self.cache_autotuning.get("min_cache_ttl")
             self.cache_autotuning["min_cache_ttl"] = self.parse_duration(min_cache_ttl)
 
-        self.sync_response_cache_duration = self.parse_duration(
+        sync_response_cache_duration_ms = self.parse_duration(
             cache_config.get("sync_response_cache_duration", "2m")
         )
 
+        self.sync_response_cache_duration = Duration(
+            milliseconds=sync_response_cache_duration_ms
+        )
+
     def resize_all_caches(self) -> None:
         """Ensure all cache sizes are up-to-date.
 
diff --git a/synapse/federation/federation_server.py b/synapse/federation/federation_server.py
@@ -166,7 +166,7 @@ def __init__(self, hs: "HomeServer"):
             clock=hs.get_clock(),
             name="fed_txn_handler",
             server_name=self.server_name,
-            timeout_ms=30000,
+            timeout=Duration(seconds=30),
         )
 
         self.transaction_actions = TransactionActions(self.store)
@@ -179,13 +179,13 @@ def __init__(self, hs: "HomeServer"):
             clock=hs.get_clock(),
             name="state_resp",
             server_name=self.server_name,
-            timeout_ms=30000,
+            timeout=Duration(seconds=30),
         )
         self._state_ids_resp_cache: ResponseCache[tuple[str, str]] = ResponseCache(
             clock=hs.get_clock(),
             name="state_ids_resp",
             server_name=self.server_name,
-            timeout_ms=30000,
+            timeout=Duration(seconds=30),
         )
 
         self._federation_metrics_domains = (
diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py
@@ -185,7 +185,7 @@ def __init__(self, hs: "HomeServer"):
             clock=hs.get_clock(),
             name="room_upgrade",
             server_name=self.server_name,
-            timeout_ms=FIVE_MINUTES_IN_MS,
+            timeout=Duration(minutes=5),
         )
         self._server_notices_mxid = hs.config.servernotices.server_notices_mxid
 
diff --git a/synapse/handlers/room_list.py b/synapse/handlers/room_list.py
@@ -44,6 +44,7 @@
 from synapse.types import JsonDict, JsonMapping, ThirdPartyInstanceID
 from synapse.util.caches.descriptors import _CacheContext, cached
 from synapse.util.caches.response_cache import ResponseCache
+from synapse.util.duration import Duration
 
 if TYPE_CHECKING:
     from synapse.server import HomeServer
@@ -79,7 +80,7 @@ def __init__(self, hs: "HomeServer"):
             clock=hs.get_clock(),
             name="remote_room_list",
             server_name=self.server_name,
-            timeout_ms=30 * 1000,
+            timeout=Duration(seconds=30),
         )
 
     async def get_local_public_room_list(
diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py
@@ -77,6 +77,7 @@
 from synapse.util.caches.expiringcache import ExpiringCache
 from synapse.util.caches.lrucache import LruCache
 from synapse.util.caches.response_cache import ResponseCache, ResponseCacheContext
+from synapse.util.cancellation import cancellable
 from synapse.util.metrics import Measure
 from synapse.visibility import filter_and_transform_events_for_client
 
@@ -307,7 +308,7 @@ def __init__(self, hs: "HomeServer"):
             clock=hs.get_clock(),
             name="sync",
             server_name=self.server_name,
-            timeout_ms=hs.config.caches.sync_response_cache_duration,
+            timeout=hs.config.caches.sync_response_cache_duration,
         )
 
         # ExpiringCache((User, Device)) -> LruCache(user_id => event_id)
@@ -367,6 +368,10 @@ async def wait_for_sync_for_user(
         logger.debug("Returning sync response for %s", user_id)
         return res
 
+    # TODO: We mark this as cancellable, and we have tests for it, but we
+    # haven't gone through and exhaustively checked that all the code paths in
+    # this method are actually cancellable.
+    @cancellable
     async def _wait_for_sync_for_user(
         self,
         sync_config: SyncConfig,
diff --git a/synapse/handlers/typing.py b/synapse/handlers/typing.py
@@ -102,7 +102,7 @@ def __init__(self, hs: "HomeServer"):
         self._room_typing: dict[str, set[str]] = {}
 
         self._member_last_federation_poke: dict[RoomMember, int] = {}
-        self.wheel_timer: WheelTimer[RoomMember] = WheelTimer(bucket_size=5000)
+        self.wheel_timer: WheelTimer[RoomMember] = WheelTimer()
         self._latest_room_serial = 0
 
         self._rooms_updated: set[str] = set()
@@ -120,7 +120,7 @@ def _reset(self) -> None:
         self._rooms_updated = set()
 
         self._member_last_federation_poke = {}
-        self.wheel_timer = WheelTimer(bucket_size=5000)
+        self.wheel_timer = WheelTimer()
 
     @wrap_as_background_process("typing._handle_timeouts")
     async def _handle_timeouts(self) -> None:
diff --git a/synapse/replication/http/_base.py b/synapse/replication/http/_base.py
@@ -130,7 +130,7 @@ def __init__(self, hs: "HomeServer"):
                 clock=hs.get_clock(),
                 name="repl." + self.NAME,
                 server_name=self.server_name,
-                timeout_ms=30 * 60 * 1000,
+                timeout=Duration(minutes=30),
             )
 
         # We reserve `instance_name` as a parameter to sending requests, so we
diff --git a/synapse/rest/client/sync.py b/synapse/rest/client/sync.py
@@ -59,6 +59,7 @@
 from synapse.types import JsonDict, Requester, SlidingSyncStreamToken, StreamToken
 from synapse.types.rest.client import SlidingSyncBody
 from synapse.util.caches.lrucache import LruCache
+from synapse.util.cancellation import cancellable
 from synapse.util.json import json_decoder
 
 from ._base import client_patterns, set_timeline_upper_limit
@@ -138,6 +139,7 @@ def __init__(self, hs: "HomeServer"):
             cfg=hs.config.ratelimiting.rc_presence_per_user,
         )
 
+    @cancellable
     async def on_GET(self, request: SynapseRequest) -> tuple[int, JsonDict]:
         # This will always be set by the time Twisted calls us.
         assert request.args is not None
diff --git a/synapse/util/async_helpers.py b/synapse/util/async_helpers.py
@@ -57,6 +57,7 @@
     run_coroutine_in_background,
     run_in_background,
 )
+from synapse.util.cancellation import cancellable
 from synapse.util.clock import Clock
 from synapse.util.duration import Duration
 
@@ -83,6 +84,13 @@ def observe(self) -> "defer.Deferred[_T]":
         """
         ...
 
+    @abc.abstractmethod
+    def has_observers(self) -> bool:
+        """Returns True if there are any observers currently observing this
+        ObservableDeferred.
+        """
+        ...
+
 
 class ObservableDeferred(Generic[_T], AbstractObservableDeferred[_T]):
     """Wraps a deferred object so that we can add observer deferreds. These
@@ -122,6 +130,11 @@ def callback(r: _T) -> _T:
             for observer in observers:
                 try:
                     observer.callback(r)
+                except defer.CancelledError:
+                    # We do not want to propagate cancellations to the original
+                    # deferred, or to other observers, so we can just ignore
+                    # this.
+                    pass
                 except Exception as e:
                     logger.exception(
                         "%r threw an exception on .callback(%r), ignoring...",
@@ -145,6 +158,11 @@ def errback(f: Failure) -> Failure | None:
                 f.value.__failure__ = f
                 try:
                     observer.errback(f)
+                except defer.CancelledError:
+                    # We do not want to propagate cancellations to the original
+                    # deferred, or to other observers, so we can just ignore
+                    # this.
+                    pass
                 except Exception as e:
                     logger.exception(
                         "%r threw an exception on .errback(%r), ignoring...",
@@ -160,6 +178,7 @@ def errback(f: Failure) -> Failure | None:
 
         deferred.addCallbacks(callback, errback)
 
+    @cancellable
     def observe(self) -> "defer.Deferred[_T]":
         """Observe the underlying deferred.
 
@@ -169,7 +188,7 @@ def observe(self) -> "defer.Deferred[_T]":
         """
         if not self._result:
             assert isinstance(self._observers, list)
-            d: "defer.Deferred[_T]" = defer.Deferred()
+            d: "defer.Deferred[_T]" = defer.Deferred(canceller=self._remove_observer)
             self._observers.append(d)
             return d
         elif self._result[0]:
@@ -180,6 +199,12 @@ def observe(self) -> "defer.Deferred[_T]":
     def observers(self) -> "Collection[defer.Deferred[_T]]":
         return self._observers
 
+    def has_observers(self) -> bool:
+        """Returns True if there are any observers currently observing this
+        ObservableDeferred.
+        """
+        return bool(self._observers)
+
     def has_called(self) -> bool:
         return self._result is not None
 
@@ -204,6 +229,28 @@ def __repr__(self) -> str:
             self._deferred,
         )
 
+    def _remove_observer(self, observer: "defer.Deferred[_T]") -> None:
+        """Removes an observer from the list of observers.
+
+        Used as a canceller for the observer deferreds, so that if an observer
+        is cancelled it is removed from the list of observers.
+        """
+        if self._result is not None:
+            # The underlying deferred has already resolved, so the observer has
+            # already been resolved. Nothing to do.
+            return
+
+        assert isinstance(self._observers, list)
+        try:
+            self._observers.remove(observer)
+        except ValueError:
+            # The observer was not in the list. This can happen if the underlying
+            # deferred resolves at around the same time as we try to remove the
+            # observer. In this case, it's possible that we tried to remove the
+            # observer just after it was added to the list, but before it was
+            # resolved and removed from the list by the callback/errback above.
+            pass
+
 
 T = TypeVar("T")
 
@@ -962,6 +1009,27 @@ def handle_cancel(new_deferred: "defer.Deferred[T]") -> None:
     return new_deferred
 
 
+def observe_deferred(d: "defer.Deferred[T]") -> "defer.Deferred[T]":
+    """Returns a new `Deferred` that observes the given `Deferred`.
+
+    The returned `Deferred` will resolve with the same result as the given
+    `Deferred`, but will not "chain" on the deferred so that using the returned
+    deferred does not affect the given `Deferred` in any way.
+    """
+    new_deferred: "defer.Deferred[T]" = defer.Deferred()
+
+    def callback(r: T) -> T:
+        new_deferred.callback(r)
+        return r
+
+    def errback(f: Failure) -> Failure:
+        new_deferred.errback(f)
+        return f
+
+    d.addCallbacks(callback, errback)
+    return new_deferred
+
+
 class AwakenableSleeper:
     """Allows explicitly waking up deferreds related to an entity that are
     currently sleeping.
diff --git a/synapse/util/caches/response_cache.py b/synapse/util/caches/response_cache.py
diff --git a/synapse/util/clock.py b/synapse/util/clock.py
diff --git a/synapse/util/wheel_timer.py b/synapse/util/wheel_timer.py
diff --git a/tests/rest/client/test_sync.py b/tests/rest/client/test_sync.py
diff --git a/tests/util/caches/test_response_cache.py b/tests/util/caches/test_response_cache.py
diff --git a/tests/util/test_async_helpers.py b/tests/util/test_async_helpers.py
diff --git a/tests/util/test_wheel_timer.py b/tests/util/test_wheel_timer.py

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1 @@`
	`1`	`+Cancel long-running sync requests if the client has gone away.`
Original file line number	Diff line number	Diff line change
`@@ -185,7 +185,7 @@ def __init__(self, hs: "HomeServer"):`
`185`	`185`	`clock=hs.get_clock(),`
`186`	`186`	`name="room_upgrade",`
`187`	`187`	`server_name=self.server_name,`
`188`		`- timeout_ms=FIVE_MINUTES_IN_MS,`
	`188`	`+ timeout=Duration(minutes=5),`
`189`	`189`	`)`
`190`	`190`	`self._server_notices_mxid = hs.config.servernotices.server_notices_mxid`
`191`	`191`