Summary of fixes:

1. current_context() switched from _thread_local to _current_context_var (ContextVar) — This is the key fix. In asyncio, multiple tasks share a thread but each task gets its own copy of ContextVars. Using _thread_local
  meant all concurrent tasks would corrupt each other's logging contexts. This was the root cause of all the "Background process re-entered without a proc" and "Expected logging context X was lost" errors.
  2. _started_processing() activates logcontext for metrics — RequestMetrics.start() reads current_context() to track resource usage. Previously it ran in sentinel context because the request's logcontext hadn't been
  activated yet.
  3. processing() context manager activates the request's logcontext — Uses PreserveLoggingContext(self.logcontext) so the request handler runs in the correct logging context throughout its lifetime.

synapse/http/aiohttp_shim.py

@@ -904,14 +904,16 @@ class SynapseRequest:
 
         This mirrors ``SynapseRequest.processing()`` from ``site.py``.
         While the context manager is active the request is considered "in
-        progress" and completion logging is deferred until exit.
+        progress", the request's LoggingContext is active, and completion
+        logging is deferred until exit.
         """
         if self._is_processing:
             raise RuntimeError("Request is already processing")
         self._is_processing = True
 
         try:
-            yield
+            with PreserveLoggingContext(self.logcontext):
+                yield
         except Exception:
             logger.exception(
                 "Asynchronous message handler raised an uncaught exception"
@@ -931,9 +933,13 @@ class SynapseRequest:
         """Record that request processing has begun (for metrics/logging)."""
         self.start_time = time.time()
         self.request_metrics = RequestMetrics(our_server_name=self.our_server_name)
-        self.request_metrics.start(
-            self.start_time, name=servlet_name, method=self.get_method()
-        )
+
+        # Temporarily activate the request's logcontext for metrics.start()
+        # which reads current_context() to track resource usage.
+        with PreserveLoggingContext(self.logcontext):
+            self.request_metrics.start(
+                self.start_time, name=servlet_name, method=self.get_method()
+            )
 
         self.synapse_site.access_logger.debug(
             "%s - %s - Received request: %s %s",

synapse/logging/context.py

@@ -746,17 +746,20 @@ class PreserveLoggingContext:
                 )
 
 
-_thread_local = threading.local()
-_thread_local.current_context = SENTINEL_CONTEXT
-
 _current_context_var: contextvars.ContextVar[
     "LoggingContextOrSentinel"
 ] = contextvars.ContextVar("synapse_logging_context", default=SENTINEL_CONTEXT)
 
 
 def current_context() -> LoggingContextOrSentinel:
-    """Get the current logging context."""
-    return getattr(_thread_local, "current_context", SENTINEL_CONTEXT)
+    """Get the current logging context.
+
+    Uses a ContextVar so that each asyncio task has its own isolated context.
+    (The old _thread_local approach was fine for Twisted where only one callback
+    ran at a time, but in asyncio multiple tasks share a thread and would
+    corrupt each other's contexts.)
+    """
+    return _current_context_var.get(SENTINEL_CONTEXT)
 
 
 def set_current_context(context: LoggingContextOrSentinel) -> LoggingContextOrSentinel:
@@ -776,7 +779,6 @@ def set_current_context(context: LoggingContextOrSentinel) -> LoggingContextOrSe
     if current is not context:
         rusage = get_thread_resource_usage()
         current.stop(rusage)
-        _thread_local.current_context = context
         _current_context_var.set(context)
         context.start(rusage)