DEBUG: Logging contexts getting lost

Running a normal server (`daemonize: false`):

```
poetry run synapse_homeserver --config-path homeserver.yaml
```

Bad logs being seen:

```
PreserveLoggingContext: Expected logging context sentinel but found main
```

```
LoggingContext: Expected logging context main was lost
```

Author: MadLittleMods

synapse/app/_base.py

@@ -72,6 +72,7 @@
 from synapse.events.presence_router import load_legacy_presence_router
 from synapse.handlers.auth import load_legacy_password_auth_providers
 from synapse.http.site import SynapseSite
+from synapse.logging.context import PreserveLoggingContext, current_context
 from synapse.logging.opentracing import init_tracer
 from synapse.metrics import install_gc_manager, register_threadpool
 from synapse.metrics.background_process_metrics import run_as_background_process
@@ -182,7 +183,9 @@ def run() -> None:
         if gc_thresholds:
             gc.set_threshold(*gc_thresholds)
         install_gc_manager()
+        logger.info("run-asdf1")
         run_command()
+        logger.info("run-asdf2")
 
     if daemonize:
         assert pid_file is not None
@@ -192,7 +195,9 @@ def run() -> None:
 
         daemonize_process(pid_file, logger)
 
+    logger.info("asdf1")
     run()
+    logger.info("asdf2")
 
 
 def quit_with_error(error_string: str) -> NoReturn:

synapse/app/homeserver.py

@@ -376,18 +376,18 @@ def setup(config_options: List[str]) -> SynapseHomeServer:
     except Exception as e:
         handle_startup_exception(e)
 
-    async def start() -> None:
-        # Load the OIDC provider metadatas, if OIDC is enabled.
-        if hs.config.oidc.oidc_enabled:
-            oidc = hs.get_oidc_handler()
-            # Loading the provider metadata also ensures the provider config is valid.
-            await oidc.load_metadata()
+    # async def start() -> None:
+    #     # Load the OIDC provider metadatas, if OIDC is enabled.
+    #     if hs.config.oidc.oidc_enabled:
+    #         oidc = hs.get_oidc_handler()
+    #         # Loading the provider metadata also ensures the provider config is valid.
+    #         await oidc.load_metadata()
 
-        await _base.start(hs)
+    #     await _base.start(hs)
 
-        hs.get_datastores().main.db_pool.updates.start_doing_background_updates()
+    #     hs.get_datastores().main.db_pool.updates.start_doing_background_updates()
 
-    register_start(start)
+    # register_start(start)
 
     return hs
 

synapse/handlers/presence.py

@@ -529,14 +529,14 @@ def __init__(self, hs: "HomeServer"):
             self.send_stop_syncing, UPDATE_SYNCING_USERS_MS
         )
 
-        hs.get_reactor().addSystemEventTrigger(
-            "before",
-            "shutdown",
-            run_as_background_process,
-            "generic_presence.on_shutdown",
-            self.server_name,
-            self._on_shutdown,
-        )
+        # hs.get_reactor().addSystemEventTrigger(
+        #     "before",
+        #     "shutdown",
+        #     run_as_background_process,
+        #     "generic_presence.on_shutdown",
+        #     self.server_name,
+        #     self._on_shutdown,
+        # )
 
     async def _on_shutdown(self) -> None:
         if self._track_presence:

synapse/logging/context.py

@@ -411,10 +411,13 @@ def __exit__(
         current = set_current_context(self.previous_context)
         if current is not self:
             if current is SENTINEL_CONTEXT:
-                logcontext_error("Expected logging context %s was lost" % (self,))
+                logcontext_error(
+                    "LoggingContext: Expected logging context %s was lost" % (self,)
+                )
             else:
                 logcontext_error(
-                    "Expected logging context %s but found %s" % (self, current)
+                    "LoggingContext: Expected logging context %s but found %s"
+                    % (self, current)
                 )
 
         # the fact that we are here suggests that the caller thinks that everything
@@ -636,23 +639,44 @@ def __init__(
 
     def __enter__(self) -> None:
         self._old_context = set_current_context(self._new_context)
+        import traceback
+
+        logger.info(
+            "asdf PreserveLoggingContext enter new=%s old=%s %s",
+            self._new_context,
+            self._old_context,
+            traceback.format_stack()[-2].replace(
+                "/home/eric/Documents/github/element/synapse/", ""
+            ),
+        )
 
     def __exit__(
         self,
         type: Optional[Type[BaseException]],
         value: Optional[BaseException],
         traceback: Optional[TracebackType],
     ) -> None:
+        import traceback
+
+        logger.info(
+            "asdf PreserveLoggingContext exit new=%s old=%s %s",
+            self._new_context,
+            self._old_context,
+            traceback.format_stack()[-2].replace(
+                "/home/eric/Documents/github/element/synapse/", ""
+            ),
+        )
         context = set_current_context(self._old_context)
 
         if context != self._new_context:
             if not context:
                 logcontext_error(
-                    "Expected logging context %s was lost" % (self._new_context,)
+                    "PreserveLoggingContext: Expected logging context %s was lost"
+                    % (self._new_context,)
                 )
             else:
                 logcontext_error(
-                    "Expected logging context %s but found %s"
+                    "PreserveLoggingContext: Expected logging context %s but found %s"
                     % (
                         self._new_context,
                         context,
@@ -684,6 +708,15 @@ def set_current_context(context: LoggingContextOrSentinel) -> LoggingContextOrSe
 
     current = current_context()
 
+    # import traceback
+
+    logger.info(
+        "asdf set_current_context new=%s old=%s",
+        context,
+        current,
+        # traceback.format_stack(),
+    )
+
     if current is not context:
         rusage = get_thread_resource_usage()
         current.stop(rusage)

synapse/metrics/background_process_metrics.py

@@ -260,6 +260,12 @@ async def run() -> Optional[R]:
         with BackgroundProcessLoggingContext(
             name=desc, server_name=server_name, instance_id=count
         ) as context:
+            logger.info(
+                "asdf run_as_background_process %s %s previous_context=%s",
+                desc,
+                context,
+                context.previous_context,
+            )
             try:
                 if bg_start_span:
                     ctx = start_active_span(

synapse/server.py

@@ -366,8 +366,8 @@ def setup(self) -> None:
         # Register background tasks required by this server. This must be done
         # somewhat manually due to the background tasks not being registered
         # unless handlers are instantiated.
-        if self.config.worker.run_background_tasks:
-            self.setup_background_tasks()
+        # if self.config.worker.run_background_tasks:
+        #     self.setup_background_tasks()
 
     def start_listening(self) -> None:  # noqa: B027 (no-op by design)
         """Start the HTTP, manhole, metrics, etc listeners

synapse/storage/database.py

@@ -52,6 +52,7 @@
 
 from twisted.enterprise import adbapi
 from twisted.internet.interfaces import IReactorCore
+from twisted.internet import defer
 
 from synapse.api.errors import StoreError
 from synapse.config.database import DatabaseConnectionConfig
@@ -642,13 +643,31 @@ def __init__(
 
         # Check ASAP (and then later, every 1s) to see if we have finished
         # background updates of tables that aren't safe to update.
-        self._clock.call_later(
-            0.0,
-            run_as_background_process,
-            "upsert_safety_check",
-            self.server_name,
-            self._check_safe_to_upsert,
-        )
+        # self._clock.call_later(
+        #     0.0,
+        #     run_as_background_process,
+        #     "upsert_safety_check",
+        #     self.server_name,
+        #     self._check_safe_to_upsert,
+        # )
+
+        # self._clock.call_later(
+        #     0.0,
+        #     run_as_background_process,
+        #     "asdf_call_later",
+        #     self.server_name,
+        #     self.asdf,
+        # )
+        run_as_background_process("asdf_call_later", self.server_name, self.asdf)
+        run_as_background_process("qwer_call_later", self.server_name, self.qwer)
+
+    async def asdf(self) -> None:
+        # Await some dummy value
+        await self._clock.sleep(1.0)
+
+    async def qwer(self) -> None:
+        # Await some dummy value
+        await self._clock.sleep(1.0)
 
     def name(self) -> str:
         "Return the name of this database"

synapse/storage/databases/main/cache.py

@@ -26,7 +26,10 @@
 
 from synapse.api.constants import EventTypes
 from synapse.config._base import Config
-from synapse.metrics.background_process_metrics import wrap_as_background_process
+from synapse.metrics.background_process_metrics import (
+    wrap_as_background_process,
+    run_as_background_process,
+)
 from synapse.replication.tcp.streams import BackfillStream, CachesStream
 from synapse.replication.tcp.streams.events import (
     EventsStream,
@@ -128,13 +131,19 @@ def __init__(
         # old rows.
         # This is only applicable when Postgres is in use; this table is unused
         # and not populated at all when SQLite is the active database engine.
-        if hs.config.worker.run_background_tasks and isinstance(
-            self.database_engine, PostgresEngine
-        ):
-            self.hs.get_clock().call_later(
-                CATCH_UP_CLEANUP_INTERVAL_MS / 1000,
-                self._clean_up_cache_invalidation_wrapper,
-            )
+        # if hs.config.worker.run_background_tasks and isinstance(
+        #     self.database_engine, PostgresEngine
+        # ):
+        #     self.hs.get_clock().call_later(
+        #         CATCH_UP_CLEANUP_INTERVAL_MS / 1000,
+        #         self._clean_up_cache_invalidation_wrapper,
+        #     )
+
+    #     run_as_background_process("qwer_call_later", self.server_name, self.qwer)
+
+    # async def qwer(self) -> None:
+    #     # Await some dummy value
+    #     await self._clock.sleep(1.0)
 
     async def get_all_updated_caches(
         self, instance_name: str, last_id: int, current_id: int, limit: int

synapse/storage/databases/main/lock.py

@@ -99,11 +99,11 @@ def __init__(
         # lead to a race, as we may drop the lock while we are still processing.
         # However, a) it should be a small window, b) the lock is best effort
         # anyway and c) we want to really avoid leaking locks when we restart.
-        hs.get_reactor().addSystemEventTrigger(
-            "before",
-            "shutdown",
-            self._on_shutdown,
-        )
+        # hs.get_reactor().addSystemEventTrigger(
+        #     "before",
+        #     "shutdown",
+        #     self._on_shutdown,
+        # )
 
         self._acquiring_locks: Set[Tuple[str, str]] = set()
 

synapse/util/__init__.py

@@ -124,6 +124,7 @@ async def sleep(self, seconds: float) -> None:
         with context.PreserveLoggingContext():
             self._reactor.callLater(seconds, d.callback, seconds)
             await d
+            logger.info("asdf sleep done")
 
     def time(self) -> float:
         """Returns the current system time in seconds since epoch."""
@@ -158,6 +159,8 @@ def looping_call(
             *args: Positional arguments to pass to function.
             **kwargs: Key arguments to pass to function.
         """
+        # TODO: remove
+        return
         return self._looping_call_common(f, msec, False, *args, **kwargs)
 
     def looping_call_now(
@@ -213,12 +216,14 @@ def call_later(
             **kwargs: Key arguments to pass to function.
         """
 
-        def wrapped_callback(*args: Any, **kwargs: Any) -> None:
+        def wrapped_call_later_callback(*args: Any, **kwargs: Any) -> None:
             with context.PreserveLoggingContext():
                 callback(*args, **kwargs)
 
         with context.PreserveLoggingContext():
-            return self._reactor.callLater(delay, wrapped_callback, *args, **kwargs)
+            return self._reactor.callLater(
+                delay, wrapped_call_later_callback, *args, **kwargs
+            )
 
     def cancel_call_later(self, timer: IDelayedCall, ignore_errs: bool = False) -> None:
         try: