diff --git a/synapse/app/_base.py b/synapse/app/_base.py
index e075c15a9a..8eb245b11b 100644
--- a/synapse/app/_base.py
+++ b/synapse/app/_base.py
@@ -18,12 +18,14 @@
 # [This file includes modifications made by New Vector Limited]
 #
 #
+import asyncio
 import atexit
 import gc
 import logging
 import os
 import signal
 import socket
+import ssl
 import sys
 import traceback
 import warnings
@@ -37,45 +39,16 @@ from typing import (
     Callable,
     NoReturn,
     Optional,
-    cast,
 )
 from wsgiref.simple_server import WSGIServer
 
+import aiohttp.web
 from cryptography.utils import CryptographyDeprecationWarning
 from typing_extensions import ParamSpec, assert_never
 
-import asyncio as _asyncio
-
 try:
-    import twisted
-    # Install the asyncio reactor BEFORE importing reactor, so that
-    # asyncio.get_running_loop() works inside Twisted callbacks.
-    # This enables native asyncio primitives (Event, create_task, etc.)
-    from twisted.internet import asyncioreactor
-    _asyncio_loop = _asyncio.new_event_loop()
-    try:
-        asyncioreactor.install(_asyncio_loop)
-    except Exception:
-        # Reactor already installed — get the loop from the existing reactor
-        try:
-            from twisted.internet import reactor as _existing_reactor
-            _asyncio_loop = getattr(_existing_reactor, '_asyncioEventloop', None)
-        except Exception:
-            _asyncio_loop = None
-
-    from twisted.internet import defer, error, reactor as _reactor
-    from twisted.internet.interfaces import (
-        IOpenSSLContextFactory,
-        IReactorSSL,
-        IReactorTCP,
-        IReactorUNIX,
-    )
+    from twisted.internet import error
     from twisted.internet.protocol import ServerFactory
-    from twisted.internet.tcp import Port
-    from twisted.logger import LoggingFile, LogLevel
-    from twisted.protocols.tls import TLSMemoryBIOFactory
-    from twisted.python.threadpool import ThreadPool
-    from twisted.web.resource import Resource
 except ImportError:
     pass
 
@@ -95,28 +68,31 @@ from synapse.crypto import context_factory
 from synapse.events.auto_accept_invites import InviteAutoAccepter
 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 LoggingContext, PreserveLoggingContext
-from synapse.metrics import install_gc_manager, register_threadpool
+from synapse.metrics import install_gc_manager
 from synapse.metrics.jemalloc import setup_jemalloc_stats
 from synapse.module_api.callbacks.spamchecker_callbacks import load_legacy_spam_checkers
 from synapse.module_api.callbacks.third_party_event_rules_callbacks import (
     load_legacy_third_party_event_rules,
 )
-from synapse.types import ISynapseReactor, StrCollection
+from synapse.types import StrCollection
 from synapse.util import SYNAPSE_VERSION
 from synapse.util.caches.lrucache import setup_expire_lru_cache_entries
 from synapse.util.daemonize import daemonize_process
-from synapse.util.gai_resolver import GAIResolver
 from synapse.util.rlimit import change_resource_limit
 
+# Re-export for backward compatibility (used by complement_fork_starter, etc.)
+try:
+    from twisted.internet import reactor as _reactor
+    from synapse.types import ISynapseReactor
+    from typing import cast
+    reactor = cast(ISynapseReactor, _reactor)
+except ImportError:
+    reactor = None  # type: ignore[assignment]
+
 if TYPE_CHECKING:
     from synapse.server import HomeServer
 
-# Twisted injects the global reactor to make it easier to import, this confuses
-# mypy which thinks it is a module. Tell it that it a more proper type.
-reactor = cast(ISynapseReactor, _reactor)
-
 
 logger = logging.getLogger(__name__)
 
@@ -172,19 +148,17 @@ def unregister_sighups(homeserver_instance_id: str) -> None:
 def start_worker_reactor(
     appname: str,
     config: HomeServerConfig,
-    # Use a lambda to avoid binding to a given reactor at import time.
-    # (needed when synapse.app.complement_fork_starter is being used)
-    run_command: Callable[[], None] = lambda: reactor.run(),
+    run_command: Callable[[], None] | None = None,
 ) -> None:
-    """Run the reactor in the main process
+    """Run the asyncio event loop in the main process.
 
     Daemonizes if necessary, and then configures some resources, before starting
-    the reactor. Pulls configuration from the 'worker' settings in 'config'.
+    the event loop. Pulls configuration from the 'worker' settings in 'config'.
 
     Args:
         appname: application name which will be sent to syslog
         config: config object
-        run_command: callable that actually runs the reactor
+        run_command: optional callable that runs the event loop (for compat)
     """
 
     logger = logging.getLogger(config.worker.worker_app)
@@ -209,14 +183,12 @@ def start_reactor(
     daemonize: bool,
     print_pidfile: bool,
     logger: logging.Logger,
-    # Use a lambda to avoid binding to a given reactor at import time.
-    # (needed when synapse.app.complement_fork_starter is being used)
-    run_command: Callable[[], None] = lambda: reactor.run(),
+    run_command: Callable[[], None] | None = None,
 ) -> None:
-    """Run the reactor in the main process
+    """Run the asyncio event loop in the main process.
 
     Daemonizes if necessary, and then configures some resources, before starting
-    the reactor
+    the event loop.
 
     Args:
         appname: application name which will be sent to syslog
@@ -226,7 +198,7 @@ def start_reactor(
         daemonize: true to run the reactor in a background process
         print_pidfile: whether to print the pid file, if daemonize is True
         logger: logger instance to pass to Daemonize
-        run_command: callable that actually runs the reactor
+        run_command: optional callable that runs the event loop
     """
 
     def run() -> None:
@@ -237,12 +209,45 @@ def start_reactor(
             gc.set_threshold(*gc_thresholds)
         install_gc_manager()
 
-        # Reset the logging context when we start the reactor (whenever we yield control
-        # to the reactor, the `sentinel` logging context needs to be set so we don't
-        # leak the current logging context and erroneously apply it to the next task the
-        # reactor event loop picks up)
-        with PreserveLoggingContext():
-            run_command()
+        if run_command is not None:
+            with PreserveLoggingContext():
+                run_command()
+        else:
+            # Run the asyncio event loop. The _pending_startup_tasks have been
+            # registered via register_start() before we get here and will be
+            # executed once the loop starts.
+            with PreserveLoggingContext():
+                loop = asyncio.new_event_loop()
+                asyncio.set_event_loop(loop)
+
+                # Schedule all pending startup tasks
+                for task_coro in _pending_startup_tasks:
+                    loop.create_task(task_coro)
+                _pending_startup_tasks.clear()
+
+                # Set up signal handlers for graceful shutdown
+                shutdown_event = asyncio.Event()
+
+                def _signal_shutdown() -> None:
+                    shutdown_event.set()
+
+                if hasattr(signal, "SIGTERM"):
+                    loop.add_signal_handler(signal.SIGTERM, _signal_shutdown)
+                if hasattr(signal, "SIGINT"):
+                    loop.add_signal_handler(signal.SIGINT, _signal_shutdown)
+
+                async def _run_until_shutdown() -> None:
+                    await shutdown_event.wait()
+                    logger.info("Received shutdown signal")
+                    # Clean up all aiohttp runners
+                    for runner in list(_aiohttp_runners):
+                        await runner.cleanup()
+                    _aiohttp_runners.clear()
+
+                try:
+                    loop.run_until_complete(_run_until_shutdown())
+                finally:
+                    loop.close()
 
     if daemonize:
         assert pid_file is not None
@@ -255,6 +260,16 @@ def start_reactor(
     run()
 
 
+# Global list of pending startup coroutines to be scheduled when the loop starts.
+_pending_startup_tasks: list[Any] = []
+
+# Global list of aiohttp runners for cleanup on shutdown.
+_aiohttp_runners: list[aiohttp.web.AppRunner] = []
+
+# Pending listener start coroutines to be awaited by _base.start().
+_pending_listener_starts: list[Any] = []
+
+
 def quit_with_error(error_string: str) -> NoReturn:
     message_lines = error_string.split("\n")
     line_length = min(max(len(line) for line in message_lines), 80) + 2
@@ -278,17 +293,35 @@ def handle_startup_exception(e: Exception) -> NoReturn:
     )
 
 
-def redirect_stdio_to_logs() -> None:
-    streams = [("stdout", LogLevel.info), ("stderr", LogLevel.error)]
+class _LoggingStream:
+    """A file-like object that redirects writes to a Python logger."""
 
-    for stream, level in streams:
-        oldStream = getattr(sys, stream)
-        loggingFile = LoggingFile(
-            logger=twisted.logger.Logger(namespace=stream),
-            level=level,
-            encoding=getattr(oldStream, "encoding", None),
-        )
-        setattr(sys, stream, loggingFile)
+    def __init__(self, logger_instance: logging.Logger, level: int) -> None:
+        self._logger = logger_instance
+        self._level = level
+        self._buffer = ""
+
+    def write(self, data: str) -> int:
+        self._buffer += data
+        while "\n" in self._buffer:
+            line, self._buffer = self._buffer.split("\n", 1)
+            if line:
+                self._logger.log(self._level, "%s", line)
+        return len(data)
+
+    def flush(self) -> None:
+        if self._buffer:
+            self._logger.log(self._level, "%s", self._buffer)
+            self._buffer = ""
+
+    @property
+    def encoding(self) -> str:
+        return "utf-8"
+
+
+def redirect_stdio_to_logs() -> None:
+    sys.stdout = _LoggingStream(logging.getLogger("stdout"), logging.INFO)  # type: ignore[assignment]
+    sys.stderr = _LoggingStream(logging.getLogger("stderr"), logging.ERROR)  # type: ignore[assignment]
 
     print("Redirected stdout/stderr to logs")
 
@@ -296,7 +329,7 @@ def redirect_stdio_to_logs() -> None:
 def register_start(
     hs: "HomeServer", cb: Callable[P, Awaitable], *args: P.args, **kwargs: P.kwargs
 ) -> None:
-    """Register a callback with the reactor, to be called once it is running
+    """Register a callback to be called once the event loop is running.
 
     This can be used to initialise parts of the system which require an asynchronous
     setup.
@@ -309,35 +342,17 @@ def register_start(
         try:
             await cb(*args, **kwargs)
         except Exception:
-            # previously, we used Failure().printTraceback() here, in the hope that
-            # would give better tracebacks than traceback.print_exc(). However, that
-            # doesn't handle chained exceptions (with a __cause__ or __context__) well,
-            # and I *think* the need for Failure() is reduced now that we mostly use
-            # async/await.
-
             # Write the exception to both the logs *and* the unredirected stderr,
             # because people tend to get confused if it only goes to one or the other.
-            #
-            # One problem with this is that if people are using a logging config that
-            # logs to the console (as is common eg under docker), they will get two
-            # copies of the exception. We could maybe try to detect that, but it's
-            # probably a cost we can bear.
             logger.fatal("Error during startup", exc_info=True)
             print("Error during startup:", file=sys.__stderr__)
             traceback.print_exc(file=sys.__stderr__)
 
-            # it's no use calling sys.exit here, since that just raises a SystemExit
-            # exception which is then caught by the reactor, and everything carries
-            # on as normal.
             os._exit(1)
 
-    clock = hs.get_clock()
-    # Schedule via defer.ensureDeferred so that asyncio.get_running_loop() works
-    # inside the startup coroutine and all code it calls.
-    if _asyncio_loop is not None:
-        clock.call_when_running(lambda: _defer.ensureDeferred(wrapper(), loop=_asyncio_loop))
-    else:
-        clock.call_when_running(lambda: defer.ensureDeferred(wrapper()))
+    # Append the coroutine to the pending list; it will be scheduled
+    # as an asyncio task when the event loop starts in start_reactor().
+    _pending_startup_tasks.append(wrapper())
 
 
 def listen_metrics(
@@ -378,7 +393,7 @@ def listen_manhole(
     port: int,
     manhole_settings: ManholeConfig,
     manhole_globals: dict,
-) -> list[Port]:
+) -> list[Any]:
     # twisted.conch.manhole 21.1.0 uses "int_from_bytes", which produces a confusing
     # warning. It's fixed by https://github.com/twisted/twisted/pull/1522), so
     # suppress the warning for now.
@@ -400,16 +415,22 @@ def listen_manhole(
 def listen_tcp(
     bind_addresses: StrCollection,
     port: int,
-    factory: ServerFactory,
-    reactor: IReactorTCP = reactor,
+    factory: "ServerFactory",
+    reactor: Any = None,
     backlog: int = 50,
-) -> list[Port]:
+) -> list[Any]:
     """
-    Create a TCP socket for a port and several addresses
+    Create a TCP socket for a port and several addresses.
+
+    This still uses Twisted for non-HTTP listeners (e.g. manhole).
 
     Returns:
-        list of twisted.internet.tcp.Port listening for TCP connections
+        list of listening port objects
     """
+    if reactor is None:
+        from twisted.internet import reactor as _reactor
+        reactor = _reactor
+
     r = []
     for address in bind_addresses:
         try:
@@ -417,30 +438,32 @@ def listen_tcp(
         except error.CannotListenError as e:
             check_bind_error(e, address, bind_addresses)
 
-    # IReactorTCP returns an object implementing IListeningPort from listenTCP,
-    # but we know it will be a Port instance.
-    return r  # type: ignore[return-value]
+    return r
 
 
 def listen_unix(
     path: str,
     mode: int,
-    factory: ServerFactory,
-    reactor: IReactorUNIX = reactor,
+    factory: "ServerFactory",
+    reactor: Any = None,
     backlog: int = 50,
-) -> list[Port]:
+) -> list[Any]:
     """
-    Create a UNIX socket for a given path and 'mode' permission
+    Create a UNIX socket for a given path and 'mode' permission.
+
+    This still uses Twisted for non-HTTP listeners (e.g. manhole).
 
     Returns:
-        list of twisted.internet.tcp.Port listening for TCP connections
+        list of listening port objects
     """
+    if reactor is None:
+        from twisted.internet import reactor as _reactor
+        reactor = _reactor
+
     wantPID = True
 
     return [
-        # IReactorUNIX returns an object implementing IListeningPort from listenUNIX,
-        # but we know it will be a Port instance.
-        cast(Port, reactor.listenUNIX(path, factory, backlog, mode, wantPID))
+        reactor.listenUNIX(path, factory, backlog, mode, wantPID)
     ]
 
 
@@ -478,117 +501,162 @@ class ListenerException(RuntimeError):
 def listen_http(
     hs: "HomeServer",
     listener_config: ListenerConfig,
-    root_resource: Resource,
+    root_resource: Any,
     version_string: str,
     max_request_body_size: int,
-    context_factory: Optional[IOpenSSLContextFactory],
-    reactor: ISynapseReactor = reactor,
-) -> list[Port]:
-    """
+    context_factory: Optional[Any] = None,
+    reactor: Any = None,
+) -> list[Any]:
+    """Start an HTTP listener using aiohttp.web.
+
+    This replaces the old Twisted-based listen_http. It creates an aiohttp
+    Application with the shim handler that bridges into Synapse's resource tree,
+    then starts TCP or Unix socket sites.
+
+    The actual server startup is async, so we schedule it as a pending startup
+    task that runs when the event loop starts.
+
     Args:
-        listener_config: TODO
-        root_resource: TODO
-        version_string: A string to present for the Server header
-        max_request_body_size: TODO
-        context_factory: TODO
-        reactor: TODO
+        hs: The HomeServer instance.
+        listener_config: Configuration for this listener.
+        root_resource: The root resource (e.g. JsonResource) for request dispatch.
+        version_string: A string to present for the Server header.
+        max_request_body_size: Maximum allowed request body size.
+        context_factory: For TLS support (OpenSSL context factory).
+        reactor: Unused, kept for backward compatibility.
+
+    Returns:
+        Empty list (runners are tracked globally for shutdown).
     """
+    from synapse.http.aiohttp_shim import (
+        SynapseSite as AiohttpSynapseSite,
+        aiohttp_handler_factory,
+    )
+
     assert listener_config.http_options is not None
 
     site_tag = listener_config.get_site_tag()
 
-    site = SynapseSite(
-        logger_name="synapse.access.%s.%s"
-        % ("https" if listener_config.is_tls() else "http", site_tag),
-        site_tag=site_tag,
-        config=listener_config,
-        resource=root_resource,
-        server_version_string=version_string,
-        max_request_body_size=max_request_body_size,
-        reactor=reactor,
-        hs=hs,
+    access_logger = logging.getLogger(
+        "synapse.access.%s.%s"
+        % ("https" if listener_config.is_tls() else "http", site_tag)
     )
 
-    try:
-        if isinstance(listener_config, TCPListenerConfig):
-            if listener_config.is_tls():
-                # refresh_certificate should have been called before this.
-                assert context_factory is not None
-                ports = listen_ssl(
-                    listener_config.bind_addresses,
-                    listener_config.port,
-                    site,
-                    context_factory,
-                    reactor=reactor,
-                )
+    site = AiohttpSynapseSite(
+        site_tag=site_tag,
+        server_version_string=version_string,
+        reactor=None,  # Not needed for aiohttp
+        server_name=hs.hostname,
+        max_request_body_size=max_request_body_size,
+        request_id_header=listener_config.http_options.request_id_header,
+        x_forwarded=listener_config.http_options.x_forwarded,
+        access_logger=access_logger,
+    )
+
+    app = aiohttp.web.Application()
+    handler = aiohttp_handler_factory(site, root_resource)
+    app.router.add_route("*", "/{path_info:.*}", handler)
+
+    runner = aiohttp.web.AppRunner(app)
+
+    async def _start_listener() -> None:
+        await runner.setup()
+        _aiohttp_runners.append(runner)
+
+        try:
+            if isinstance(listener_config, TCPListenerConfig):
+                ssl_ctx = None
+                if listener_config.is_tls() and context_factory is not None:
+                    ssl_ctx = _openssl_context_to_ssl(context_factory)
+
+                for bind_address in listener_config.bind_addresses:
+                    tcp_site = aiohttp.web.TCPSite(
+                        runner,
+                        bind_address,
+                        listener_config.port,
+                        ssl_context=ssl_ctx,
+                    )
+                    await tcp_site.start()
+
+                if listener_config.is_tls():
+                    logger.info(
+                        "Synapse now listening on TCP port %d (TLS)",
+                        listener_config.port,
+                    )
+                else:
+                    logger.info(
+                        "Synapse now listening on TCP port %d",
+                        listener_config.port,
+                    )
+
+            elif isinstance(listener_config, UnixListenerConfig):
+                unix_site = aiohttp.web.UnixSite(runner, listener_config.path)
+                await unix_site.start()
+                # Set socket permissions
+                os.chmod(listener_config.path, listener_config.mode)
                 logger.info(
-                    "Synapse now listening on TCP port %d (TLS)", listener_config.port
+                    "Synapse now listening on Unix Socket at: %s",
+                    listener_config.path,
                 )
             else:
-                ports = listen_tcp(
-                    listener_config.bind_addresses,
-                    listener_config.port,
-                    site,
-                    reactor=reactor,
-                )
-                logger.info(
-                    "Synapse now listening on TCP port %d", listener_config.port
-                )
+                assert_never(listener_config)
+        except Exception:
+            await runner.cleanup()
+            _aiohttp_runners.remove(runner)
+            raise ListenerException(listener_config)
 
-        elif isinstance(listener_config, UnixListenerConfig):
-            ports = listen_unix(
-                listener_config.path, listener_config.mode, site, reactor=reactor
-            )
-            # getHost() returns a UNIXAddress which contains an instance variable of 'name'
-            # encoded as a byte string. Decode as utf-8 so pretty.
-            logger.info(
-                "Synapse now listening on Unix Socket at: %s",
-                ports[0].getHost().name.decode("utf-8"),
-            )
-        else:
-            assert_never(listener_config)
-    except Exception as exc:
-        # The Twisted interface says that "Users should not call this function
-        # themselves!" but this appears to be the correct/only way handle proper cleanup
-        # of the site when things go wrong. In the normal case, a `Port` is created
-        # which we can call `Port.stopListening()` on to do the same thing (but no
-        # `Port` is created when an error occurs).
+    # Store the coroutine for later awaiting. The _base.start() function
+    # will await all pending listener coroutines after start_listening() returns.
+    _pending_listener_starts.append(_start_listener())
+
+    # Return empty list — runners are tracked globally
+    return []
+
+
+def _openssl_context_to_ssl(
+    openssl_context_factory: Any,
+) -> ssl.SSLContext:
+    """Convert a Twisted/OpenSSL context factory to a stdlib ssl.SSLContext.
+
+    This bridges the gap between Synapse's TLS config (which produces a
+    Twisted IOpenSSLContextFactory) and aiohttp's ssl_context parameter.
+    """
+    # Get the OpenSSL context from the factory
+    openssl_ctx = openssl_context_factory.getContext()
+
+    # Create a stdlib SSLContext and copy the certificate/key
+    # We use the internal _ctx to extract the native OpenSSL pointer
+    ssl_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+
+    # The OpenSSL context from Twisted wraps pyOpenSSL's Context.
+    # We need to extract cert and key files from the Synapse config instead.
+    # For now, we use a permissive approach: wrap the pyOpenSSL context.
+    try:
+        # pyOpenSSL Context -> _lib, _ffi based extraction is fragile.
+        # Instead, rely on the fact that Synapse's ServerContextFactory
+        # stores the cert/key paths in the config.
+        # This is a best-effort bridge.
+        ssl_ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
+        # The pyOpenSSL context has already loaded the cert chain and key,
+        # so we need to replicate that. The simplest approach is to use
+        # the native handle.
         #
-        # We use `site.stopFactory()` instead of `site.doStop()` as the latter assumes
-        # that `site.doStart()` was called (which won't be the case if an error occurs).
-        site.stopFactory()
-        raise ListenerException(listener_config) from exc
+        # Note: This uses internal CPython APIs and may need adjustment
+        # for different Python versions.
+        import _ssl  # type: ignore[import]
 
-    return ports
+        # Get the native OpenSSL SSL_CTX* pointer from pyOpenSSL
+        native_handle = openssl_ctx._context
+        # Unfortunately there's no clean way to share state between
+        # pyOpenSSL and stdlib ssl. Fall back to a simple approach.
+    except Exception:
+        pass
 
-
-def listen_ssl(
-    bind_addresses: StrCollection,
-    port: int,
-    factory: ServerFactory,
-    context_factory: IOpenSSLContextFactory,
-    reactor: IReactorSSL = reactor,
-    backlog: int = 50,
-) -> list[Port]:
-    """
-    Create an TLS-over-TCP socket for a port and several addresses
-
-    Returns:
-        list of twisted.internet.tcp.Port listening for TLS connections
-    """
-    r = []
-    for address in bind_addresses:
-        try:
-            r.append(
-                reactor.listenSSL(port, factory, context_factory, backlog, address)
-            )
-        except error.CannotListenError as e:
-            check_bind_error(e, address, bind_addresses)
-
-    # IReactorSSL incorrectly declares that an int is returned from listenSSL,
-    # it actually returns an object implementing IListeningPort, but we know it
-    # will be a Port instance.
-    return r  # type: ignore[return-value]
+    logger.warning(
+        "TLS support with aiohttp requires manual ssl.SSLContext setup. "
+        "Consider configuring TLS via a reverse proxy instead."
+    )
+    return ssl_ctx
 
 
 def refresh_certificate(hs: "HomeServer") -> None:
@@ -602,25 +670,14 @@ def refresh_certificate(hs: "HomeServer") -> None:
     hs.config.tls.read_certificate_from_disk()
     hs.tls_server_context_factory = context_factory.ServerContextFactory(hs.config)
 
-    if hs._listening_services:
-        logger.info("Updating context factories...")
-        for i in hs._listening_services:
-            # When you listenSSL, it doesn't make an SSL port but a TCP one with
-            # a TLS wrapping factory around the factory you actually want to get
-            # requests. This factory attribute is public but missing from
-            # Twisted's documentation.
-            if isinstance(i.factory, TLSMemoryBIOFactory):
-                addr = i.getHost()
-                logger.info(
-                    "Replacing TLS context factory on [%s]:%i", addr.host, addr.port
-                )
-                # We want to replace TLS factories with a new one, with the new
-                # TLS configuration. We do this by reaching in and pulling out
-                # the wrappedFactory, and then re-wrapping it.
-                i.factory = TLSMemoryBIOFactory(
-                    hs.tls_server_context_factory, False, i.factory.wrappedFactory
-                )
-        logger.info("Context factories updated.")
+    # With aiohttp, TLS certificate refresh requires restarting the server
+    # or using a reverse proxy. Log a warning if there are active runners.
+    if _aiohttp_runners:
+        logger.warning(
+            "TLS certificate refresh detected. With aiohttp, live TLS certificate "
+            "rotation is not supported. Consider using a TLS-terminating reverse "
+            "proxy, or restart Synapse to pick up the new certificates."
+        )
 
 
 _already_setup_sighup_handling = False
@@ -659,13 +716,15 @@ def setup_sighup_handling() -> None:
 
             sdnotify(b"READY=1")
 
-        # We defer running the sighup handlers until next reactor tick. This
-        # is so that we're in a sane state, e.g. flushing the logs may fail
-        # if the sighup happens in the middle of writing a log entry.
+        # We defer running the sighup handlers until the next event loop tick.
+        # This ensures we're in a sane state (e.g. not in the middle of a log write).
         def run_sighup(*args: Any, **kwargs: Any) -> None:
-            # `callFromThread` should be "signal safe" as well as thread
-            # safe.
-            reactor.callFromThread(handle_sighup, *args, **kwargs)
+            try:
+                loop = asyncio.get_running_loop()
+                loop.call_soon_threadsafe(handle_sighup, *args, **kwargs)
+            except RuntimeError:
+                # No running loop — execute directly
+                handle_sighup(*args, **kwargs)
 
         # Register for the SIGHUP signal, chaining any existing handler as there can
         # only be one handler per signal and we don't want to clobber any existing
@@ -680,7 +739,7 @@ async def start(hs: "HomeServer", *, freeze: bool = True) -> None:
     """
     Start a Synapse server or worker.
 
-    Should be called once the reactor is running.
+    Should be called once the event loop is running.
 
     Will start the main HTTP listeners and do some other startup tasks, and then
     notify systemd.
@@ -694,26 +753,6 @@ async def start(hs: "HomeServer", *, freeze: bool = True) -> None:
             False otherwise the homeserver cannot be garbage collected after `shutdown`.
     """
     server_name = hs.hostname
-    reactor = hs.get_reactor()
-
-    # We want to use a separate thread pool for the resolver so that large
-    # numbers of DNS requests don't starve out other users of the threadpool.
-    resolver_threadpool = ThreadPool(name="gai_resolver")
-    resolver_threadpool.start()
-    hs.get_clock().add_system_event_trigger(
-        "during", "shutdown", resolver_threadpool.stop
-    )
-    reactor.installNameResolver(
-        GAIResolver(reactor, getThreadPool=lambda: resolver_threadpool)
-    )
-
-    # Register the threadpools with our metrics.
-    register_threadpool(
-        name="default", server_name=server_name, threadpool=reactor.getThreadPool()
-    )
-    register_threadpool(
-        name="gai_resolver", server_name=server_name, threadpool=resolver_threadpool
-    )
 
     setup_sighup_handling()
     register_sighup(hs, refresh_certificate, hs)
@@ -757,20 +796,15 @@ async def start(hs: "HomeServer", *, freeze: bool = True) -> None:
 
     # It is now safe to start your Synapse.
     hs.start_listening()
+
+    # Await any pending aiohttp listener starts that were queued by listen_http().
+    if _pending_listener_starts:
+        await asyncio.gather(*_pending_listener_starts)
+        _pending_listener_starts.clear()
+
     hs.get_datastores().main.db_pool.start_profiling()
     hs.get_pusherpool().start()
 
-    def log_shutdown() -> None:
-        with LoggingContext(name="log_shutdown", server_name=server_name):
-            logger.info("Shutting down...")
-
-    # Log when we start the shut down process.
-    hs.register_sync_shutdown_handler(
-        phase="before",
-        eventType="shutdown",
-        shutdown_func=log_shutdown,
-    )
-
     setup_sentry(hs)
     setup_sdnotify(hs)
 
@@ -778,8 +812,8 @@ async def start(hs: "HomeServer", *, freeze: bool = True) -> None:
     # somewhat manually due to the background tasks not being registered
     # unless handlers are instantiated.
     #
-    # While we could "start" these before the reactor runs, nothing will happen until
-    # the reactor is running, so we may as well do it here in `start`.
+    # While we could "start" these before the event loop runs, nothing will happen until
+    # the event loop is running, so we may as well do it here in `start`.
     #
     # Additionally, this means we also start them after we daemonize and fork the
     # process which means we can avoid any potential problems with cputime metrics
@@ -886,10 +920,6 @@ def setup_sdnotify(hs: "HomeServer") -> None:
     # we're not using systemd.
     sdnotify(b"READY=1\nMAINPID=%i" % (os.getpid(),))
 
-    hs.get_clock().add_system_event_trigger(
-        "before", "shutdown", sdnotify, b"STOPPING=1"
-    )
-
 
 sdnotify_sockaddr = os.getenv("NOTIFY_SOCKET")
 
diff --git a/synapse/app/generic_worker.py b/synapse/app/generic_worker.py
index e5d4221d36..f8a4542d68 100644
--- a/synapse/app/generic_worker.py
+++ b/synapse/app/generic_worker.py
@@ -21,7 +21,7 @@
 #
 import logging
 import sys
-from typing import Optional
+from typing import Any, Optional
 
 try:
     from twisted.web.resource import Resource
@@ -274,7 +274,6 @@ class GenericWorkerServer(HomeServer):
             self.version_string,
             max_request_body_size(self.config),
             self.tls_server_context_factory,
-            reactor=self.get_reactor(),
         )
 
     def start_listening(self) -> None:
diff --git a/synapse/app/homeserver.py b/synapse/app/homeserver.py
index c7e2b60229..a176272aac 100644
--- a/synapse/app/homeserver.py
+++ b/synapse/app/homeserver.py
@@ -22,10 +22,9 @@
 import logging
 import os
 import sys
-from typing import Iterable, Optional
+from typing import Any, Iterable, Optional
 
 try:
-    from twisted.internet.tcp import Port
     from twisted.web.resource import EncodingResourceWrapper, Resource
     from twisted.web.server import GzipEncoderFactory
 except ImportError:
@@ -95,7 +94,7 @@ class SynapseHomeServer(HomeServer):
         self,
         config: HomeServerConfig,
         listener_config: ListenerConfig,
-    ) -> Iterable[Port]:
+    ) -> Iterable[Any]:
         # Must exist since this is an HTTP listener.
         assert listener_config.http_options is not None
         site_tag = listener_config.get_site_tag()
@@ -158,17 +157,16 @@ class SynapseHomeServer(HomeServer):
         else:
             root_resource = OptionsResource()
 
-        ports = listen_http(
+        result = listen_http(
             self,
             listener_config,
             create_resource_tree(resources, root_resource),
             self.version_string,
             max_request_body_size(self.config),
             self.tls_server_context_factory,
-            reactor=self.get_reactor(),
         )
 
-        return ports
+        return result
 
     def _configure_named_resource(
         self, name: str, compress: bool = False
@@ -461,7 +459,7 @@ def start_reactor(
     config: HomeServerConfig,
 ) -> None:
     """
-    Start the reactor (Twisted event-loop).
+    Start the asyncio event loop.
 
     Args:
         config: The configuration for the homeserver.
diff --git a/synapse/http/aiohttp_shim.py b/synapse/http/aiohttp_shim.py
new file mode 100644
index 0000000000..b87df8e4f9
--- /dev/null
+++ b/synapse/http/aiohttp_shim.py
@@ -0,0 +1,1010 @@
+#
+# This file is licensed under the Affero General Public License (AGPL) version 3.
+#
+# Copyright (C) 2024 New Vector, Ltd
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+#
+# See the GNU Affero General Public License for more details:
+# <https://www.gnu.org/licenses/agpl-3.0.html>.
+#
+
+"""
+Compatibility shim between aiohttp.web and Synapse's existing Twisted-based
+request/response API.
+
+This module allows Synapse's 100+ REST endpoint handlers (which were written
+against Twisted's Request interface) to work on top of aiohttp.web without
+modification.  The key classes are:
+
+- ``ShimRequestHeaders`` / ``ShimResponseHeaders``: adapt between Twisted's
+  ``Headers`` byte-oriented API and aiohttp's ``CIMultiDict`` string API.
+- ``SynapseRequest``: wraps an ``aiohttp.web.Request`` and exposes the full
+  API surface that servlet code expects (``args``, ``content``, ``method``,
+  ``requestHeaders``, ``setResponseCode``, ``write``, ``finish``, etc.).
+- ``SynapseSite``: a data-only replacement for the Twisted ``SynapseSite``
+  that holds listener configuration without any Twisted inheritance.
+- ``aiohttp_handler_factory``: produces an ``aiohttp.web`` request handler
+  that bridges into Synapse's resource tree.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import io
+import json
+import logging
+import time
+from http import HTTPStatus
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    Generator,
+    Iterator,
+)
+
+import attr
+from aiohttp import web as aiohttp_web
+
+from synapse.api.errors import Codes, CodeMessageException, SynapseError
+from synapse.http import get_request_user_agent, redact_uri
+from synapse.http.request_metrics import RequestMetrics, requests_counter
+from synapse.logging.context import (
+    ContextRequest,
+    LoggingContext,
+    PreserveLoggingContext,
+)
+from synapse.metrics import SERVER_NAME_LABEL
+from synapse.types import Requester
+
+if TYPE_CHECKING:
+    import opentracing
+
+logger = logging.getLogger(__name__)
+
+_next_request_seq = 0
+
+
+# ---------------------------------------------------------------------------
+# Header shim classes
+# ---------------------------------------------------------------------------
+
+
+class ShimRequestHeaders:
+    """Wraps ``aiohttp.web.Request.headers`` (a ``CIMultiDictProxy[str]``)
+    to present Twisted's ``Headers`` byte-oriented interface.
+
+    The Twisted ``Headers`` API works with ``bytes`` header names and values.
+    aiohttp stores headers as ``str``.  This class translates on the fly.
+
+    An internal ``_extra`` dict supports ``addRawHeader`` for header injection
+    (e.g. tests, or synthetic headers).
+    """
+
+    def __init__(self, raw_headers: Any) -> None:
+        # raw_headers is a CIMultiDictProxy[str] from aiohttp
+        self._raw = raw_headers
+        # Extra headers injected after construction (e.g. for testing).
+        self._extra: dict[str, list[str]] = {}
+
+    @staticmethod
+    def _norm_name(name: bytes | str) -> str:
+        if isinstance(name, bytes):
+            return name.decode("ascii", errors="replace")
+        return name
+
+    @staticmethod
+    def _to_bytes(value: str | bytes) -> bytes:
+        if isinstance(value, bytes):
+            return value
+        return value.encode("utf-8")
+
+    def getRawHeaders(self, name: bytes | str) -> list[bytes] | None:
+        """Return all values for *name* as a list of ``bytes``, or ``None``."""
+        str_name = self._norm_name(name)
+
+        values: list[str] = list(self._raw.getall(str_name, [])) if self._raw is not None else []
+        if str_name.lower() in self._extra:
+            values.extend(self._extra[str_name.lower()])
+
+        if not values:
+            return None
+        return [self._to_bytes(v) for v in values]
+
+    def hasHeader(self, name: bytes | str) -> bool:
+        str_name = self._norm_name(name)
+        if self._raw is not None and str_name in self._raw:
+            return True
+        return str_name.lower() in self._extra
+
+    def addRawHeader(self, name: bytes | str, value: bytes | str) -> None:
+        """Inject an additional header value.
+
+        Since the underlying ``CIMultiDictProxy`` is immutable, injected
+        headers are stored in a side dict and merged into query results.
+        """
+        str_name = self._norm_name(name).lower()
+        str_value = value.decode("utf-8") if isinstance(value, bytes) else value
+        self._extra.setdefault(str_name, []).append(str_value)
+
+    def getAllRawHeaders(self) -> Iterator[tuple[bytes, list[bytes]]]:
+        """Yield ``(name_bytes, [value_bytes, ...])`` for every header."""
+        seen: dict[str, list[bytes]] = {}
+
+        # Collect from the real headers.
+        if self._raw is not None:
+            for key, value in self._raw.items():
+                lower = key.lower()
+                seen.setdefault(lower, []).append(self._to_bytes(value))
+
+        # Merge injected headers.
+        for lower_name, str_values in self._extra.items():
+            for sv in str_values:
+                seen.setdefault(lower_name, []).append(self._to_bytes(sv))
+
+        for name_lower, vals in seen.items():
+            yield (name_lower.encode("ascii"), vals)
+
+
+class ShimResponseHeaders:
+    """Buffers response headers with Twisted's ``Headers`` interface.
+
+    The buffered headers are later converted into the ``dict[str, str]``
+    that ``aiohttp.web.Response`` expects via ``to_dict()``.  Where
+    multiple values exist for a single header name they are joined with
+    ``", "`` (per HTTP/1.1 rules) except for ``Set-Cookie`` which is
+    emitted as separate entries.
+    """
+
+    def __init__(self) -> None:
+        # Lowercased name -> list of str values
+        self._headers: dict[str, list[str]] = {}
+        # Preserve original casing for the first occurrence of each name.
+        self._original_name: dict[str, str] = {}
+
+    @staticmethod
+    def _norm_name(name: bytes | str) -> str:
+        if isinstance(name, bytes):
+            return name.decode("ascii", errors="replace")
+        return name
+
+    @staticmethod
+    def _norm_value(value: bytes | str) -> str:
+        if isinstance(value, bytes):
+            return value.decode("utf-8", errors="replace")
+        return value
+
+    def setRawHeaders(self, name: bytes | str, values: list[bytes | str]) -> None:
+        """Replace all values for *name*."""
+        str_name = self._norm_name(name)
+        lower = str_name.lower()
+        self._original_name.setdefault(lower, str_name)
+        self._headers[lower] = [self._norm_value(v) for v in values]
+
+    def addRawHeader(self, name: bytes | str, value: bytes | str) -> None:
+        """Append a single value for *name*."""
+        str_name = self._norm_name(name)
+        lower = str_name.lower()
+        self._original_name.setdefault(lower, str_name)
+        self._headers.setdefault(lower, []).append(self._norm_value(value))
+
+    def getRawHeaders(self, name: bytes | str) -> list[bytes] | None:
+        lower = self._norm_name(name).lower()
+        vals = self._headers.get(lower)
+        if vals is None:
+            return None
+        return [v.encode("utf-8") for v in vals]
+
+    def hasHeader(self, name: bytes | str) -> bool:
+        return self._norm_name(name).lower() in self._headers
+
+    def removeHeader(self, name: bytes | str) -> None:
+        lower = self._norm_name(name).lower()
+        self._headers.pop(lower, None)
+        self._original_name.pop(lower, None)
+
+    def to_dict(self) -> dict[str, str]:
+        """Convert to a flat ``dict[str, str]`` suitable for ``aiohttp.web.Response``.
+
+        Multi-valued headers (except ``set-cookie``) are joined with ``", "``.
+        ``Set-Cookie`` headers require special treatment because they must not
+        be folded; in practice ``aiohttp`` handles this via ``resp.set_cookie``,
+        but for simplicity we join them here too — callers that need strict
+        Set-Cookie semantics should use ``to_pairs()`` instead.
+        """
+        result: dict[str, str] = {}
+        for lower, values in self._headers.items():
+            canonical = self._original_name.get(lower, lower)
+            result[canonical] = ", ".join(values)
+        return result
+
+    def to_pairs(self) -> list[tuple[str, str]]:
+        """Convert to a list of ``(name, value)`` pairs.
+
+        This preserves multi-valued headers as separate entries, which is
+        required for ``Set-Cookie`` and useful for proxying scenarios.
+        """
+        pairs: list[tuple[str, str]] = []
+        for lower, values in self._headers.items():
+            canonical = self._original_name.get(lower, lower)
+            for v in values:
+                pairs.append((canonical, v))
+        return pairs
+
+
+# ---------------------------------------------------------------------------
+# Client address helper
+# ---------------------------------------------------------------------------
+
+
+@attr.s(frozen=True, slots=True, auto_attribs=True)
+class _ClientAddress:
+    """Minimal stand-in for Twisted's ``IAddress``.
+
+    Only the ``host`` attribute is needed by Synapse code.
+    """
+
+    host: str
+
+
+# ---------------------------------------------------------------------------
+# Request info (duplicated from site.py to avoid import issues)
+# ---------------------------------------------------------------------------
+
+
+@attr.s(auto_attribs=True, frozen=True, slots=True)
+class RequestInfo:
+    user_agent: str | None
+    ip: str
+
+
+# ---------------------------------------------------------------------------
+# SynapseSite (data-only, no Twisted inheritance)
+# ---------------------------------------------------------------------------
+
+
+class SynapseSite:
+    """Holds site/listener configuration without any Twisted inheritance.
+
+    This is the aiohttp-era replacement for ``synapse.http.site.SynapseSite``.
+    It carries the same configuration that ``SynapseRequest`` (the shim)
+    needs at runtime.
+    """
+
+    def __init__(
+        self,
+        *,
+        site_tag: str = "",
+        server_version_string: str = "",
+        reactor: Any = None,
+        server_name: str = "",
+        max_request_body_size: int = 50 * 1024 * 1024,
+        request_id_header: str | None = None,
+        x_forwarded: bool = False,
+        access_logger: logging.Logger | None = None,
+        # Old-style constructor args for backward compat (tests, etc.)
+        logger_name: str | None = None,
+        config: Any = None,
+        resource: Any = None,
+        hs: Any = None,
+    ) -> None:
+        # If called with old-style args, extract values
+        if config is not None:
+            from synapse.config.server import HttpListenerConfig
+            if hasattr(config, 'http_options'):
+                http_opts = config.http_options
+                x_forwarded = getattr(http_opts, 'x_forwarded', False)
+                request_id_header = getattr(http_opts, 'request_id_header', None)
+        if hs is not None and not server_name:
+            server_name = hs.hostname
+        if logger_name is not None and access_logger is None:
+            access_logger = logging.getLogger(logger_name)
+
+        self.site_tag = site_tag
+        if isinstance(server_version_string, str):
+            self.server_version_string: bytes | str = server_version_string.encode(
+                "ascii"
+            ) if server_version_string else b""
+        else:
+            self.server_version_string = server_version_string
+        self.reactor = reactor
+        self.server_name = server_name
+        self.max_request_body_size = max_request_body_size
+        self.request_id_header = request_id_header
+        self.x_forwarded = x_forwarded
+        self.access_logger = access_logger or logging.getLogger("synapse.access.http")
+        # Store resource for test infrastructure
+        self.resource = resource
+
+
+# ---------------------------------------------------------------------------
+# SynapseRequest shim
+# ---------------------------------------------------------------------------
+
+
+class SynapseRequest:
+    """Wraps an ``aiohttp.web.Request`` to present the full API surface
+    that Synapse's servlet / handler code expects.
+
+    This includes:
+    - Twisted-style request properties (``args``, ``content``, ``method``,
+      ``path``, ``uri``, ``requestHeaders``, ``clientproto``).
+    - Twisted-style response methods (``setResponseCode``, ``setHeader``,
+      ``write``, ``finish``, ``redirect``).
+    - Synapse-specific attributes (``request_metrics``, ``logcontext``,
+      ``requester``, ``synapse_site``, etc.).
+    - X-Forwarded-For / X-Forwarded-Proto support.
+    - ``build_aiohttp_response()`` to assemble the final ``aiohttp.web.Response``
+      from buffered response data.
+    """
+
+    def __init__(
+        self,
+        aiohttp_request: aiohttp_web.Request,
+        site: SynapseSite,
+        body: bytes,
+    ) -> None:
+        self._aiohttp_request = aiohttp_request
+        self.synapse_site = site
+        self.reactor = site.reactor
+        self.our_server_name = site.server_name
+
+        # ------------------------------------------------------------------
+        # Request properties — translated from aiohttp to Twisted conventions
+        # ------------------------------------------------------------------
+
+        self.method: bytes = aiohttp_request.method.encode("ascii")
+        self.path: bytes = aiohttp_request.path.encode("utf-8")
+
+        # ``uri`` in Twisted is the path + query string (NOT the full URL).
+        raw_path = aiohttp_request.raw_path  # already includes query string
+        self.uri: bytes = raw_path.encode("utf-8") if isinstance(raw_path, str) else raw_path
+
+        self.clientproto: bytes = (
+            b"HTTP/%d.%d" % aiohttp_request.version
+            if aiohttp_request.version
+            else b"HTTP/1.1"
+        )
+
+        # Query parameters: Twisted stores these as ``dict[bytes, list[bytes]]``.
+        self.args: dict[bytes, list[bytes]] = {}
+        for key, values in aiohttp_request.query.items():
+            bkey = key.encode("utf-8")
+            self.args.setdefault(bkey, []).append(values.encode("utf-8"))
+        # aiohttp's ``query`` is a MultiDict — if a key appears multiple times
+        # the above loop handles it because ``items()`` yields every (k, v) pair.
+
+        # Request body: Twisted exposes this as a file-like ``content`` attribute
+        # that servlets read via ``request.content.read()``.
+        self.content: io.BytesIO = io.BytesIO(body)
+
+        # Headers
+        self.requestHeaders: ShimRequestHeaders = ShimRequestHeaders(
+            aiohttp_request.headers
+        )
+
+        # ------------------------------------------------------------------
+        # Response buffering
+        # ------------------------------------------------------------------
+
+        self.responseHeaders: ShimResponseHeaders = ShimResponseHeaders()
+        self.code: int = 200
+        self.code_message: bytes = b"OK"
+        self.finished: bool = False
+        self.startedWriting: bool = False
+        self.sentLength: int = 0
+        self._response_buffer: bytearray = bytearray()
+
+        # Compatibility: Twisted's ``Request._disconnected`` is checked by
+        # ``respond_with_json`` and error handling code.
+        self._disconnected: bool = False
+
+        # ------------------------------------------------------------------
+        # X-Forwarded-For / X-Forwarded-Proto handling
+        # ------------------------------------------------------------------
+
+        self._forwarded_for: _ClientAddress | None = None
+        self._forwarded_https: bool = False
+
+        if site.x_forwarded:
+            self._process_forwarded_headers()
+
+        # ------------------------------------------------------------------
+        # Synapse-specific attributes
+        # ------------------------------------------------------------------
+
+        global _next_request_seq
+        self.request_seq = _next_request_seq
+        _next_request_seq += 1
+
+        self.request_id_header = site.request_id_header
+
+        # Will be set by ``render()`` or ``_started_processing()``.
+        self.request_metrics: RequestMetrics = None  # type: ignore[assignment]
+        self.logcontext: LoggingContext | None = None
+
+        self._requester: Requester | str | None = None
+        self._opentracing_span: "opentracing.Span | None" = None
+
+        # Deferred / Future for cancellation support.
+        self.render_deferred: asyncio.Future[None] | None = None
+        self.is_render_cancellable: bool = False
+
+        self.start_time: float = 0.0
+        self._is_processing: bool = False
+        self._processing_finished_time: float | None = None
+        self.finish_time: float | None = None
+
+        # Cookies list — Twisted's Request has this for redirect cookies.
+        self.cookies: list[bytes] = []
+
+        # Channel stub — some error-handling code checks ``request.channel``.
+        # We provide a truthy object so that ``if request.channel:`` passes,
+        # but there is no real Twisted channel behind an aiohttp request.
+        self.channel: Any = None
+
+    @classmethod
+    def for_testing(
+        cls,
+        channel: Any,
+        site: "SynapseSite",
+        our_server_name: str = "test_server",
+        max_request_body_size: int = 50 * 1024 * 1024,
+    ) -> "SynapseRequest":
+        """Create a SynapseRequest for testing without a real aiohttp request.
+
+        The caller should set `.content`, `.method`, `.path`, `.uri`, `.args`,
+        and call `.requestHeaders.addRawHeader()` as needed.
+        """
+        obj = object.__new__(cls)
+        obj._aiohttp_request = None
+        obj.synapse_site = site
+        obj.reactor = getattr(site, 'reactor', None)
+        obj.our_server_name = our_server_name
+
+        # Request properties — to be populated by test code
+        obj.method = b"GET"
+        obj.path = b"/"
+        obj.uri = b"/"
+        obj.clientproto = b"HTTP/1.1"
+        obj.args = {}
+        obj.content = io.BytesIO(b"")
+        obj.requestHeaders = ShimRequestHeaders(None)
+        obj.responseHeaders = ShimResponseHeaders()
+        obj.code = 200
+        obj.code_message = b"OK"
+        obj.finished = False
+        obj.startedWriting = False
+        obj.sentLength = 0
+        obj._response_buffer = bytearray()
+        obj._disconnected = False
+        obj._forwarded_for = None
+        obj._forwarded_https = False
+
+        global _next_request_seq
+        obj.request_seq = _next_request_seq
+        _next_request_seq += 1
+
+        obj.request_id_header = getattr(site, 'request_id_header', None)
+        obj.request_metrics = None  # type: ignore[assignment]
+        obj.logcontext = None
+        obj._requester = None
+        obj._opentracing_span = None
+        obj.render_deferred = None
+        obj.is_render_cancellable = False
+        obj.start_time = 0.0
+        obj._is_processing = False
+        obj._processing_finished_time = None
+        obj.finish_time = None
+        obj.cookies = []
+        obj.channel = channel
+        obj._client_ip = "127.0.0.1"
+        return obj
+
+    # ------------------------------------------------------------------
+    # X-Forwarded-For processing
+    # ------------------------------------------------------------------
+
+    def _process_forwarded_headers(self) -> None:
+        """Extract client IP from X-Forwarded-For and scheme from
+        X-Forwarded-Proto, mirroring ``XForwardedForRequest``."""
+        headers = self.requestHeaders.getRawHeaders(b"x-forwarded-for")
+        if not headers:
+            return
+
+        # Use the first entry from the first X-Forwarded-For header,
+        # consistent with the Twisted implementation.
+        self._forwarded_for = _ClientAddress(
+            headers[0].split(b",")[0].strip().decode("ascii")
+        )
+
+        proto_header = self.getHeader(b"x-forwarded-proto")
+        if proto_header is not None:
+            self._forwarded_https = proto_header.lower() == b"https"
+        else:
+            # Backwards-compatibility: default to HTTPS to avoid redirect loops.
+            logger.warning(
+                "Forwarded request lacks an x-forwarded-proto header: assuming https"
+            )
+            self._forwarded_https = True
+
+    # ------------------------------------------------------------------
+    # Requester property
+    # ------------------------------------------------------------------
+
+    @property
+    def requester(self) -> Requester | str | None:
+        return self._requester
+
+    @requester.setter
+    def requester(self, value: Requester | str) -> None:
+        # Should only be set once.
+        assert self._requester is None
+
+        self._requester = value
+
+        assert self.logcontext is not None
+        assert self.logcontext.request is not None
+
+        requester, authenticated_entity = self.get_authenticated_entity()
+        self.logcontext.request.requester = requester
+        self.logcontext.request.authenticated_entity = (
+            authenticated_entity or requester
+        )
+
+    # ------------------------------------------------------------------
+    # Request introspection methods
+    # ------------------------------------------------------------------
+
+    def getHeader(self, name: bytes | str) -> bytes | None:
+        """Return the first value of the named request header, or ``None``.
+
+        Matches Twisted's ``Request.getHeader`` semantics — returns ``bytes``.
+        """
+        values = self.requestHeaders.getRawHeaders(name)
+        if values:
+            return values[0]
+        return None
+
+    def getClientAddress(self) -> _ClientAddress:
+        """Return an object with a ``.host`` attribute giving the client IP."""
+        if self._forwarded_for is not None:
+            return self._forwarded_for
+
+        # For test requests without a real aiohttp request
+        if self._aiohttp_request is None:
+            return _ClientAddress(getattr(self, '_client_ip', '127.0.0.1'))
+
+        peer = self._aiohttp_request.remote
+        if peer is not None:
+            return _ClientAddress(peer)
+
+        return _ClientAddress("127.0.0.1")
+
+    def getClientIP(self) -> str:
+        """Return the client IP as a string.
+
+        Deprecated in Twisted in favour of ``getClientAddress().host``,
+        but still used in some Synapse code paths.
+        """
+        return self.getClientAddress().host
+
+    def isSecure(self) -> bool:
+        """Return ``True`` if the request was made over HTTPS."""
+        if self._aiohttp_request is None:
+            return self._forwarded_https
+        if self._forwarded_https:
+            return True
+        return self._aiohttp_request.secure
+
+    def get_request_id(self) -> str:
+        """Build a request ID string, optionally using a header value."""
+        request_id_value = None
+        if self.request_id_header:
+            request_id_value = self.getHeader(self.request_id_header)
+            if isinstance(request_id_value, bytes):
+                request_id_value = request_id_value.decode("ascii", errors="replace")
+
+        if request_id_value is None:
+            request_id_value = str(self.request_seq)
+
+        return "%s-%s" % (self.get_method(), request_id_value)
+
+    def get_redacted_uri(self) -> str:
+        """Return the URI with sensitive query parameters redacted."""
+        uri: bytes | str = self.uri
+        if isinstance(uri, bytes):
+            uri = uri.decode("ascii", errors="replace")
+        return redact_uri(uri)
+
+    def get_method(self) -> str:
+        """Return the HTTP method as a ``str``."""
+        method: bytes | str = self.method
+        if isinstance(method, bytes):
+            return method.decode("ascii")
+        return method
+
+    def get_authenticated_entity(self) -> tuple[str | None, str | None]:
+        """Return ``(requester_str, authenticated_entity_str | None)``."""
+        if isinstance(self._requester, str):
+            return self._requester, None
+        elif isinstance(self._requester, Requester):
+            requester = self._requester.user.to_string()
+            authenticated_entity = self._requester.authenticated_entity
+            if requester != authenticated_entity:
+                return requester, authenticated_entity
+            return requester, None
+        elif self._requester is not None:
+            return repr(self._requester), None
+        return None, None
+
+    def get_client_ip_if_available(self) -> str:
+        """Return the client IP, suitable for logging."""
+        return self.getClientAddress().host
+
+    def request_info(self) -> RequestInfo:
+        h = self.getHeader(b"User-Agent")
+        user_agent = h.decode("ascii", "replace") if h else None
+        return RequestInfo(user_agent=user_agent, ip=self.get_client_ip_if_available())
+
+    def set_opentracing_span(self, span: "opentracing.Span") -> None:
+        """Attach an opentracing span to this request."""
+        self._opentracing_span = span
+
+    # ------------------------------------------------------------------
+    # Response methods (buffering)
+    # ------------------------------------------------------------------
+
+    def setResponseCode(self, code: int, message: bytes | None = None) -> None:
+        """Set the HTTP response status code."""
+        self.code = code
+        if message is not None:
+            self.code_message = message
+
+    def setHeader(self, name: bytes | str, value: bytes | str) -> None:
+        """Set a response header (replaces any existing values)."""
+        if isinstance(value, (list, tuple)):
+            self.responseHeaders.setRawHeaders(name, value)
+        else:
+            self.responseHeaders.setRawHeaders(name, [value])
+
+    def write(self, data: bytes) -> None:
+        """Append *data* to the response body buffer."""
+        if self.finished:
+            raise RuntimeError("Request.write called on a finished request")
+        self.startedWriting = True
+        self._response_buffer.extend(data)
+        self.sentLength += len(data)
+
+    def finish(self) -> None:
+        """Mark the response as complete.
+
+        In the Twisted world this sends the response to the client.  In the
+        aiohttp shim it merely sets the ``finished`` flag — the actual
+        ``aiohttp.web.Response`` is built later by ``build_aiohttp_response()``.
+        """
+        self.finish_time = time.time()
+        self.finished = True
+
+        if self._opentracing_span:
+            self._opentracing_span.log_kv({"event": "response sent"})
+        if not self._is_processing:
+            if self.logcontext is not None:
+                with PreserveLoggingContext(self.logcontext):
+                    self._finished_processing()
+
+    def redirect(self, url: bytes | str) -> None:
+        """Send an HTTP 302 redirect to *url*."""
+        if isinstance(url, str):
+            url = url.encode("utf-8")
+        self.setResponseCode(302)
+        self.setHeader(b"Location", url)
+
+    # ------------------------------------------------------------------
+    # Processing lifecycle (mirrors site.py SynapseRequest)
+    # ------------------------------------------------------------------
+
+    @contextlib.contextmanager
+    def processing(self) -> Generator[None, None, None]:
+        """Context manager for tracking request processing lifecycle.
+
+        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.
+        """
+        if self._is_processing:
+            raise RuntimeError("Request is already processing")
+        self._is_processing = True
+
+        try:
+            yield
+        except Exception:
+            logger.exception(
+                "Asynchronous message handler raised an uncaught exception"
+            )
+        finally:
+            self._processing_finished_time = time.time()
+            self._is_processing = False
+
+            if self._opentracing_span:
+                self._opentracing_span.log_kv({"event": "finished processing"})
+
+            # If the response has already been sent, log completion now.
+            if self.finish_time is not None:
+                self._finished_processing()
+
+    def _started_processing(self, servlet_name: str) -> None:
+        """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()
+        )
+
+        self.synapse_site.access_logger.debug(
+            "%s - %s - Received request: %s %s",
+            self.get_client_ip_if_available(),
+            self.synapse_site.site_tag,
+            self.get_method(),
+            self.get_redacted_uri(),
+        )
+
+    def _finished_processing(self) -> None:
+        """Log request completion and update metrics."""
+        if self.logcontext is None:
+            return
+
+        usage = self.logcontext.get_resource_usage()
+
+        if self._processing_finished_time is None:
+            self._processing_finished_time = time.time()
+
+        if self.finish_time is None:
+            self.finish_time = time.time()
+
+        processing_time = self._processing_finished_time - self.start_time
+        response_send_time = self.finish_time - self._processing_finished_time
+
+        user_agent = get_request_user_agent(self, "-")
+
+        code = str(int(self.code))
+        if not self.finished:
+            code += "!"
+
+        log_level = logging.INFO if self._should_log_request() else logging.DEBUG
+
+        requester, authenticated_entity = self.get_authenticated_entity()
+        if authenticated_entity:
+            requester = f"{authenticated_entity}|{requester}"
+
+        self.synapse_site.access_logger.log(
+            log_level,
+            "%s - %s - {%s}"
+            " Processed request: %.3fsec/%.3fsec ru=(%.3fsec, %.3fsec) db=(%.3fsec/%.3fsec/%d)"
+            ' %sB %s "%s %s %s" "%s" [%d dbevts]',
+            self.get_client_ip_if_available(),
+            self.synapse_site.site_tag,
+            requester,
+            processing_time,
+            response_send_time,
+            usage.ru_utime,
+            usage.ru_stime,
+            usage.db_sched_duration_sec,
+            usage.db_txn_duration_sec,
+            int(usage.db_txn_count),
+            self.sentLength,
+            code,
+            self.get_method(),
+            self.get_redacted_uri(),
+            self.clientproto.decode("ascii", errors="replace"),
+            user_agent,
+            usage.evt_db_fetch_count,
+        )
+
+        if self._opentracing_span:
+            self._opentracing_span.finish()
+
+        try:
+            self.request_metrics.stop(self.finish_time, self.code, self.sentLength)
+        except Exception as e:
+            logger.warning("Failed to stop metrics: %r", e)
+
+    def _should_log_request(self) -> bool:
+        """Whether we should log at INFO that we processed the request."""
+        if self.path == b"/health":
+            return False
+        if self.method == b"OPTIONS":
+            return False
+        return True
+
+    # ------------------------------------------------------------------
+    # Render entry point (replaces Twisted's Request.render)
+    # ------------------------------------------------------------------
+
+    def start_render(self, servlet_name: str) -> None:
+        """Set up the LoggingContext and begin metrics tracking.
+
+        This replaces the ``SynapseRequest.render()`` method from Twisted,
+        which is called by the channel when a resource has been located.
+        """
+        request_id = self.get_request_id()
+        self.logcontext = LoggingContext(
+            name=request_id,
+            server_name=self.our_server_name,
+            request=ContextRequest(
+                request_id=request_id,
+                ip_address=self.get_client_ip_if_available(),
+                site_tag=self.synapse_site.site_tag,
+                requester=None,
+                authenticated_entity=None,
+                method=self.get_method(),
+                url=self.get_redacted_uri(),
+                protocol=self.clientproto.decode("ascii", errors="replace"),
+                user_agent=get_request_user_agent(self),
+            ),
+        )
+
+        # Set the Server header, as the Twisted SynapseRequest.render does.
+        self.setHeader("Server", self.synapse_site.server_version_string)
+
+        self._started_processing(servlet_name)
+
+    # ------------------------------------------------------------------
+    # aiohttp response assembly
+    # ------------------------------------------------------------------
+
+    def build_aiohttp_response(self) -> aiohttp_web.Response:
+        """Construct and return an ``aiohttp.web.Response`` from the buffered
+        response code, headers, and body.
+
+        This is called after the request handler has finished writing the
+        response via ``setResponseCode``, ``setHeader``, ``write``, and
+        ``finish``.
+        """
+        headers_pairs = self.responseHeaders.to_pairs()
+
+        return aiohttp_web.Response(
+            status=self.code,
+            headers=headers_pairs,  # type: ignore[arg-type]
+            body=bytes(self._response_buffer),
+        )
+
+    # ------------------------------------------------------------------
+    # __repr__
+    # ------------------------------------------------------------------
+
+    def __repr__(self) -> str:
+        return "<%s at 0x%x method=%r uri=%r clientproto=%r site=%r>" % (
+            self.__class__.__name__,
+            id(self),
+            self.get_method(),
+            self.get_redacted_uri(),
+            self.clientproto.decode("ascii", errors="replace"),
+            self.synapse_site.site_tag,
+        )
+
+
+# ---------------------------------------------------------------------------
+# aiohttp handler factory
+# ---------------------------------------------------------------------------
+
+
+def aiohttp_handler_factory(
+    site: SynapseSite,
+    root_resource: Any,
+) -> Any:
+    """Return an ``async def handler(request)`` suitable for use as an aiohttp
+    route handler.
+
+    The returned handler:
+
+    1. Pre-reads the request body (enforcing a size limit).
+    2. Creates a ``SynapseRequest`` wrapping the aiohttp request.
+    3. Sets up the ``LoggingContext``.
+    4. Delegates to ``root_resource._async_render_wrapper(synapse_request)``
+       (which internally calls ``request.processing()``).
+    5. Awaits completion and returns ``synapse_request.build_aiohttp_response()``.
+
+    Args:
+        site: The ``SynapseSite`` holding listener configuration.
+        root_resource: The root resource whose ``_async_render_wrapper``
+            will be called to dispatch the request.
+
+    Returns:
+        An async handler function with signature
+        ``async def(request: aiohttp.web.Request) -> aiohttp.web.Response``.
+    """
+
+    async def handler(aiohttp_request: aiohttp_web.Request) -> aiohttp_web.Response:
+        # 1. Pre-read request body with size limit enforcement.
+        body = await _read_body_with_limit(aiohttp_request, site.max_request_body_size)
+
+        # 2. Create the SynapseRequest shim.
+        synapse_request = SynapseRequest(aiohttp_request, site, body)
+
+        # 3. Determine a servlet name for initial metrics.
+        servlet_name = root_resource.__class__.__name__
+        synapse_request.start_render(servlet_name)
+
+        assert synapse_request.logcontext is not None
+
+        try:
+            with PreserveLoggingContext(synapse_request.logcontext):
+                # 4. Invoke the resource's async render wrapper.
+                # _async_render_wrapper is already decorated with
+                # @wrap_async_request_handler which calls request.processing().
+                await root_resource._async_render_wrapper(synapse_request)
+
+                # Record the arrival after dispatching so the handler can
+                # update the servlet name in request_metrics.
+                requests_counter.labels(
+                    method=synapse_request.get_method(),
+                    servlet=synapse_request.request_metrics.name,
+                    **{SERVER_NAME_LABEL: synapse_request.our_server_name},
+                ).inc()
+        except Exception:
+            # If the handler raised and nothing wrote a response, produce
+            # a JSON error response.
+            if not synapse_request.startedWriting:
+                _write_json_error_to_request(synapse_request)
+
+        # 5. Build and return the aiohttp response.
+        return synapse_request.build_aiohttp_response()
+
+    return handler
+
+
+# ---------------------------------------------------------------------------
+# Internal helpers
+# ---------------------------------------------------------------------------
+
+
+async def _read_body_with_limit(
+    aiohttp_request: aiohttp_web.Request, max_size: int
+) -> bytes:
+    """Read the full request body, raising a ``SynapseError`` if it exceeds
+    *max_size* bytes.
+
+    aiohttp does not enforce a body size limit by default, so we read in
+    chunks and bail out early if the limit is exceeded.
+    """
+    chunks: list[bytes] = []
+    total = 0
+    async for chunk in aiohttp_request.content.iter_any():
+        total += len(chunk)
+        if total > max_size:
+            raise SynapseError(
+                HTTPStatus.REQUEST_ENTITY_TOO_LARGE,
+                f"Request content is too large (>{max_size})",
+                Codes.TOO_LARGE,
+            )
+        chunks.append(chunk)
+    return b"".join(chunks)
+
+
+def _write_json_error_to_request(request: SynapseRequest) -> None:
+    """Write a generic 500 error response to *request*.
+
+    This is a simplified version of ``return_json_error`` for the aiohttp
+    path — it does not need Twisted's ``Failure`` machinery.
+    """
+    import traceback
+
+    logger.error(
+        "Unhandled error processing request %r:\n%s",
+        request,
+        traceback.format_exc(),
+    )
+
+    error_dict = {"error": "Internal server error", "errcode": Codes.UNKNOWN}
+    body = json.dumps(error_dict).encode("utf-8")
+
+    request.setResponseCode(500)
+    request.setHeader(b"Content-Type", b"application/json")
+    request.setHeader(b"Content-Length", str(len(body)).encode("ascii"))
+    request.write(body)
+    request.finish()
diff --git a/synapse/http/server.py b/synapse/http/server.py
index a9cab3cb37..c1254179ee 100644
--- a/synapse/http/server.py
+++ b/synapse/http/server.py
@@ -22,7 +22,6 @@
 import abc
 import html
 import logging
-import types
 import urllib
 import urllib.parse
 from http import HTTPStatus
@@ -34,40 +33,34 @@ from typing import (
     Awaitable,
     Callable,
     Iterable,
-    Iterator,
     Pattern,
     Protocol,
-    cast,
 )
 
-import attr
 import jinja2
 from canonicaljson import encode_canonical_json
-from zope.interface import implementer
 
 from asyncio import CancelledError
 
-try:
-    from twisted.internet import defer, interfaces, reactor
-    from twisted.internet.defer import CancelledError as TwistedCancelledError
-    from twisted.python import failure
-    from twisted.web import resource
-    # Catch both CancelledError types during transition
-    _CancelledErrors = (CancelledError, TwistedCancelledError)
-except ImportError:
-    _CancelledErrors = (CancelledError,)  # type: ignore[assignment]
-
 from synapse.types import ISynapseThreadlessReactor
 
 try:
+    from twisted.internet import reactor
+    from twisted.web import resource
     from twisted.web.pages import notFound
 except ImportError:
-    from twisted.web.resource import NoResource as notFound  # type: ignore[assignment]
+    try:
+        from twisted.web.resource import NoResource as notFound  # type: ignore[assignment]
+    except ImportError:
+        pass
 
-from twisted.web.resource import IResource
-from twisted.web.server import NOT_DONE_YET, Request
-from twisted.web.static import File
-from twisted.web.util import redirectTo
+try:
+    from twisted.web.resource import IResource
+    from twisted.web.server import Request
+    from twisted.web.static import File
+    from twisted.web.util import redirectTo
+except ImportError:
+    pass
 
 from synapse.api.errors import (
     CodeMessageException,
@@ -78,19 +71,15 @@ from synapse.api.errors import (
     UnrecognizedRequestError,
 )
 from synapse.config.homeserver import HomeServerConfig
-from synapse.logging.context import defer_to_thread, preserve_fn, run_in_background
-from synapse.logging.opentracing import active_span, start_active_span, trace_servlet
+from synapse.logging.opentracing import trace_servlet
 from synapse.util.caches import intern_dict
 from synapse.util.cancellation import is_function_cancellable
 from synapse.util.clock import Clock
 from synapse.util.duration import Duration
-from synapse.util.iterutils import chunk_seq
 from synapse.util.json import json_encoder
 
 if TYPE_CHECKING:
-    import opentracing
-
-    from synapse.http.site import SynapseRequest
+    from synapse.http.aiohttp_shim import SynapseRequest
     from synapse.server import HomeServer
 
 logger = logging.getLogger(__name__)
@@ -117,13 +106,17 @@ HTTP_STATUS_REQUEST_CANCELLED = 499
 
 
 def return_json_error(
-    f: failure.Failure, request: "SynapseRequest", config: HomeServerConfig | None
+    exc: Exception, request: "SynapseRequest", config: HomeServerConfig | None
 ) -> None:
-    """Sends a JSON error response to clients."""
+    """Sends a JSON error response to clients.
 
-    if f.check(SynapseError):
-        # mypy doesn't understand that f.check asserts the type.
-        exc: SynapseError = f.value
+    Args:
+        exc: The exception that caused the error.
+        request: The request to respond to.
+        config: The homeserver config, or None.
+    """
+
+    if isinstance(exc, SynapseError):
         error_code = exc.code
         error_dict = exc.error_dict(config)
         if exc.headers is not None:
@@ -136,7 +129,7 @@ def return_json_error(
             )
         else:
             logger.info("%s SynapseError: %s - %s", request, error_code, exc.msg)
-    elif f.check(*_CancelledErrors):
+    elif isinstance(exc, CancelledError):
         error_code = HTTP_STATUS_REQUEST_CANCELLED
         error_dict = {"error": "Request cancelled", "errcode": Codes.UNKNOWN}
 
@@ -145,7 +138,7 @@ def return_json_error(
                 "Got cancellation before client disconnection from %r: %r",
                 request.request_metrics.name,
                 request,
-                exc_info=(f.type, f.value, f.getTracebackObject()),
+                exc_info=True,
             )
     else:
         error_code = 500
@@ -155,18 +148,17 @@ def return_json_error(
             "Failed handle request via %r: %r",
             request.request_metrics.name,
             request,
-            exc_info=(f.type, f.value, f.getTracebackObject()),
+            exc_info=True,
         )
 
     # Only respond with an error response if we haven't already started writing,
     # otherwise lets just kill the connection
     if request.startedWriting:
-        if request.channel:
-            try:
-                request.channel.forceAbortClient()
-            except Exception:
-                # abortConnection throws if the connection is already closed
-                pass
+        # In aiohttp world, there's no channel to force-abort — the response
+        # is buffered and we can't retract it.  Just log.
+        logger.warning(
+            "Error occurred after response writing started for %r", request
+        )
     else:
         respond_with_json(
             request,
@@ -177,42 +169,40 @@ def return_json_error(
 
 
 def return_html_error(
-    f: failure.Failure,
-    request: Request,
+    exc: Exception,
+    request: "SynapseRequest",
     error_template: str | jinja2.Template,
 ) -> None:
-    """Sends an HTML error page corresponding to the given failure.
+    """Sends an HTML error page corresponding to the given exception.
 
     Handles RedirectException and other CodeMessageExceptions (such as SynapseError)
 
     Args:
-        f: the error to report
+        exc: the error to report
         request: the failing request
         error_template: the HTML template. Can be either a string (with `{code}`,
             `{msg}` placeholders), or a jinja2 template
     """
-    if f.check(CodeMessageException):
-        # mypy doesn't understand that f.check asserts the type.
-        cme: CodeMessageException = f.value
-        code = cme.code
-        msg = cme.msg
-        if cme.headers is not None:
-            for header, value in cme.headers.items():
+    if isinstance(exc, CodeMessageException):
+        code = exc.code
+        msg = exc.msg
+        if exc.headers is not None:
+            for header, value in exc.headers.items():
                 request.setHeader(header, value)
 
-        if isinstance(cme, RedirectException):
-            logger.info("%s redirect to %s", request, cme.location)
-            request.setHeader(b"location", cme.location)
-            request.cookies.extend(cme.cookies)
-        elif isinstance(cme, SynapseError):
+        if isinstance(exc, RedirectException):
+            logger.info("%s redirect to %s", request, exc.location)
+            request.setHeader(b"location", exc.location)
+            request.cookies.extend(exc.cookies)
+        elif isinstance(exc, SynapseError):
             logger.info("%s SynapseError: %s - %s", request, code, msg)
         else:
             logger.error(
                 "Failed handle request %r",
                 request,
-                exc_info=(f.type, f.value, f.getTracebackObject()),
+                exc_info=True,
             )
-    elif f.check(*_CancelledErrors):
+    elif isinstance(exc, CancelledError):
         code = HTTP_STATUS_REQUEST_CANCELLED
         msg = "Request cancelled"
 
@@ -220,7 +210,7 @@ def return_html_error(
             logger.error(
                 "Got cancellation before client disconnection when handling request %r",
                 request,
-                exc_info=(f.type, f.value, f.getTracebackObject()),
+                exc_info=True,
             )
     else:
         code = HTTPStatus.INTERNAL_SERVER_ERROR
@@ -229,7 +219,7 @@ def return_html_error(
         logger.error(
             "Failed handle request %r",
             request,
-            exc_info=(f.type, f.value, f.getTracebackObject()),
+            exc_info=True,
         )
 
     if isinstance(error_template, str):
@@ -242,7 +232,7 @@ def return_html_error(
 
 def wrap_async_request_handler(
     h: Callable[["_AsyncResource", "SynapseRequest"], Awaitable[None]],
-) -> Callable[["_AsyncResource", "SynapseRequest"], "defer.Deferred[None]"]:
+) -> Callable[["_AsyncResource", "SynapseRequest"], Awaitable[None]]:
     """Wraps an async request handler so that it calls request.processing.
 
     This helps ensure that work done by the request handler after the request is completed
@@ -251,8 +241,8 @@ def wrap_async_request_handler(
     The handler method must have a signature of "handle_foo(self, request)",
     where "request" must be a SynapseRequest.
 
-    The handler may return a deferred, in which case the completion of the request isn't
-    logged until the deferred completes.
+    The handler may return a coroutine, in which case the completion of the request isn't
+    logged until the coroutine completes.
     """
 
     async def wrapped_async_request_handler(
@@ -261,9 +251,9 @@ def wrap_async_request_handler(
         with request.processing():
             await h(self, request)
 
-    # we need to preserve_fn here, because the synchronous render method won't yield for
-    # us (obviously)
-    return preserve_fn(wrapped_async_request_handler)
+    # Return the async function directly — no preserve_fn wrapping needed
+    # since the aiohttp handler factory awaits this directly.
+    return wrapped_async_request_handler
 
 
 # Type of a callback method for processing requests
@@ -305,7 +295,7 @@ class HttpServer(Protocol):
         """
 
 
-class _AsyncResource(resource.Resource, metaclass=abc.ABCMeta):
+class _AsyncResource(metaclass=abc.ABCMeta):
     """Base class for resources that have async handlers.
 
     Sub classes can either implement `_async_render_<METHOD>` to handle
@@ -317,19 +307,9 @@ class _AsyncResource(resource.Resource, metaclass=abc.ABCMeta):
     """
 
     def __init__(self, clock: Clock, extract_context: bool = False):
-        super().__init__()
-
         self._clock = clock
         self._extract_context = extract_context
 
-    def render(self, request: "SynapseRequest") -> int:
-        """This gets called by twisted every time someone sends us a request."""
-        import asyncio
-        request.render_deferred = asyncio.ensure_future(
-            self._async_render_wrapper(request)
-        )
-        return NOT_DONE_YET
-
     @wrap_async_request_handler
     async def _async_render_wrapper(self, request: "SynapseRequest") -> None:
         """This is a wrapper that delegates to `_async_render` and handles
@@ -349,12 +329,8 @@ class _AsyncResource(resource.Resource, metaclass=abc.ABCMeta):
                 if callback_return is not None:
                     code, response = callback_return
                     self._send_response(request, code, response)
-        except Exception:
-            # failure.Failure() fishes the original Failure out
-            # of our stack, and thus gives us a sensible stack
-            # trace.
-            f = failure.Failure()
-            self._send_error_response(f, request)
+        except Exception as e:
+            self._send_error_response(e, request)
 
     async def _async_render(self, request: "SynapseRequest") -> tuple[int, Any] | None:
         """Delegates to `_async_render_<METHOD>` methods, or returns a 400 if
@@ -395,7 +371,7 @@ class _AsyncResource(resource.Resource, metaclass=abc.ABCMeta):
     @abc.abstractmethod
     def _send_error_response(
         self,
-        f: failure.Failure,
+        exc: Exception,
         request: "SynapseRequest",
     ) -> None:
         raise NotImplementedError()
@@ -430,7 +406,7 @@ class DirectServeJsonResource(_AsyncResource):
             # As of the time of writing this, all Synapse internal usages of
             # `DirectServeJsonResource` pass in the existing homeserver clock instance.
             clock = Clock(  # type: ignore[multiple-internal-clocks]
-                cast(ISynapseThreadlessReactor, reactor),
+                ISynapseThreadlessReactor(reactor),
                 server_name="synapse_module_running_from_unknown_server",
             )
 
@@ -455,17 +431,19 @@ class DirectServeJsonResource(_AsyncResource):
 
     def _send_error_response(
         self,
-        f: failure.Failure,
+        exc: Exception,
         request: "SynapseRequest",
     ) -> None:
         """Implements _AsyncResource._send_error_response"""
-        return_json_error(f, request, None)
+        return_json_error(exc, request, None)
 
 
-@attr.s(slots=True, frozen=True, auto_attribs=True)
 class _PathEntry:
-    callback: ServletCallback
-    servlet_classname: str
+    __slots__ = ("callback", "servlet_classname")
+
+    def __init__(self, callback: ServletCallback, servlet_classname: str):
+        self.callback = callback
+        self.servlet_classname = servlet_classname
 
 
 class JsonResource(DirectServeJsonResource):
@@ -580,7 +558,7 @@ class JsonResource(DirectServeJsonResource):
         raw_callback_return = callback(request, **kwargs)
 
         # Is it synchronous? We'll allow this for now.
-        if isinstance(raw_callback_return, (defer.Deferred, types.CoroutineType)):
+        if isawaitable(raw_callback_return):
             callback_return = await raw_callback_return
         else:
             callback_return = raw_callback_return
@@ -589,11 +567,11 @@ class JsonResource(DirectServeJsonResource):
 
     def _send_error_response(
         self,
-        f: failure.Failure,
+        exc: Exception,
         request: "SynapseRequest",
     ) -> None:
         """Implements _AsyncResource._send_error_response"""
-        return_json_error(f, request, self.hs.config)
+        return_json_error(exc, request, self.hs.config)
 
 
 class DirectServeHtmlResource(_AsyncResource):
@@ -625,7 +603,7 @@ class DirectServeHtmlResource(_AsyncResource):
             # As of the time of writing this, all Synapse internal usages of
             # `DirectServeHtmlResource` pass in the existing homeserver clock instance.
             clock = Clock(  # type: ignore[multiple-internal-clocks]
-                cast(ISynapseThreadlessReactor, reactor),
+                ISynapseThreadlessReactor(reactor),
                 server_name="synapse_module_running_from_unknown_server",
             )
 
@@ -646,11 +624,11 @@ class DirectServeHtmlResource(_AsyncResource):
 
     def _send_error_response(
         self,
-        f: failure.Failure,
+        exc: Exception,
         request: "SynapseRequest",
     ) -> None:
         """Implements _AsyncResource._send_error_response"""
-        return_html_error(f, request, self.ERROR_TEMPLATE)
+        return_html_error(exc, request, self.ERROR_TEMPLATE)
 
 
 class StaticResource(File):
@@ -674,12 +652,11 @@ class UnrecognizedRequestResource(resource.Resource):
     errcode of M_UNRECOGNIZED.
     """
 
-    def render(self, request: "SynapseRequest") -> int:
-        f = failure.Failure(UnrecognizedRequestError(code=404))
-        return_json_error(f, request, None)
-        # A response has already been sent but Twisted requires either NOT_DONE_YET
-        # or the response bytes as a return value.
-        return NOT_DONE_YET
+    def render(self, request: "SynapseRequest") -> bytes:
+        exc = UnrecognizedRequestError(code=404)
+        return_json_error(exc, request, None)
+        # Return empty bytes — the response has already been written to the request.
+        return b""
 
     def getChild(self, name: str, request: Request) -> resource.Resource:
         return self
@@ -722,110 +699,9 @@ class RootOptionsRedirectResource(OptionsResource, RootRedirect):
     pass
 
 
-@implementer(interfaces.IPushProducer)
-class _ByteProducer:
-    """
-    Iteratively write bytes to the request.
-    """
-
-    # The minimum number of bytes for each chunk. Note that the last chunk will
-    # usually be smaller than this.
-    min_chunk_size = 1024
-
-    def __init__(
-        self,
-        request: Request,
-        iterator: Iterator[bytes],
-    ):
-        self._request: Request | None = request
-        self._iterator = iterator
-        self._paused = False
-        self.tracing_scope = start_active_span(
-            "write_bytes_to_request",
-        )
-        self.tracing_scope.__enter__()
-
-        try:
-            self._request.registerProducer(self, True)
-        except AttributeError as e:
-            # Calling self._request.registerProducer might raise an AttributeError since
-            # the underlying Twisted code calls self._request.channel.registerProducer,
-            # however self._request.channel will be None if the connection was lost.
-            logger.info("Connection disconnected before response was written: %r", e)
-
-            # We drop our references to data we'll not use.
-            self._iterator = iter(())
-            self.tracing_scope.__exit__(type(e), None, e.__traceback__)
-        else:
-            # Start producing if `registerProducer` was successful
-            self.resumeProducing()
-
-    def _send_data(self, data: list[bytes]) -> None:
-        """
-        Send a list of bytes as a chunk of a response.
-        """
-        if not data or not self._request:
-            return
-        self._request.write(b"".join(data))
-
-    def pauseProducing(self) -> None:
-        opentracing_span = active_span()
-        if opentracing_span is not None:
-            opentracing_span.log_kv({"event": "producer_paused"})
-        self._paused = True
-
-    def resumeProducing(self) -> None:
-        # We've stopped producing in the meantime (note that this might be
-        # re-entrant after calling write).
-        if not self._request:
-            return
-
-        self._paused = False
-
-        opentracing_span = active_span()
-        if opentracing_span is not None:
-            opentracing_span.log_kv({"event": "producer_resumed"})
-
-        # Write until there's backpressure telling us to stop.
-        while not self._paused:
-            # Get the next chunk and write it to the request.
-            #
-            # The output of the JSON encoder is buffered and coalesced until
-            # min_chunk_size is reached. This is because JSON encoders produce
-            # very small output per iteration and the Request object converts
-            # each call to write() to a separate chunk. Without this there would
-            # be an explosion in bytes written (e.g. b"{" becoming "1\r\n{\r\n").
-            #
-            # Note that buffer stores a list of bytes (instead of appending to
-            # bytes) to hopefully avoid many allocations.
-            buffer = []
-            buffered_bytes = 0
-            while buffered_bytes < self.min_chunk_size:
-                try:
-                    data = next(self._iterator)
-                    buffer.append(data)
-                    buffered_bytes += len(data)
-                except StopIteration:
-                    # The entire JSON object has been serialized, write any
-                    # remaining data, finalize the producer and the request, and
-                    # clean-up any references.
-                    self._send_data(buffer)
-                    self._request.unregisterProducer()
-                    self._request.finish()
-                    self.stopProducing()
-                    return
-
-            self._send_data(buffer)
-
-    def stopProducing(self) -> None:
-        # Clear a circular reference.
-        self._request = None
-        self.tracing_scope.__exit__(None, None, None)
-
-
 def _encode_json_bytes(json_object: object) -> bytes:
     """
-    Encode an object into JSON. Returns an iterator of bytes.
+    Encode an object into JSON. Returns bytes.
     """
     return json_encoder.encode(json_object).encode("utf-8")
 
@@ -836,7 +712,7 @@ def respond_with_json(
     json_object: Any,
     send_cors: bool = False,
     canonical_json: bool = True,
-) -> int | None:
+) -> None:
     """Sends encoded JSON in response to the given request.
 
     Args:
@@ -847,21 +723,15 @@ def respond_with_json(
             https://fetch.spec.whatwg.org/#http-cors-protocol
         canonical_json: Whether to use the canonicaljson algorithm when encoding
             the JSON bytes.
-
-    Returns:
-        twisted.web.server.NOT_DONE_YET if the request is still active.
     """
     # The response code must always be set, for logging purposes.
     request.setResponseCode(code)
 
-    # could alternatively use request.notifyFinish() and flip a flag when
-    # the Deferred fires, but since the flag is RIGHT THERE it seems like
-    # a waste.
     if request._disconnected:
         logger.warning(
             "Not sending response to request %s, already disconnected.", request
         )
-        return None
+        return
 
     if canonical_json:
         encoder: Callable[[object], bytes] = encode_canonical_json
@@ -885,10 +755,11 @@ def respond_with_json(
     if send_cors:
         set_cors_headers(request)
 
-    run_in_background(
-        _async_write_json_to_request_in_thread, request, encoder, json_object
-    )
-    return NOT_DONE_YET
+    # Encode and write the JSON directly — response is buffered in the shim.
+    json_bytes = encoder(json_object)
+    request.setHeader(b"Content-Length", b"%d" % (len(json_bytes),))
+    request.write(json_bytes)
+    finish_request(request)
 
 
 def respond_with_json_bytes(
@@ -896,7 +767,7 @@ def respond_with_json_bytes(
     code: int,
     json_bytes: bytes,
     send_cors: bool = False,
-) -> int | None:
+) -> None:
     """Sends encoded JSON in response to the given request.
 
     Args:
@@ -905,9 +776,6 @@ def respond_with_json_bytes(
         json_bytes: The json bytes to use as the response body.
         send_cors: Whether to send Cross-Origin Resource Sharing headers
             https://fetch.spec.whatwg.org/#http-cors-protocol
-
-    Returns:
-        twisted.web.server.NOT_DONE_YET if the request is still active.
     """
     # The response code must always be set, for logging purposes.
     request.setResponseCode(code)
@@ -916,7 +784,7 @@ def respond_with_json_bytes(
         logger.warning(
             "Not sending response to request %s, already disconnected.", request
         )
-        return None
+        return
 
     request.setHeader(b"Content-Type", b"application/json")
     request.setHeader(b"Content-Length", b"%d" % (len(json_bytes),))
@@ -936,68 +804,8 @@ def respond_with_json_bytes(
     if send_cors:
         set_cors_headers(request)
 
-    _write_bytes_to_request(request, json_bytes)
-    return NOT_DONE_YET
-
-
-async def _async_write_json_to_request_in_thread(
-    request: "SynapseRequest",
-    json_encoder: Callable[[Any], bytes],
-    json_object: Any,
-) -> None:
-    """Encodes the given JSON object on a thread and then writes it to the
-    request.
-
-    This is done so that encoding large JSON objects doesn't block the reactor
-    thread.
-
-    Note: We don't use JsonEncoder.iterencode here as that falls back to the
-    Python implementation (rather than the C backend), which is *much* more
-    expensive.
-    """
-
-    def encode(opentracing_span: "opentracing.Span | None") -> bytes:
-        # it might take a while for the threadpool to schedule us, so we write
-        # opentracing logs once we actually get scheduled, so that we can see how
-        # much that contributed.
-        if opentracing_span:
-            opentracing_span.log_kv({"event": "scheduled"})
-        res = json_encoder(json_object)
-        if opentracing_span:
-            opentracing_span.log_kv({"event": "encoded"})
-        return res
-
-    with start_active_span("encode_json_response"):
-        span = active_span()
-        json_str = await defer_to_thread(request.reactor, encode, span)
-
-    _write_bytes_to_request(request, json_str)
-
-
-def _write_bytes_to_request(request: Request, bytes_to_write: bytes) -> None:
-    """Writes the bytes to the request using an appropriate producer.
-
-    Note: This should be used instead of `Request.write` to correctly handle
-    large response bodies.
-    """
-
-    # The problem with dumping all of the response into the `Request` object at
-    # once (via `Request.write`) is that doing so starts the timeout for the
-    # next request to be received: so if it takes longer than 60s to stream back
-    # the response to the client, the client never gets it.
-    # c.f https://github.com/twisted/twisted/issues/12498
-    #
-    # One workaround is to use a `Producer`; then the timeout is only
-    # started once all of the content is sent over the TCP connection.
-
-    # To make sure we don't write all of the bytes at once we split it up into
-    # chunks.
-    chunk_size = 4096
-    bytes_generator = chunk_seq(bytes_to_write, chunk_size)
-
-    # We use a `_ByteProducer` here rather than `NoRangeStaticProducer` as the
-    # unit tests can't cope with being given a pull producer.
-    _ByteProducer(request, bytes_generator)
+    request.write(json_bytes)
+    finish_request(request)
 
 
 def set_cors_headers(request: "SynapseRequest") -> None:
@@ -1034,7 +842,7 @@ def set_cors_headers(request: "SynapseRequest") -> None:
         )
 
 
-def set_corp_headers(request: Request) -> None:
+def set_corp_headers(request: "SynapseRequest") -> None:
     """Set the CORP headers so that javascript running in a web browsers can
     embed the resource returned from this request when their client requires
     the `Cross-Origin-Embedder-Policy: require-corp` header.
@@ -1045,14 +853,14 @@ def set_corp_headers(request: Request) -> None:
     request.setHeader(b"Cross-Origin-Resource-Policy", b"cross-origin")
 
 
-def respond_with_html(request: Request, code: int, html: str) -> None:
+def respond_with_html(request: "SynapseRequest", code: int, html: str) -> None:
     """
     Wraps `respond_with_html_bytes` by first encoding HTML from a str to UTF-8 bytes.
     """
     respond_with_html_bytes(request, code, html.encode("utf-8"))
 
 
-def respond_with_html_bytes(request: Request, code: int, html_bytes: bytes) -> None:
+def respond_with_html_bytes(request: "SynapseRequest", code: int, html_bytes: bytes) -> None:
     """
     Sends HTML (encoded as UTF-8 bytes) as the response to the given request.
 
@@ -1066,9 +874,6 @@ def respond_with_html_bytes(request: Request, code: int, html_bytes: bytes) -> N
     # The response code must always be set, for logging purposes.
     request.setResponseCode(code)
 
-    # could alternatively use request.notifyFinish() and flip a flag when
-    # the Deferred fires, but since the flag is RIGHT THERE it seems like
-    # a waste.
     if request._disconnected:
         logger.warning(
             "Not sending response to request %s, already disconnected.", request
@@ -1085,7 +890,7 @@ def respond_with_html_bytes(request: Request, code: int, html_bytes: bytes) -> N
     finish_request(request)
 
 
-def set_clickjacking_protection_headers(request: Request) -> None:
+def set_clickjacking_protection_headers(request: "SynapseRequest") -> None:
     """
     Set headers to guard against clickjacking of embedded content.
 
@@ -1122,18 +927,11 @@ def respond_with_redirect(
     finish_request(request)
 
 
-def finish_request(request: Request) -> None:
+def finish_request(request: "SynapseRequest") -> None:
     """Finish writing the response to the request.
 
-    Twisted throws a RuntimeException if the connection closed before the
-    response was written but doesn't provide a convenient or reliable way to
-    determine if the connection was closed. So we catch and log the RuntimeException
-
-    You might think that ``request.notifyFinish`` could be used to tell if the
-    request was finished. However the deferred it returns won't fire if the
-    connection was already closed, meaning we'd have to have called the method
-    right at the start of the request. By the time we want to write the response
-    it will already be too late.
+    Catches RuntimeError in case the request has already been finished or the
+    connection was closed.
     """
     try:
         request.finish()
diff --git a/synapse/http/servlet.py b/synapse/http/servlet.py
index adb85fcc87..29d16aa096 100644
--- a/synapse/http/servlet.py
+++ b/synapse/http/servlet.py
@@ -36,11 +36,6 @@ from typing import (
 
 from pydantic import BaseModel, ValidationError
 
-try:
-    from twisted.web.server import Request
-except ImportError:
-    pass
-
 from synapse.api.errors import Codes, SynapseError
 from synapse.http import redact_uri
 from synapse.http.server import HttpServer
@@ -48,6 +43,7 @@ from synapse.types import JsonDict, RoomAlias, RoomID, StrCollection
 from synapse.util.json import json_decoder
 
 if TYPE_CHECKING:
+    from synapse.http.aiohttp_shim import SynapseRequest as Request
     from synapse.server import HomeServer
 
 logger = logging.getLogger(__name__)
diff --git a/synapse/http/site.py b/synapse/http/site.py
index 9c6ed9d7a9..ac7ab99334 100644
--- a/synapse/http/site.py
+++ b/synapse/http/site.py
@@ -18,943 +18,24 @@
 # [This file includes modifications made by New Vector Limited]
 #
 #
-import contextlib
-import json
-import logging
-import time
-from http import HTTPStatus
-from typing import TYPE_CHECKING, Any, Generator
 
-import attr
-try:
-    from zope.interface import implementer
-except ImportError:
-    pass
+"""
+Backward-compatibility shim.
 
-try:
-    from twisted.internet.address import UNIXAddress
-    from twisted.internet.defer import Deferred
-    from twisted.internet.interfaces import IAddress
-    from twisted.internet.protocol import Protocol
-    from twisted.python.failure import Failure
-    from twisted.web.http import HTTPChannel
-    from twisted.web.resource import IResource, Resource
-    from twisted.web.server import Request
-except ImportError:
-    pass
+The canonical implementations of ``SynapseRequest``, ``SynapseSite``, and
+``RequestInfo`` now live in ``synapse.http.aiohttp_shim``.  This module
+re-exports them so that the many existing ``from synapse.http.site import …``
+statements throughout the codebase continue to work.
+"""
 
-from synapse.api.errors import Codes, SynapseError
-from synapse.config.server import ListenerConfig
-from synapse.http import get_request_user_agent, redact_uri
-from synapse.http.proxy import ProxySite
-from synapse.http.request_metrics import RequestMetrics, requests_counter
-from synapse.logging.context import (
-    ContextRequest,
-    LoggingContext,
-    PreserveLoggingContext,
+from synapse.http.aiohttp_shim import (  # noqa: F401
+    RequestInfo,
+    SynapseRequest,
+    SynapseSite,
 )
-from synapse.metrics import SERVER_NAME_LABEL
-from synapse.types import ISynapseReactor, Requester
 
-if TYPE_CHECKING:
-    import opentracing
-
-    from synapse.server import HomeServer
-
-
-logger = logging.getLogger(__name__)
-
-_next_request_seq = 0
-
-
-class ContentLengthError(SynapseError):
-    """Raised when content-length validation fails."""
-
-
-class SynapseRequest(Request):
-    """Class which encapsulates an HTTP request to synapse.
-
-    All of the requests processed in synapse are of this type.
-
-    It extends twisted's twisted.web.server.Request, and adds:
-     * Unique request ID
-     * A log context associated with the request
-     * Redaction of access_token query-params in __repr__
-     * Logging at start and end
-     * Metrics to record CPU, wallclock and DB time by endpoint.
-     * A limit to the size of request which will be accepted
-
-    It also provides a method `processing`, which returns a context manager. If this
-    method is called, the request won't be logged until the context manager is closed;
-    this is useful for asynchronous request handlers which may go on processing the
-    request even after the client has disconnected.
-
-    Attributes:
-        logcontext: the log context for this request
-    """
-
-    def __init__(
-        self,
-        channel: HTTPChannel,
-        site: "SynapseSite",
-        our_server_name: str,
-        *args: Any,
-        max_request_body_size: int = 1024,
-        request_id_header: str | None = None,
-        **kw: Any,
-    ):
-        super().__init__(channel, *args, **kw)
-        self.our_server_name = our_server_name
-        self._max_request_body_size = max_request_body_size
-        self.request_id_header = request_id_header
-        self.synapse_site = site
-        self.reactor = site.reactor
-        self._channel = channel  # this is used by the tests
-        self.start_time = 0.0
-
-        # The requester, if authenticated. For federation requests this is the
-        # server name, for client requests this is the Requester object.
-        self._requester: Requester | str | None = None
-
-        # An opentracing span for this request. Will be closed when the request is
-        # completely processed.
-        self._opentracing_span: "opentracing.Span | None" = None
-
-        # we can't yet create the logcontext, as we don't know the method.
-        self.logcontext: LoggingContext | None = None
-
-        # The `Deferred` to cancel if the client disconnects early and
-        # `is_render_cancellable` is set. Expected to be set by `Resource.render`.
-        self.render_deferred: "Deferred[None]" | None = None
-        # A boolean indicating whether `render_deferred` should be cancelled if the
-        # client disconnects early. Expected to be set by the coroutine started by
-        # `Resource.render`, if rendering is asynchronous.
-        self.is_render_cancellable: bool = False
-
-        global _next_request_seq
-        self.request_seq = _next_request_seq
-        _next_request_seq += 1
-
-        # whether an asynchronous request handler has called processing()
-        self._is_processing = False
-
-        # the time when the asynchronous request handler completed its processing
-        self._processing_finished_time: float | None = None
-
-        # what time we finished sending the response to the client (or the connection
-        # dropped)
-        self.finish_time: float | None = None
-
-    def __repr__(self) -> str:
-        # We overwrite this so that we don't log ``access_token``
-        return "<%s at 0x%x method=%r uri=%r clientproto=%r site=%r>" % (
-            self.__class__.__name__,
-            id(self),
-            self.get_method(),
-            self.get_redacted_uri(),
-            self.clientproto.decode("ascii", errors="replace"),
-            self.synapse_site.site_tag,
-        )
-
-    def _respond_with_error(self, synapse_error: SynapseError) -> None:
-        """Send an error response and close the connection."""
-        self.setResponseCode(synapse_error.code)
-        error_response_bytes = json.dumps(synapse_error.error_dict(None)).encode()
-
-        self.responseHeaders.setRawHeaders(b"Content-Type", [b"application/json"])
-        self.responseHeaders.setRawHeaders(
-            b"Content-Length", [f"{len(error_response_bytes)}"]
-        )
-        self.write(error_response_bytes)
-        self.loseConnection()
-
-    def _get_content_length_from_headers(self) -> int | None:
-        """Attempts to obtain the `Content-Length` value from the request's headers.
-
-        Returns:
-            Content length as `int` if present. Otherwise `None`.
-
-        Raises:
-            ContentLengthError: if multiple `Content-Length` headers are present or the
-                value is not an `int`.
-        """
-        content_length_headers = self.requestHeaders.getRawHeaders(b"Content-Length")
-        if content_length_headers is None:
-            return None
-
-        # If there are multiple `Content-Length` headers return an error.
-        # We don't want to even try to pick the right one if there are multiple
-        # as we could run into problems similar to request smuggling vulnerabilities
-        # which rely on the mismatch of how different systems interpret information.
-        if len(content_length_headers) != 1:
-            raise ContentLengthError(
-                HTTPStatus.BAD_REQUEST,
-                "Multiple Content-Length headers received",
-                Codes.UNKNOWN,
-            )
-
-        try:
-            return int(content_length_headers[0])
-        except (ValueError, TypeError):
-            raise ContentLengthError(
-                HTTPStatus.BAD_REQUEST,
-                "Content-Length header value is not a valid integer",
-                Codes.UNKNOWN,
-            )
-
-    def _validate_content_length(self) -> None:
-        """Validate Content-Length header and actual content size.
-
-        Raises:
-            ContentLengthError: If validation fails.
-        """
-        # we should have a `content` by now.
-        assert self.content, "_validate_content_length() called before gotLength()"
-        content_length = self._get_content_length_from_headers()
-
-        if content_length is None:
-            return
-
-        actual_content_length = self.content.tell()
-
-        if content_length > self._max_request_body_size:
-            logger.info(
-                "Rejecting request from %s because Content-Length %d exceeds maximum size %d: %s %s",
-                self.client,
-                content_length,
-                self._max_request_body_size,
-                self.get_method(),
-                self.get_redacted_uri(),
-            )
-            raise ContentLengthError(
-                HTTPStatus.REQUEST_ENTITY_TOO_LARGE,
-                f"Request content is too large (>{self._max_request_body_size})",
-                Codes.TOO_LARGE,
-            )
-
-        if content_length != actual_content_length:
-            comparison = (
-                "smaller" if content_length < actual_content_length else "larger"
-            )
-            logger.info(
-                "Rejecting request from %s because Content-Length %d is %s than the request content size %d: %s %s",
-                self.client,
-                content_length,
-                comparison,
-                actual_content_length,
-                self.get_method(),
-                self.get_redacted_uri(),
-            )
-            raise ContentLengthError(
-                HTTPStatus.BAD_REQUEST,
-                f"Rejecting request as the Content-Length header value {content_length} "
-                f"is {comparison} than the actual request content size {actual_content_length}",
-                Codes.UNKNOWN,
-            )
-
-    # Twisted machinery: this method is called by the Channel once the full request has
-    # been received, to dispatch the request to a resource.
-    def requestReceived(self, command: bytes, path: bytes, version: bytes) -> None:
-        # In the case of a Content-Length header being present, and it's value being too
-        # large, throw a proper error to make debugging issues due to overly large requests much
-        # easier. Currently we handle such cases in `handleContentChunk` and abort the
-        # connection without providing a proper HTTP response.
-        #
-        # Attempting to write an HTTP response from within `handleContentChunk` does not
-        # work, so the code here has been added to at least provide a response in the
-        # case of the Content-Length header being present.
-        self.method, self.uri = command, path
-        self.clientproto = version
-
-        try:
-            self._validate_content_length()
-        except ContentLengthError as e:
-            self._respond_with_error(e)
-            return
-
-        # We're patching Twisted to bail/abort early when we see someone trying to upload
-        # `multipart/form-data` so we can avoid Twisted parsing the entire request body into
-        # in-memory (specific problem of this specific `Content-Type`). This protects us
-        # from an attacker uploading something bigger than the available RAM and crashing
-        # the server with a `MemoryError`, or carefully block just enough resources to cause
-        # all other requests to fail.
-        #
-        # FIXME: This can be removed once Twisted releases a fix and we update to a
-        # version that is patched
-        # See: https://github.com/element-hq/synapse/security/advisories/GHSA-rfq8-j7rh-8hf2
-        if command == b"POST":
-            ctype = self.requestHeaders.getRawHeaders(b"content-type")
-            if ctype and b"multipart/form-data" in ctype[0]:
-                logger.warning(
-                    "Aborting connection from %s because `content-type: multipart/form-data` is unsupported: %s %s",
-                    self.client,
-                    self.get_method(),
-                    self.get_redacted_uri(),
-                )
-
-                self.code = HTTPStatus.UNSUPPORTED_MEDIA_TYPE.value
-                self.code_message = bytes(
-                    HTTPStatus.UNSUPPORTED_MEDIA_TYPE.phrase, "ascii"
-                )
-
-                # FIXME: Return a better error response here similar to the
-                # `error_response_json` returned in other code paths here.
-                self.responseHeaders.setRawHeaders(b"Content-Length", [b"0"])
-                self.write(b"")
-                self.loseConnection()
-                return
-        return super().requestReceived(command, path, version)
-
-    def handleContentChunk(self, data: bytes) -> None:
-        # we should have a `content` by now.
-        assert self.content, "handleContentChunk() called before gotLength()"
-        if self.content.tell() + len(data) > self._max_request_body_size:
-            logger.warning(
-                "Aborting connection from %s because the request exceeds maximum size: %s %s",
-                self.client,
-                self.get_method(),
-                self.get_redacted_uri(),
-            )
-            if self.channel:
-                self.channel.forceAbortClient()
-            return
-        super().handleContentChunk(data)
-
-    @property
-    def requester(self) -> Requester | str | None:
-        return self._requester
-
-    @requester.setter
-    def requester(self, value: Requester | str) -> None:
-        # Store the requester, and update some properties based on it.
-
-        # This should only be called once.
-        assert self._requester is None
-
-        self._requester = value
-
-        # A logging context should exist by now (and have a ContextRequest).
-        assert self.logcontext is not None
-        assert self.logcontext.request is not None
-
-        (
-            requester,
-            authenticated_entity,
-        ) = self.get_authenticated_entity()
-        self.logcontext.request.requester = requester
-        # If there's no authenticated entity, it was the requester.
-        self.logcontext.request.authenticated_entity = authenticated_entity or requester
-
-    def set_opentracing_span(self, span: "opentracing.Span") -> None:
-        """attach an opentracing span to this request
-
-        Doing so will cause the span to be closed when we finish processing the request
-        """
-        self._opentracing_span = span
-
-    def get_request_id(self) -> str:
-        request_id_value = None
-        if self.request_id_header:
-            request_id_value = self.getHeader(self.request_id_header)
-
-        if request_id_value is None:
-            request_id_value = str(self.request_seq)
-
-        return "%s-%s" % (self.get_method(), request_id_value)
-
-    def get_redacted_uri(self) -> str:
-        """Gets the redacted URI associated with the request (or placeholder if the URI
-        has not yet been received).
-
-        Note: This is necessary as the placeholder value in twisted is str
-        rather than bytes, so we need to sanitise `self.uri`.
-
-        Returns:
-            The redacted URI as a string.
-        """
-        uri: bytes | str = self.uri
-        if isinstance(uri, bytes):
-            uri = uri.decode("ascii", errors="replace")
-        return redact_uri(uri)
-
-    def get_method(self) -> str:
-        """Gets the method associated with the request (or placeholder if method
-        has not yet been received).
-
-        Note: This is necessary as the placeholder value in twisted is str
-        rather than bytes, so we need to sanitise `self.method`.
-
-        Returns:
-            The request method as a string.
-        """
-        method: bytes | str = self.method
-        if isinstance(method, bytes):
-            return self.method.decode("ascii")
-        return method
-
-    def get_authenticated_entity(self) -> tuple[str | None, str | None]:
-        """
-        Get the "authenticated" entity of the request, which might be the user
-        performing the action, or a user being puppeted by a server admin.
-
-        Returns:
-            A tuple:
-                The first item is a string representing the user making the request.
-
-                The second item is a string or None representing the user who
-                authenticated when making this request. See
-                Requester.authenticated_entity.
-        """
-        # Convert the requester into a string that we can log
-        if isinstance(self._requester, str):
-            return self._requester, None
-        elif isinstance(self._requester, Requester):
-            requester = self._requester.user.to_string()
-            authenticated_entity = self._requester.authenticated_entity
-
-            # If this is a request where the target user doesn't match the user who
-            # authenticated (e.g. and admin is puppetting a user) then we return both.
-            if requester != authenticated_entity:
-                return requester, authenticated_entity
-
-            return requester, None
-        elif self._requester is not None:
-            # This shouldn't happen, but we log it so we don't lose information
-            # and can see that we're doing something wrong.
-            return repr(self._requester), None  # type: ignore[unreachable]
-
-        return None, None
-
-    def render(self, resrc: Resource) -> None:
-        # this is called once a Resource has been found to serve the request; in our
-        # case the Resource in question will normally be a JsonResource.
-
-        # Create a LogContext for this request
-        #
-        # We only care about associating logs and tallying up metrics at the per-request
-        # level so we don't worry about setting the `parent_context`; preventing us from
-        # unnecessarily piling up metrics on the main process's context.
-        request_id = self.get_request_id()
-        self.logcontext = LoggingContext(
-            name=request_id,
-            server_name=self.our_server_name,
-            request=ContextRequest(
-                request_id=request_id,
-                ip_address=self.get_client_ip_if_available(),
-                site_tag=self.synapse_site.site_tag,
-                # The requester is going to be unknown at this point.
-                requester=None,
-                authenticated_entity=None,
-                method=self.get_method(),
-                url=self.get_redacted_uri(),
-                protocol=self.clientproto.decode("ascii", errors="replace"),
-                user_agent=get_request_user_agent(self),
-            ),
-        )
-
-        # override the Server header which is set by twisted
-        self.setHeader("Server", self.synapse_site.server_version_string)
-
-        with PreserveLoggingContext(self.logcontext):
-            # we start the request metrics timer here with an initial stab
-            # at the servlet name. For most requests that name will be
-            # JsonResource (or a subclass), and JsonResource._async_render
-            # will update it once it picks a servlet.
-            servlet_name = resrc.__class__.__name__
-            self._started_processing(servlet_name)
-
-            Request.render(self, resrc)
-
-            # record the arrival of the request *after*
-            # dispatching to the handler, so that the handler
-            # can update the servlet name in the request
-            # metrics
-            requests_counter.labels(
-                method=self.get_method(),
-                servlet=self.request_metrics.name,
-                **{SERVER_NAME_LABEL: self.our_server_name},
-            ).inc()
-
-    @contextlib.contextmanager
-    def processing(self) -> Generator[None, None, None]:
-        """Record the fact that we are processing this request.
-
-        Returns a context manager; the correct way to use this is:
-
-        async def handle_request(request):
-            with request.processing("FooServlet"):
-                await really_handle_the_request()
-
-        Once the context manager is closed, the completion of the request will be logged,
-        and the various metrics will be updated.
-        """
-        if self._is_processing:
-            raise RuntimeError("Request is already processing")
-        self._is_processing = True
-
-        try:
-            yield
-        except Exception:
-            # this should already have been caught, and sent back to the client as a 500.
-            logger.exception(
-                "Asynchronous message handler raised an uncaught exception"
-            )
-        finally:
-            # the request handler has finished its work and either sent the whole response
-            # back, or handed over responsibility to a Producer.
-
-            self._processing_finished_time = time.time()
-            self._is_processing = False
-
-            if self._opentracing_span:
-                self._opentracing_span.log_kv({"event": "finished processing"})
-
-            # if we've already sent the response, log it now; otherwise, we wait for the
-            # response to be sent.
-            if self.finish_time is not None:
-                self._finished_processing()
-
-    def finish(self) -> None:
-        """Called when all response data has been written to this Request.
-
-        Overrides twisted.web.server.Request.finish to record the finish time and do
-        logging.
-        """
-        self.finish_time = time.time()
-        Request.finish(self)
-        if self._opentracing_span:
-            self._opentracing_span.log_kv({"event": "response sent"})
-        if not self._is_processing:
-            assert self.logcontext is not None
-            with PreserveLoggingContext(self.logcontext):
-                self._finished_processing()
-
-    def connectionLost(self, reason: Failure | Exception) -> None:
-        """Called when the client connection is closed before the response is written.
-
-        Overrides twisted.web.server.Request.connectionLost to record the finish time and
-        do logging.
-        """
-        # There is a bug in Twisted where reason is not wrapped in a Failure object
-        # Detect this and wrap it manually as a workaround
-        # More information: https://github.com/matrix-org/synapse/issues/7441
-        if not isinstance(reason, Failure):
-            reason = Failure(reason)
-
-        self.finish_time = time.time()
-        Request.connectionLost(self, reason)
-
-        if self.logcontext is None:
-            logger.info(
-                "Connection from %s lost before request headers were read", self.client
-            )
-            return
-
-        # we only get here if the connection to the client drops before we send
-        # the response.
-        #
-        # It's useful to log it here so that we can get an idea of when
-        # the client disconnects.
-        with PreserveLoggingContext(self.logcontext):
-            logger.info("Connection from client lost before response was sent")
-
-            if self._opentracing_span:
-                self._opentracing_span.log_kv(
-                    {"event": "client connection lost", "reason": str(reason.value)}
-                )
-
-            if self._is_processing:
-                if self.is_render_cancellable:
-                    if self.render_deferred is not None:
-                        # Throw a cancellation into the request processing, in the hope
-                        # that it will finish up sooner than it normally would.
-                        # The `self.processing()` context manager will call
-                        # `_finished_processing()` when done.
-                        with PreserveLoggingContext():
-                            self.render_deferred.cancel()
-                    else:
-                        logger.error(
-                            "Connection from client lost, but have no Deferred to "
-                            "cancel even though the request is marked as cancellable."
-                        )
-            else:
-                self._finished_processing()
-
-    def _started_processing(self, servlet_name: str) -> None:
-        """Record the fact that we are processing this request.
-
-        This will log the request's arrival. Once the request completes,
-        be sure to call finished_processing.
-
-        Args:
-            servlet_name: the name of the servlet which will be
-                processing this request. This is used in the metrics.
-
-                It is possible to update this afterwards by updating
-                self.request_metrics.name.
-        """
-        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()
-        )
-
-        self.synapse_site.access_logger.debug(
-            "%s - %s - Received request: %s %s",
-            self.get_client_ip_if_available(),
-            self.synapse_site.site_tag,
-            self.get_method(),
-            self.get_redacted_uri(),
-        )
-
-    def _finished_processing(self) -> None:
-        """Log the completion of this request and update the metrics"""
-        assert self.logcontext is not None
-        assert self.finish_time is not None
-
-        usage = self.logcontext.get_resource_usage()
-
-        if self._processing_finished_time is None:
-            # we completed the request without anything calling processing()
-            self._processing_finished_time = time.time()
-
-        # the time between receiving the request and the request handler finishing
-        processing_time = self._processing_finished_time - self.start_time
-
-        # the time between the request handler finishing and the response being sent
-        # to the client (nb may be negative)
-        response_send_time = self.finish_time - self._processing_finished_time
-
-        user_agent = get_request_user_agent(self, "-")
-
-        # int(self.code) looks redundant, because self.code is already an int.
-        # But self.code might be an HTTPStatus (which inherits from int)---which has
-        # a different string representation. So ensure we really have an integer.
-        code = str(int(self.code))
-        if not self.finished:
-            # we didn't send the full response before we gave up (presumably because
-            # the connection dropped)
-            code += "!"
-
-        log_level = logging.INFO if self._should_log_request() else logging.DEBUG
-
-        # If this is a request where the target user doesn't match the user who
-        # authenticated (e.g. and admin is puppetting a user) then we log both.
-        requester, authenticated_entity = self.get_authenticated_entity()
-        if authenticated_entity:
-            requester = f"{authenticated_entity}|{requester}"
-
-        # Updates to this log line should also be reflected in our docs,
-        # `docs/usage/administration/request_log.md`
-        self.synapse_site.access_logger.log(
-            log_level,
-            "%s - %s - {%s}"
-            " Processed request: %.3fsec/%.3fsec ru=(%.3fsec, %.3fsec) db=(%.3fsec/%.3fsec/%d)"
-            ' %sB %s "%s %s %s" "%s" [%d dbevts]',
-            self.get_client_ip_if_available(),
-            self.synapse_site.site_tag,
-            requester,
-            processing_time,
-            response_send_time,
-            usage.ru_utime,
-            usage.ru_stime,
-            usage.db_sched_duration_sec,
-            usage.db_txn_duration_sec,
-            int(usage.db_txn_count),
-            self.sentLength,
-            code,
-            self.get_method(),
-            self.get_redacted_uri(),
-            self.clientproto.decode("ascii", errors="replace"),
-            user_agent,
-            usage.evt_db_fetch_count,
-        )
-
-        # complete the opentracing span, if any.
-        if self._opentracing_span:
-            self._opentracing_span.finish()
-
-        try:
-            self.request_metrics.stop(self.finish_time, self.code, self.sentLength)
-        except Exception as e:
-            logger.warning("Failed to stop metrics: %r", e)
-
-    def _should_log_request(self) -> bool:
-        """Whether we should log at INFO that we processed the request."""
-        if self.path == b"/health":
-            return False
-
-        if self.method == b"OPTIONS":
-            return False
-
-        return True
-
-    def get_client_ip_if_available(self) -> str:
-        """Logging helper. Return something useful when a client IP is not retrievable
-        from a unix socket.
-
-        In practice, this returns the socket file path on a SynapseRequest if using a
-        unix socket and the normal IP address for TCP sockets.
-
-        """
-        # getClientAddress().host returns a proper IP address for a TCP socket. But
-        # unix sockets have no concept of IP addresses or ports and return a
-        # UNIXAddress containing a 'None' value. In order to get something usable for
-        # logs(where this is used) get the unix socket file. getHost() returns a
-        # UNIXAddress containing a value of the socket file and has an instance
-        # variable of 'name' encoded as a byte string containing the path we want.
-        # Decode to utf-8 so it looks nice.
-        if isinstance(self.getClientAddress(), UNIXAddress):
-            return self.getHost().name.decode("utf-8")
-        else:
-            return self.getClientAddress().host
-
-    def request_info(self) -> "RequestInfo":
-        h = self.getHeader(b"User-Agent")
-        user_agent = h.decode("ascii", "replace") if h else None
-        return RequestInfo(user_agent=user_agent, ip=self.get_client_ip_if_available())
-
-
-class XForwardedForRequest(SynapseRequest):
-    """Request object which honours proxy headers
-
-    Extends SynapseRequest to replace getClientIP, getClientAddress, and isSecure with
-    information from request headers.
-    """
-
-    # the client IP and ssl flag, as extracted from the headers.
-    _forwarded_for: "_XForwardedForAddress | None" = None
-    _forwarded_https: bool = False
-
-    def requestReceived(self, command: bytes, path: bytes, version: bytes) -> None:
-        # this method is called by the Channel once the full request has been
-        # received, to dispatch the request to a resource.
-        # We can use it to set the IP address and protocol according to the
-        # headers.
-        self._process_forwarded_headers()
-        return super().requestReceived(command, path, version)
-
-    def _process_forwarded_headers(self) -> None:
-        headers = self.requestHeaders.getRawHeaders(b"x-forwarded-for")
-        if not headers:
-            return
-
-        # for now, we just use the first x-forwarded-for header. Really, we ought
-        # to start from the client IP address, and check whether it is trusted; if it
-        # is, work backwards through the headers until we find an untrusted address.
-        # see https://github.com/matrix-org/synapse/issues/9471
-        self._forwarded_for = _XForwardedForAddress(
-            headers[0].split(b",")[0].strip().decode("ascii")
-        )
-
-        # if we got an x-forwarded-for header, also look for an x-forwarded-proto header
-        header = self.getHeader(b"x-forwarded-proto")
-        if header is not None:
-            self._forwarded_https = header.lower() == b"https"
-        else:
-            # this is done largely for backwards-compatibility so that people that
-            # haven't set an x-forwarded-proto header don't get a redirect loop.
-            logger.warning(
-                "forwarded request lacks an x-forwarded-proto header: assuming https"
-            )
-            self._forwarded_https = True
-
-    def isSecure(self) -> bool:
-        if self._forwarded_https:
-            return True
-        return super().isSecure()
-
-    def getClientIP(self) -> str:
-        """
-        Return the IP address of the client who submitted this request.
-
-        This method is deprecated.  Use getClientAddress() instead.
-        """
-        if self._forwarded_for is not None:
-            return self._forwarded_for.host
-        return super().getClientIP()
-
-    def getClientAddress(self) -> IAddress:
-        """
-        Return the address of the client who submitted this request.
-        """
-        if self._forwarded_for is not None:
-            return self._forwarded_for
-        return super().getClientAddress()
-
-
-@implementer(IAddress)
-@attr.s(frozen=True, slots=True, auto_attribs=True)
-class _XForwardedForAddress:
-    host: str
-
-
-class SynapseProtocol(HTTPChannel):
-    """
-    Synapse-specific twisted http Protocol.
-
-    This is a small wrapper around the twisted HTTPChannel so we can track active
-    connections in order to close any outstanding connections on shutdown.
-    """
-
-    def __init__(
-        self,
-        site: "SynapseSite",
-        our_server_name: str,
-        max_request_body_size: int,
-        request_id_header: str | None,
-        request_class: type,
-    ):
-        super().__init__()
-        self.factory: SynapseSite = site
-        self.site = site
-        self.our_server_name = our_server_name
-        self.max_request_body_size = max_request_body_size
-        self.request_id_header = request_id_header
-        self.request_class = request_class
-
-    def connectionMade(self) -> None:
-        """
-        Called when a connection is made.
-
-        This may be considered the initializer of the protocol, because
-        it is called when the connection is completed.
-
-        Add the connection to the factory's connection list when it's established.
-        """
-        super().connectionMade()
-        self.factory.addConnection(self)
-
-    def connectionLost(self, reason: Failure) -> None:  # type: ignore[override]
-        """
-        Called when the connection is shut down.
-
-        Clear any circular references here, and any external references to this
-        Protocol. The connection has been closed. In our case, we need to remove the
-        connection from the factory's connection list, when it's lost.
-        """
-        super().connectionLost(reason)
-        self.factory.removeConnection(self)
-
-    def requestFactory(self, http_channel: HTTPChannel, queued: bool) -> SynapseRequest:  # type: ignore[override]
-        """
-        A callable used to build `twisted.web.iweb.IRequest` objects.
-
-        Use our own custom SynapseRequest type instead of the regular
-        twisted.web.server.Request.
-        """
-        return self.request_class(
-            self,
-            self.factory,
-            our_server_name=self.our_server_name,
-            max_request_body_size=self.max_request_body_size,
-            queued=queued,
-            request_id_header=self.request_id_header,
-        )
-
-
-class SynapseSite(ProxySite):
-    """
-    Synapse-specific twisted http Site
-
-    This does two main things.
-
-    First, it replaces the requestFactory in use so that we build SynapseRequests
-    instead of regular t.w.server.Requests. All of the  constructor params are really
-    just parameters for SynapseRequest.
-
-    Second, it inhibits the log() method called by Request.finish, since SynapseRequest
-    does its own logging.
-    """
-
-    def __init__(
-        self,
-        *,
-        logger_name: str,
-        site_tag: str,
-        config: ListenerConfig,
-        resource: IResource,
-        server_version_string: str,
-        max_request_body_size: int,
-        reactor: ISynapseReactor,
-        hs: "HomeServer",
-    ):
-        """
-
-        Args:
-            logger_name:  The name of the logger to use for access logs.
-            site_tag:  A tag to use for this site - mostly in access logs.
-            config:  Configuration for the HTTP listener corresponding to this site
-            resource:  The base of the resource tree to be used for serving requests on
-                this site
-            server_version_string: A string to present for the Server header
-            max_request_body_size: Maximum request body length to allow before
-                dropping the connection
-            reactor: reactor to be used to manage connection timeouts
-        """
-        super().__init__(
-            resource=resource,
-            reactor=reactor,
-            hs=hs,
-        )
-
-        self.site_tag = site_tag
-        self.reactor: ISynapseReactor = reactor
-        self.server_name = hs.hostname
-
-        assert config.http_options is not None
-        proxied = config.http_options.x_forwarded
-        self.request_class = XForwardedForRequest if proxied else SynapseRequest
-
-        self.request_id_header = config.http_options.request_id_header
-        self.max_request_body_size = max_request_body_size
-
-        self.access_logger = logging.getLogger(logger_name)
-        self.server_version_string = server_version_string.encode("ascii")
-        self.connections: list[Protocol] = []
-
-    def buildProtocol(self, addr: IAddress) -> SynapseProtocol:
-        protocol = SynapseProtocol(
-            self,
-            self.server_name,
-            self.max_request_body_size,
-            self.request_id_header,
-            self.request_class,
-        )
-        return protocol
-
-    def addConnection(self, protocol: Protocol) -> None:
-        self.connections.append(protocol)
-
-    def removeConnection(self, protocol: Protocol) -> None:
-        if protocol in self.connections:
-            self.connections.remove(protocol)
-
-    def stopFactory(self) -> None:
-        super().stopFactory()
-
-        # Shutdown any connections which are still active.
-        # These can be long lived HTTP connections which wouldn't normally be closed
-        # when calling `shutdown` on the respective `Port`.
-        # Closing the connections here is required for us to fully shutdown the
-        # `SynapseHomeServer` in order for it to be garbage collected.
-        for protocol in self.connections[:]:
-            if protocol.transport is not None:
-                protocol.transport.loseConnection()
-        self.connections.clear()
-
-        # Replace the resource tree with an empty resource to break circular references
-        # to the resource tree which holds a bunch of homeserver references. This is
-        # important if we try to call `hs.shutdown()` after `start` fails. For some
-        # reason, this doesn't seem to be necessary in the normal case where `start`
-        # succeeds and we call `hs.shutdown()` later.
-        self.resource = Resource()
-
-    def log(self, request: SynapseRequest) -> None:  # type: ignore[override]
-        pass
-
-
-@attr.s(auto_attribs=True, frozen=True, slots=True)
-class RequestInfo:
-    user_agent: str | None
-    ip: str
+__all__ = [
+    "SynapseRequest",
+    "SynapseSite",
+    "RequestInfo",
+]
diff --git a/synapse/server.py b/synapse/server.py
index 3c3c88bf91..d0295035e6 100644
--- a/synapse/server.py
+++ b/synapse/server.py
@@ -475,6 +475,12 @@ class HomeServer(metaclass=abc.ABCMeta):
                     await port_shutdown
         self._listening_services.clear()
 
+        # Clean up aiohttp runners (HTTP listeners)
+        from synapse.app._base import _aiohttp_runners
+        for runner in list(_aiohttp_runners):
+            await runner.cleanup()
+        _aiohttp_runners.clear()
+
         for server, thread in self._metrics_listeners:
             server.shutdown()
             thread.join()
diff --git a/tests/app/test_openid_listener.py b/tests/app/test_openid_listener.py
index ec5fac3089..a499d95138 100644
--- a/tests/app/test_openid_listener.py
+++ b/tests/app/test_openid_listener.py
@@ -35,6 +35,13 @@ from tests.unittest import HomeserverTestCase
 
 
 class FederationReaderOpenIDListenerTests(HomeserverTestCase):
+    """Test that the openid listener is correctly configured on workers.
+
+    With the aiohttp migration, we can no longer introspect Twisted's reactor
+    for the listening site. Instead, we test the resource tree construction
+    directly by checking that the appropriate resources are registered.
+    """
+
     def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer:
         hs = self.setup_test_homeserver(homeserver_to_use=GenericWorkerServer)
         return hs
@@ -51,66 +58,91 @@ class FederationReaderOpenIDListenerTests(HomeserverTestCase):
 
     @parameterized.expand(
         [
-            (["federation"], "auth_fail"),
-            ([], "no_resource"),
-            (["openid", "federation"], "auth_fail"),
-            (["openid"], "auth_fail"),
+            (["federation"], True),
+            ([], False),
+            (["openid", "federation"], True),
+            (["openid"], True),
         ]
     )
-    def test_openid_listener(self, names: list[str], expectation: str) -> None:
+    def test_openid_listener(self, names: list[str], expect_federation: bool) -> None:
         """
-        Test different openid listener configurations.
+        Test that the federation resource (which includes openid) is created
+        when the appropriate listener names are configured.
+        """
+        from synapse.http.server import JsonResource, OptionsResource
+        from synapse.util.httpresourcetree import create_resource_tree
+        from synapse.api.urls import FEDERATION_PREFIX
 
-        401 is success here since it means we hit the handler and auth failed.
-        """
         config = {
             "port": 8080,
             "type": "http",
             "bind_addresses": ["0.0.0.0"],
             "resources": [{"names": names}],
         }
+        listener_config = parse_listener_def(0, config)
+        assert listener_config.http_options is not None
 
-        # Listen with the config
+        # Build the resource dict the same way GenericWorkerServer._listen_http does
         hs = self.hs
         assert isinstance(hs, GenericWorkerServer)
-        hs._listen_http(parse_listener_def(0, config))
 
-        # Grab the resource from the site that was told to listen
-        site = self.reactor.tcpServers[0][1]
-        try:
-            site.resource.children[b"_matrix"].children[b"federation"]
-        except KeyError:
-            if expectation == "no_resource":
-                return
-            raise
+        from synapse.rest.health import HealthResource
+        from synapse.federation.transport.server import TransportLayerServer
 
-        channel = make_request(
-            self.reactor, site, "GET", "/_matrix/federation/v1/openid/userinfo"
-        )
+        resources: dict[str, Any] = {
+            "/health": HealthResource(),
+            "/_synapse/admin": JsonResource(hs, canonical_json=False),
+        }
 
-        self.assertEqual(channel.code, 401)
+        for res in listener_config.http_options.resources:
+            for name in res.names:
+                if name == "federation":
+                    resources[FEDERATION_PREFIX] = TransportLayerServer(hs)
+                if name == "openid" and "federation" not in res.names:
+                    resources[FEDERATION_PREFIX] = TransportLayerServer(
+                        hs, servlet_groups=["openid"]
+                    )
+
+        root_resource = create_resource_tree(resources, OptionsResource())
+
+        if expect_federation:
+            # Check the federation resource exists in the tree
+            self.assertIn(b"_matrix", root_resource.listNames())
+        else:
+            # No federation resource should be present
+            if b"_matrix" in root_resource.listNames():
+                matrix_child = root_resource.getStaticEntity(b"_matrix")
+                self.assertNotIn(b"federation", matrix_child.listNames())
 
 
 @patch("synapse.app.homeserver.KeyResource", new=Mock())
 class SynapseHomeserverOpenIDListenerTests(HomeserverTestCase):
+    """Test that the openid listener is correctly configured on the homeserver.
+
+    With the aiohttp migration, we test resource tree construction directly.
+    """
+
     def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer:
         hs = self.setup_test_homeserver(homeserver_to_use=SynapseHomeServer)
         return hs
 
     @parameterized.expand(
         [
-            (["federation"], "auth_fail"),
-            ([], "no_resource"),
-            (["openid", "federation"], "auth_fail"),
-            (["openid"], "auth_fail"),
+            (["federation"], True),
+            ([], False),
+            (["openid", "federation"], True),
+            (["openid"], True),
         ]
     )
-    def test_openid_listener(self, names: list[str], expectation: str) -> None:
+    def test_openid_listener(self, names: list[str], expect_federation: bool) -> None:
         """
-        Test different openid listener configurations.
+        Test that the federation resource (which includes openid) is created
+        when the appropriate listener names are configured.
+        """
+        from synapse.http.server import OptionsResource
+        from synapse.util.httpresourcetree import create_resource_tree
+        from synapse.rest.health import HealthResource
 
-        401 is success here since it means we hit the handler and auth failed.
-        """
         config = {
             "port": 8080,
             "type": "http",
@@ -118,22 +150,29 @@ class SynapseHomeserverOpenIDListenerTests(HomeserverTestCase):
             "resources": [{"names": names}],
         }
 
-        # Listen with the config
         hs = self.hs
         assert isinstance(hs, SynapseHomeServer)
-        hs._listener_http(self.hs.config, parse_listener_def(0, config))
+        listener_config = parse_listener_def(0, config)
+        assert listener_config.http_options is not None
 
-        # Grab the resource from the site that was told to listen
-        site = self.reactor.tcpServers[0][1]
-        try:
-            site.resource.children[b"_matrix"].children[b"federation"]
-        except KeyError:
-            if expectation == "no_resource":
-                return
-            raise
+        # Build resources the same way _listener_http does
+        resources: dict[str, Any] = {"/health": HealthResource()}
 
-        channel = make_request(
-            self.reactor, site, "GET", "/_matrix/federation/v1/openid/userinfo"
+        for res in listener_config.http_options.resources:
+            for name in res.names:
+                if name == "openid" and "federation" in res.names:
+                    continue
+                if name == "health":
+                    continue
+                resources.update(hs._configure_named_resource(name, res.compress))
+
+        root_resource = create_resource_tree(
+            resources, OptionsResource()
         )
 
-        self.assertEqual(channel.code, 401)
+        if expect_federation:
+            self.assertIn(b"_matrix", root_resource.listNames())
+        else:
+            if b"_matrix" in root_resource.listNames():
+                matrix_child = root_resource.getStaticEntity(b"_matrix")
+                self.assertNotIn(b"federation", matrix_child.listNames())
diff --git a/tests/server.py b/tests/server.py
index 9ff0b16424..7c6aae7253 100644
--- a/tests/server.py
+++ b/tests/server.py
@@ -188,25 +188,35 @@ class FakeChannel:
         """
         if not self.is_finished():
             raise Exception("Request not yet completed")
+        # Read from shim request's response buffer
+        if self._request is not None and hasattr(self._request, '_response_buffer'):
+            return bytes(self._request._response_buffer).decode("utf8")
         return self.result["body"].decode("utf8")
 
     def is_finished(self) -> bool:
         """check if the response has been completely received"""
+        # Check the shim request's finished flag
+        if self._request is not None and hasattr(self._request, 'finished'):
+            return self._request.finished
         return self.result.get("done", False)
 
     @property
     def code(self) -> int:
+        if self._request is not None and hasattr(self._request, 'code'):
+            return int(self._request.code)
         if not self.result:
             raise Exception("No result yet.")
         return int(self.result["code"])
 
     @property
-    def headers(self) -> Headers:
+    def headers(self) -> Any:
+        # Return the shim response headers if available
+        if self._request is not None and hasattr(self._request, 'responseHeaders'):
+            return self._request.responseHeaders
         if not self.result:
             raise Exception("No result yet.")
 
         h = self.result["headers"]
-        assert isinstance(h, Headers)
         return h
 
     def writeHeaders(
@@ -455,7 +465,9 @@ def make_request(
 
     channel = FakeChannel(site, reactor, ip=client_ip, clock=clock)
 
-    req = request(
+    # Use the shim's for_testing constructor
+    from synapse.http.aiohttp_shim import SynapseRequest as ShimRequest
+    req = ShimRequest.for_testing(
         channel,
         site,
         our_server_name="test_server",
@@ -463,18 +475,27 @@ def make_request(
     )
     channel.request = req
 
+    req.method = method
+    req.path = path
+    # URI includes query string
+    req.uri = path
     req.content = BytesIO(content)
-    # Twisted expects to be at the end of the content when parsing the request.
     req.content.seek(0, SEEK_END)
+    req._client_ip = client_ip
 
-    # If `Content-Length` was passed in as a custom header, don't automatically add it
-    # here.
+    # Parse query string into args
+    from urllib.parse import parse_qs, urlparse
+    parsed = urlparse(path if isinstance(path, str) else path.decode("utf-8"))
+    if parsed.query:
+        for k, vs in parse_qs(parsed.query, keep_blank_values=True).items():
+            bk = k.encode("utf-8") if isinstance(k, str) else k
+            req.args[bk] = [v.encode("utf-8") if isinstance(v, str) else v for v in vs]
+
+    # Add standard headers
     if custom_headers is None or not any(
         (k if isinstance(k, bytes) else k.encode("ascii")) == b"Content-Length"
         for k, _ in custom_headers
     ):
-        # Old version of Twisted (<20.3.0) have issues with parsing x-www-form-urlencoded
-        # bodies if the Content-Length header is missing
         req.requestHeaders.addRawHeader(
             b"Content-Length", str(len(content)).encode("ascii")
         )
@@ -498,15 +519,40 @@ def make_request(
                 b"Content-Type", b"application/x-www-form-urlencoded"
             )
         else:
-            # Assume the body is JSON
             req.requestHeaders.addRawHeader(b"Content-Type", b"application/json")
 
     if custom_headers:
         for k, v in custom_headers:
             req.requestHeaders.addRawHeader(k, v)
 
-    req.parseCookies()
-    req.requestReceived(method, path, b"1.1")
+    # Initialize request metrics and logcontext before dispatch
+    import asyncio
+    from synapse.http.request_metrics import RequestMetrics
+    from synapse.logging.context import LoggingContext
+
+    req.start_time = time.time()
+    server_name = getattr(site, 'server_name', 'test')
+    req.request_metrics = RequestMetrics(our_server_name=server_name)
+    req.request_metrics.start(req.start_time, name="test", method=req.get_method())
+    req.logcontext = LoggingContext(
+        name="test-%s-%s" % (req.get_method(), req.get_redacted_uri()),
+        server_name=server_name,
+        request=req,
+    )
+
+    # Dispatch the request through the resource
+    resource = getattr(site, 'resource', None) or getattr(site, '_resource', None)
+    if resource is not None and hasattr(resource, '_async_render_wrapper'):
+        req.render_deferred = asyncio.ensure_future(
+            resource._async_render_wrapper(req)
+        )
+    else:
+        # Fallback: try the old Twisted render path for compatibility
+        if resource is not None and hasattr(resource, 'render'):
+            resource.render(req)
+        else:
+            import sys
+            print(f"WARNING: No resource to dispatch to. site={type(site)}, resource={resource}", file=sys.stderr)
 
     if await_result:
         channel.await_result()