WIP on clean shutdown of SynapseHomeServer class

Author: devonh

synapse/api/auth/base.py

@@ -19,6 +19,7 @@
 #
 #
 import logging
+import weakref
 from typing import TYPE_CHECKING, Optional, Tuple
 
 from netaddr import IPAddress
@@ -52,7 +53,7 @@ class BaseAuth:
     """Common base class for all auth implementations."""
 
     def __init__(self, hs: "HomeServer"):
-        self.hs = hs
+        self.hs = weakref.proxy(hs)
         self.store = hs.get_datastores().main
         self._storage_controllers = hs.get_storage_controllers()
 

synapse/api/auth_blocking.py

@@ -20,6 +20,7 @@
 #
 
 import logging
+import weakref
 from typing import TYPE_CHECKING, Optional
 
 from synapse.api.constants import LimitBlockingTypes, UserTypes
@@ -46,7 +47,7 @@ def __init__(self, hs: "HomeServer"):
         self._mau_limits_reserved_threepids = (
             hs.config.server.mau_limits_reserved_threepids
         )
-        self._is_mine_server_name = hs.is_mine_server_name
+        self.hs = weakref.proxy(hs)
         self._track_appservice_user_ips = hs.config.appservice.track_appservice_user_ips
 
     async def check_auth_blocking(
@@ -84,7 +85,7 @@ async def check_auth_blocking(
         if requester:
             if requester.authenticated_entity.startswith("@"):
                 user_id = requester.authenticated_entity
-            elif self._is_mine_server_name(requester.authenticated_entity):
+            elif self.hs._is_mine_server_name(requester.authenticated_entity):
                 # We never block the server from doing actions on behalf of
                 # users.
                 return

synapse/api/filtering.py

@@ -21,6 +21,7 @@
 #
 #
 import json
+import weakref
 from typing import (
     TYPE_CHECKING,
     Awaitable,
@@ -152,7 +153,7 @@ def matrix_user_id_validator(user_id: object) -> bool:
 
 class Filtering:
     def __init__(self, hs: "HomeServer"):
-        self._hs = hs
+        self._hs = weakref.proxy(hs)
         self.store = hs.get_datastores().main
 
         self.DEFAULT_FILTER_COLLECTION = FilterCollection(hs, {})
@@ -320,7 +321,7 @@ def blocks_all_room_timeline(self) -> bool:
 
 class Filter:
     def __init__(self, hs: "HomeServer", filter_json: JsonMapping):
-        self._hs = hs
+        self._hs = weakref.proxy(hs)
         self._store = hs.get_datastores().main
         self.filter_json = filter_json
 

synapse/app/_base.py

@@ -40,6 +40,7 @@
     Tuple,
     cast,
 )
+import weakref
 
 from cryptography.utils import CryptographyDeprecationWarning
 from typing_extensions import ParamSpec
@@ -551,7 +552,7 @@ def run_sighup(*args: Any, **kwargs: Any) -> None:
 
         signal.signal(signal.SIGHUP, run_sighup)
 
-        register_sighup(refresh_certificate, hs)
+        register_sighup(refresh_certificate, weakref.proxy(hs))
         register_sighup(reload_cache_config, hs.config)
 
     # Apply the cache config.

synapse/app/phone_stats_home.py

@@ -198,9 +198,11 @@ def performance_stats_init() -> None:
     # Rather than update on per session basis, batch up the requests.
     # If you increase the loop period, the accuracy of user_daily_visits
     # table will decrease
-    clock.looping_call(
-        hs.get_datastores().main.generate_user_daily_visits,
-        5 * ONE_MINUTE_SECONDS * MILLISECONDS_PER_SECOND,
+    hs.register_looping_call(
+        clock.looping_call(
+            hs.get_datastores().main.generate_user_daily_visits,
+            5 * MILLISECONDS_PER_SECOND,
+        )
     )
 
     # monthly active user limiting functionality
@@ -237,11 +239,13 @@ async def generate_monthly_active_users() -> None:
 
     if hs.config.metrics.report_stats:
         logger.info("Scheduling stats reporting for 3 hour intervals")
-        clock.looping_call(
-            phone_stats_home,
-            PHONE_HOME_INTERVAL_SECONDS * MILLISECONDS_PER_SECOND,
-            hs,
-            stats,
+        hs.register_looping_call(
+            clock.looping_call(
+                phone_stats_home,
+                PHONE_HOME_INTERVAL_SECONDS * MILLISECONDS_PER_SECOND,
+                hs,
+                stats,
+            )
         )
 
         # We need to defer this init for the cases that we daemonize

synapse/crypto/keyring.py

@@ -21,6 +21,7 @@
 
 import abc
 import logging
+import weakref
 from typing import TYPE_CHECKING, Callable, Dict, Iterable, List, Optional, Tuple
 
 import attr
@@ -175,7 +176,7 @@ def __init__(
             process_batch_callback=self._inner_fetch_key_requests,
         )
 
-        self._is_mine_server_name = hs.is_mine_server_name
+        self.hs = weakref.proxy(hs)
 
         # build a FetchKeyResult for each of our own keys, to shortcircuit the
         # fetcher.
@@ -279,7 +280,7 @@ async def process_request(self, verify_request: VerifyJsonRequest) -> None:
 
         # If we are the originating server, short-circuit the key-fetch for any keys
         # we already have
-        if self._is_mine_server_name(verify_request.server_name):
+        if self.hs._is_mine_server_name(verify_request.server_name):
             for key_id in verify_request.key_ids:
                 if key_id in self._local_verify_keys:
                     found_keys[key_id] = self._local_verify_keys[key_id]

synapse/federation/federation_base.py

@@ -20,6 +20,7 @@
 #
 #
 import logging
+import weakref
 from typing import TYPE_CHECKING, Awaitable, Callable, List, Optional, Sequence
 
 from synapse.api.constants import MAX_DEPTH, EventContentFields, EventTypes, Membership
@@ -56,9 +57,8 @@ def __init__(self, message: str, event_id: str):
 
 class FederationBase:
     def __init__(self, hs: "HomeServer"):
-        self.hs = hs
+        self.hs = weakref.proxy(hs)
 
-        self._is_mine_server_name = hs.is_mine_server_name
         self.keyring = hs.get_keyring()
         self._spam_checker_module_callbacks = hs.get_module_api_callbacks().spam_checker
         self.store = hs.get_datastores().main

synapse/federation/federation_client.py

@@ -926,7 +926,7 @@ async def _try_destination_list(
 
         for destination in destinations:
             # We don't want to ask our own server for information we don't have
-            if self._is_mine_server_name(destination):
+            if self.hs._is_mine_server_name(destination):
                 continue
 
             try:
@@ -1606,7 +1606,7 @@ async def forward_third_party_invite(
         self, destinations: Iterable[str], room_id: str, event_dict: JsonDict
     ) -> None:
         for destination in destinations:
-            if self._is_mine_server_name(destination):
+            if self.hs._is_mine_server_name(destination):
                 continue
 
             try:

synapse/federation/federation_server.py

@@ -980,7 +980,7 @@ async def _on_send_membership_event(
             authorising_server = get_domain_from_id(
                 event.content[EventContentFields.AUTHORISING_USER]
             )
-            if not self._is_mine_server_name(authorising_server):
+            if not self.hs._is_mine_server_name(authorising_server):
                 raise SynapseError(
                     400,
                     f"Cannot authorise membership event for {authorising_server}. We can only authorise requests from our own homeserver",

synapse/federation/send_queue.py

@@ -37,6 +37,7 @@
 """
 
 import logging
+import weakref
 from typing import (
     TYPE_CHECKING,
     Dict,
@@ -74,8 +75,7 @@ def __init__(self, hs: "HomeServer"):
         self.server_name = hs.hostname
         self.clock = hs.get_clock()
         self.notifier = hs.get_notifier()
-        self.is_mine_id = hs.is_mine_id
-        self.is_mine_server_name = hs.is_mine_server_name
+        self.hs = weakref.proxy(hs)
 
         # We may have multiple federation sender instances, so we need to track
         # their positions separately.
@@ -208,7 +208,7 @@ def build_and_send_edu(
         key: Optional[Hashable] = None,
     ) -> None:
         """As per FederationSender"""
-        if self.is_mine_server_name(destination):
+        if self.hs.is_mine_server_name(destination):
             logger.info("Not sending EDU to ourselves")
             return