Lint new Clock creation

Author: devonh

scripts-dev/mypy_synapse_plugin.py

@@ -92,6 +92,12 @@
     category="synapse-reactor-clock",
 )
 
+MULTIPLE_INTERNAL_CLOCKS_CREATED = ErrorCode(
+    "multiple-internal-clocks",
+    "Only one instance of `clock.Clock` should be created",
+    category="synapse-reactor-clock",
+)
+
 
 class Sentinel(enum.Enum):
     # defining a sentinel in this way allows mypy to correctly handle the
@@ -237,6 +243,9 @@ def get_function_signature_hook(
         if fullname == "twisted.internet.task.LoopingCall":
             return check_looping_call
 
+        if fullname == "synapse.util.clock.Clock":
+            return check_clock_creation
+
         return None
 
     def get_method_signature_hook(
@@ -331,6 +340,25 @@ def check_looping_call(ctx: FunctionSigContext) -> CallableType:
     return signature
 
 
+def check_clock_creation(ctx: FunctionSigContext) -> CallableType:
+    """
+    Ensure that the only `clock.Clock` is the one used by the `HomeServer`.
+
+    Args:
+        ctx: The `FunctionSigContext` from mypy.
+    """
+    signature: CallableType = ctx.default_signature
+    ctx.api.fail(
+        "Expected the only `clock.Clock` instance to be the one used by the `HomeServer`. "
+        "This is so that the `HomeServer` can cancel any tracked delayed or looping calls "
+        "during server shutdown",
+        ctx.context,
+        code=MULTIPLE_INTERNAL_CLOCKS_CREATED,
+    )
+
+    return signature
+
+
 def check_call_when_running(ctx: MethodSigContext) -> CallableType:
     """
     Ensure that the `reactor.callWhenRunning` callsites aren't used.

synapse/appservice/__init__.py

@@ -23,15 +23,33 @@
 import logging
 import re
 from enum import Enum
-from typing import TYPE_CHECKING, Dict, Iterable, List, Optional, Pattern, Sequence
+from typing import (
+    TYPE_CHECKING,
+    Dict,
+    Iterable,
+    List,
+    Optional,
+    Pattern,
+    Sequence,
+    cast,
+)
 
 import attr
 from netaddr import IPSet
 
+from twisted.internet import reactor
+
 from synapse.api.constants import EventTypes
 from synapse.events import EventBase
-from synapse.types import DeviceListUpdates, JsonDict, JsonMapping, UserID
+from synapse.types import (
+    DeviceListUpdates,
+    ISynapseThreadlessReactor,
+    JsonDict,
+    JsonMapping,
+    UserID,
+)
 from synapse.util.caches.descriptors import _CacheContext, cached
+from synapse.util.clock import Clock
 
 if TYPE_CHECKING:
     from synapse.appservice.api import ApplicationServiceApi
@@ -98,6 +116,13 @@ def __init__(
         self.sender = sender
         # The application service user should be part of the server's domain.
         self.server_name = sender.domain  # nb must be called this for @cached
+
+        # Ideally we would require passing in the `HomeServer` `Clock` instance.
+        # However this is not currently possible as there are places which use
+        # `@cached` that aren't aware of the `HomeServer` instance.
+        # nb must be called this for @cached
+        self.clock = Clock(cast(ISynapseThreadlessReactor, reactor))  # type: ignore[multiple-internal-clocks]
+
         self.namespaces = self._check_namespaces(namespaces)
         self.id = id
         self.ip_range_whitelist = ip_range_whitelist

synapse/handlers/profile.py

@@ -56,8 +56,8 @@ class ProfileHandler:
 
     def __init__(self, hs: "HomeServer"):
         self.server_name = hs.hostname  # nb must be called this for @cached
+        self.clock = hs.get_clock()  # nb must be called this for @cached
         self.store = hs.get_datastores().main
-        self.clock = hs.get_clock()
         self.hs = hs
 
         self.federation = hs.get_federation_client()

synapse/handlers/sync.py

@@ -982,6 +982,7 @@ def get_lazy_loaded_members_cache(
             logger.debug("creating LruCache for %r", cache_key)
             cache = LruCache(
                 max_size=LAZY_LOADED_MEMBERS_CACHE_MAX_SIZE,
+                clock=self.clock,
             )
             self.lazy_loaded_members_cache[cache_key] = cache
         else:

synapse/rest/client/sync.py

@@ -126,6 +126,7 @@ def __init__(self, hs: "HomeServer"):
 
         self._json_filter_cache: LruCache[str, bool] = LruCache(
             max_size=1000,
+            clock=self.clock,
             cache_name="sync_valid_filter",
             server_name=self.server_name,
         )

synapse/server_notices/server_notices_manager.py

@@ -36,6 +36,7 @@
 class ServerNoticesManager:
     def __init__(self, hs: "HomeServer"):
         self.server_name = hs.hostname  # nb must be called this for @cached
+        self.clock = hs.get_clock()  # nb must be called this for @cached
         self._store = hs.get_datastores().main
         self._config = hs.config
         self._account_data_handler = hs.get_account_data_handler()

synapse/storage/_base.py

@@ -56,7 +56,7 @@ def __init__(
     ):
         self.hs = hs
         self.server_name = hs.hostname  # nb must be called this for @cached
-        self._clock = hs.get_clock()
+        self.clock = hs.get_clock()  # nb must be called this for @cached
         self.database_engine = database.engine
         self.db_pool = database
 

synapse/storage/controllers/state.py

@@ -69,8 +69,8 @@ class StateStorageController:
 
     def __init__(self, hs: "HomeServer", stores: "Databases"):
         self.server_name = hs.hostname  # nb must be called this for @cached
+        self.clock = hs.get_clock()
         self._is_mine_id = hs.is_mine_id
-        self._clock = hs.get_clock()
         self.stores = stores
         self._partial_state_events_tracker = PartialStateEventsTracker(stores.main)
         self._partial_state_room_tracker = PartialCurrentStateTracker(stores.main)
@@ -818,9 +818,7 @@ async def get_joined_hosts(
             state_group = object()
 
         assert state_group is not None
-        with Measure(
-            self._clock, name="get_joined_hosts", server_name=self.server_name
-        ):
+        with Measure(self.clock, name="get_joined_hosts", server_name=self.server_name):
             return await self._get_joined_hosts(
                 room_id, state_group, state_entry=state_entry
             )

synapse/storage/databases/main/cache.py

@@ -751,7 +751,7 @@ def _send_invalidation_to_replication(
                     "instance_name": self._instance_name,
                     "cache_func": cache_name,
                     "keys": keys,
-                    "invalidation_ts": self._clock.time_msec(),
+                    "invalidation_ts": self.clock.time_msec(),
                 },
             )
 
@@ -778,7 +778,7 @@ def _send_invalidation_to_replication_bulk(
             assert self._cache_id_gen is not None
 
             stream_ids = self._cache_id_gen.get_next_mult_txn(txn, len(key_tuples))
-            ts = self._clock.time_msec()
+            ts = self.clock.time_msec()
             txn.call_after(self.hs.get_notifier().on_new_replication_data)
             self.db_pool.simple_insert_many_txn(
                 txn,

synapse/storage/databases/main/censor_events.py

@@ -77,7 +77,7 @@ async def _censor_redactions(self) -> None:
             return
 
         before_ts = (
-            self._clock.time_msec() - self.hs.config.server.redaction_retention_period
+            self.clock.time_msec() - self.hs.config.server.redaction_retention_period
         )
 
         # We fetch all redactions that: