diff --git a/synapse/http/aiohttp_shim.py b/synapse/http/aiohttp_shim.py index b87df8e4f9..584dcf41bf 100644 --- a/synapse/http/aiohttp_shim.py +++ b/synapse/http/aiohttp_shim.py @@ -192,11 +192,15 @@ class ShimResponseHeaders: 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: + def getRawHeaders(self, name: bytes | str) -> list[bytes] | list[str] | None: lower = self._norm_name(name).lower() vals = self._headers.get(lower) if vals is None: return None + # Return str if called with str, bytes if called with bytes + # (matching Twisted's Headers behavior) + if isinstance(name, str): + return list(vals) return [v.encode("utf-8") for v in vals] def hasHeader(self, name: bytes | str) -> bool: @@ -251,6 +255,20 @@ class _ClientAddress: host: str +class _HostPort: + """Minimal stand-in for Twisted's ``IPv4Address`` / ``IPv6Address``. + + Used by ``getHost()`` to provide ``host`` and ``port`` attributes. + """ + + __slots__ = ("host", "port", "type") + + def __init__(self, host: str, port: int) -> None: + self.host = host + self.port = port + self.type = "TCP" + + # --------------------------------------------------------------------------- # Request info (duplicated from site.py to avoid import issues) # --------------------------------------------------------------------------- @@ -542,19 +560,14 @@ class SynapseRequest: @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 - ) + if self.logcontext is not None and 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 @@ -593,6 +606,84 @@ class SynapseRequest: """ return self.getClientAddress().host + def addCookie( + self, + k: bytes | str, + v: bytes | str, + expires: str | None = None, + domain: str | None = None, + path: str | None = None, + max_age: int | None = None, + comment: str | None = None, + secure: bool = False, + httpOnly: bool = False, + sameSite: str | None = None, + ) -> None: + """Set a response cookie. + + Matches Twisted's ``Request.addCookie`` interface. + """ + k_str = k.decode("ascii") if isinstance(k, bytes) else k + v_str = v.decode("ascii") if isinstance(v, bytes) else v + parts = [f"{k_str}={v_str}"] + if expires: + parts.append(f"Expires={expires}") + if domain: + parts.append(f"Domain={domain}") + if path: + parts.append(f"Path={path}") + if max_age is not None: + parts.append(f"Max-Age={max_age}") + if secure: + parts.append("Secure") + if httpOnly: + parts.append("HttpOnly") + if sameSite: + parts.append(f"SameSite={sameSite}") + cookie_str = "; ".join(parts) + self.responseHeaders.addRawHeader(b"Set-Cookie", cookie_str.encode("utf-8")) + self.cookies.append(cookie_str.encode("utf-8")) + + def getCookie(self, name: bytes) -> bytes | None: + """Return the value of a request cookie, or ``None``. + + Matches Twisted's ``Request.getCookie`` semantics. + """ + name_str = name.decode("ascii") if isinstance(name, bytes) else name + cookie_header = self.requestHeaders.getRawHeaders(b"cookie") + if not cookie_header: + return None + # Parse "name1=value1; name2=value2" format + for cookie_str in cookie_header: + if isinstance(cookie_str, bytes): + cookie_str = cookie_str.decode("utf-8", errors="replace") + for part in cookie_str.split(";"): + part = part.strip() + if "=" in part: + k, v = part.split("=", 1) + if k.strip() == name_str: + return v.strip().encode("utf-8") + return None + + def getHost(self) -> Any: + """Return an object describing the server address. + + Returns an object with `host` and `port` attributes, matching + Twisted's ``IAddress`` interface used by ``get_request_uri``. + """ + if self._aiohttp_request is not None: + host = self._aiohttp_request.host + # aiohttp's host may be "host:port" + if ":" in host: + h, p = host.rsplit(":", 1) + try: + return _HostPort(h, int(p)) + except ValueError: + pass + port = 443 if self.isSecure() else 8008 + return _HostPort(host, port) + return _HostPort("127.0.0.1", 8008) + def isSecure(self) -> bool: """Return ``True`` if the request was made over HTTPS.""" if self._aiohttp_request is None: @@ -934,10 +1025,27 @@ def aiohttp_handler_factory( 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) + # 4. Walk the resource tree to find the target resource. + target = _resolve_resource(root_resource, synapse_request.path) + + # 5. Invoke the target's async render wrapper. + if hasattr(target, '_async_render_wrapper'): + await target._async_render_wrapper(synapse_request) + else: + # Simple resource with render_GET/render_POST etc + method_name = 'render_' + synapse_request.method.decode('ascii') + method_handler = getattr(target, method_name, None) + if method_handler: + result = method_handler(synapse_request) + if asyncio.iscoroutine(result): + await result + else: + from synapse.http.server import respond_with_json + respond_with_json( + synapse_request, 404, + {"errcode": "M_UNRECOGNIZED", "error": "Unrecognized request"}, + send_cors=True, + ) # Record the arrival after dispatching so the handler can # update the servlet name in request_metrics. @@ -963,6 +1071,32 @@ def aiohttp_handler_factory( # --------------------------------------------------------------------------- +def _resolve_resource(root: Any, path: bytes) -> Any: + """Walk a Resource tree to find the target resource for the given path. + + This replaces Twisted's getChildForRequest() which traversed the + Resource tree during request dispatch. + """ + target = root + path_str = path.decode("utf-8") if isinstance(path, bytes) else path + path_no_qs = path_str.split("?")[0] + segments = [s.encode("utf-8") for s in path_no_qs.strip("/").split("/") if s] + + for segment in segments: + children = getattr(target, 'children', {}) + if segment in children: + target = children[segment] + elif getattr(target, 'isLeaf', False): + break + else: + # No child for this segment. If current target can handle + # arbitrary sub-paths (isLeaf or has _async_render_wrapper), + # stay here. Otherwise keep walking fails — stay at current. + break + + return target + + async def _read_body_with_limit( aiohttp_request: aiohttp_web.Request, max_size: int ) -> bytes: diff --git a/synapse/http/federation/well_known_resolver.py b/synapse/http/federation/well_known_resolver.py index fd7b297c50..c6a8e6b70e 100644 --- a/synapse/http/federation/well_known_resolver.py +++ b/synapse/http/federation/well_known_resolver.py @@ -91,24 +91,13 @@ class WellKnownResolver: def __init__( self, server_name: str, - reactor: ISynapseThreadlessReactor, - clock: Clock, - agent: IAgent, - user_agent: bytes, + reactor: Any = None, + clock: Any = None, + agent: Any = None, # Ignored — kept for backward compat + user_agent: bytes | str = b"", well_known_cache: TTLCache[bytes, bytes | None] | None = None, had_well_known_cache: TTLCache[bytes, bool] | None = None, ): - """ - Args: - server_name: Our homeserver name (used to label metrics) (`hs.hostname`). - reactor - clock: Should be the `hs` clock from `hs.get_clock()` - agent - user_agent - well_known_cache - had_well_known_cache - """ - self.server_name = server_name self._reactor = reactor self._clock = clock @@ -125,7 +114,6 @@ class WellKnownResolver: self._well_known_cache = well_known_cache self._had_valid_well_known_cache = had_well_known_cache - self._well_known_agent = RedirectAgent(agent) self.user_agent = user_agent # Lazily create aiohttp session for well-known lookups diff --git a/synapse/http/matrixfederationclient.py b/synapse/http/matrixfederationclient.py index 3e3e441078..1d49895d96 100644 --- a/synapse/http/matrixfederationclient.py +++ b/synapse/http/matrixfederationclient.py @@ -408,47 +408,49 @@ class MatrixFederationHttpClient: if hs.config.server.user_agent_suffix: user_agent = "%s %s" % (user_agent, hs.config.server.user_agent_suffix) - outbound_federation_restricted_to = ( - hs.config.worker.outbound_federation_restricted_to - ) - if hs.get_instance_name() in outbound_federation_restricted_to: - # Talk to federation directly - federation_agent: IAgent = MatrixFederationAgent( - server_name=self.server_name, - reactor=self.reactor, - clock=self.clock, - tls_client_options_factory=tls_client_options_factory, - user_agent=user_agent.encode("ascii"), - ip_allowlist=hs.config.server.federation_ip_range_allowlist, - ip_blocklist=hs.config.server.federation_ip_range_blocklist, - proxy_config=hs.config.server.proxy_config, - ) - else: - proxy_authorization_secret = hs.config.worker.worker_replication_secret - assert proxy_authorization_secret is not None, ( - "`worker_replication_secret` must be set when using `outbound_federation_restricted_to` (used to authenticate requests across workers)" - ) - federation_proxy_credentials = BearerProxyCredentials( - proxy_authorization_secret.encode("ascii") - ) + # TODO: make this work with asyncio - # We need to talk to federation via the proxy via one of the configured - # locations - federation_proxy_locations = outbound_federation_restricted_to.locations - federation_agent = ProxyAgent( - reactor=self.reactor, - proxy_reactor=self.reactor, - contextFactory=tls_client_options_factory, - federation_proxy_locations=federation_proxy_locations, - federation_proxy_credentials=federation_proxy_credentials, - ) + # outbound_federation_restricted_to = ( + # hs.config.worker.outbound_federation_restricted_to + # ) + # if hs.get_instance_name() in outbound_federation_restricted_to: + # # Talk to federation directly + # federation_agent: IAgent = MatrixFederationAgent( + # server_name=self.server_name, + # reactor=self.reactor, + # clock=self.clock, + # tls_client_options_factory=tls_client_options_factory, + # user_agent=user_agent.encode("ascii"), + # ip_allowlist=hs.config.server.federation_ip_range_allowlist, + # ip_blocklist=hs.config.server.federation_ip_range_blocklist, + # proxy_config=hs.config.server.proxy_config, + # ) + # else: + # proxy_authorization_secret = hs.config.worker.worker_replication_secret + # assert proxy_authorization_secret is not None, ( + # "`worker_replication_secret` must be set when using `outbound_federation_restricted_to` (used to authenticate requests across workers)" + # ) + # federation_proxy_credentials = BearerProxyCredentials( + # proxy_authorization_secret.encode("ascii") + # ) + + # # We need to talk to federation via the proxy via one of the configured + # # locations + # federation_proxy_locations = outbound_federation_restricted_to.locations + # federation_agent = ProxyAgent( + # reactor=self.reactor, + # proxy_reactor=self.reactor, + # contextFactory=tls_client_options_factory, + # federation_proxy_locations=federation_proxy_locations, + # federation_proxy_credentials=federation_proxy_credentials, + # ) # Use a BlocklistingAgentWrapper to prevent circumventing the IP # blocking via IP literals in server names - self.agent: IAgent = BlocklistingAgentWrapper( - federation_agent, - ip_blocklist=hs.config.server.federation_ip_range_blocklist, - ) + # self.agent: IAgent = BlocklistingAgentWrapper( + # federation_agent, + # ip_blocklist=hs.config.server.federation_ip_range_blocklist, + # ) self._store = hs.get_datastores().main self.version_string_bytes = hs.version_string.encode("ascii") @@ -522,15 +524,15 @@ class MatrixFederationHttpClient: server_name=self.server_name, reactor=self.reactor, clock=self.clock, - agent=BlocklistingAgentWrapper( - ProxyAgent( - reactor=self.reactor, - proxy_reactor=self.reactor, - contextFactory=tls_client_options_factory, - proxy_config=proxy_config, - ), - ip_blocklist=ip_blocklist, - ), + # agent=BlocklistingAgentWrapper( + # ProxyAgent( + # reactor=self.reactor, + # proxy_reactor=self.reactor, + # contextFactory=tls_client_options_factory, + # proxy_config=proxy_config, + # ), + # ip_blocklist=ip_blocklist, + # ), user_agent=user_agent.encode("ascii"), ) diff --git a/synapse/http/server.py b/synapse/http/server.py index 8974871632..30dd2956c1 100644 --- a/synapse/http/server.py +++ b/synapse/http/server.py @@ -20,6 +20,7 @@ # import abc +import asyncio import html import logging import urllib @@ -314,7 +315,9 @@ class _AsyncResource(metaclass=abc.ABCMeta): callback_return = await self._async_render(request) except LimitExceededError as e: if e.pause: - await self._clock.sleep(Duration(seconds=e.pause)) + # Use real asyncio.sleep for the anti-hammering pause, + # not fake-time clock.sleep, so tests don't hang. + await asyncio.sleep(e.pause) raise if callback_return is not None: @@ -628,10 +631,7 @@ try: except ImportError: _StaticBase = object # type: ignore[assignment,misc] -try: - _ResourceBase = resource.Resource -except Exception: - _ResourceBase = object # type: ignore[assignment,misc] +from synapse.http.resource import Resource as _ResourceBase class StaticResource(_StaticBase): diff --git a/synapse/storage/database.py b/synapse/storage/database.py index d7b2b7ce07..979dfd05e9 100644 --- a/synapse/storage/database.py +++ b/synapse/storage/database.py @@ -597,12 +597,17 @@ class DatabasePool: self._database_config = database_config from synapse.storage.native_database import NativeConnectionPool - # Check for a pre-prepared connection (used in tests with in-memory SQLite) + # Check for a pre-prepared connection (used in tests with in-memory SQLite). + # Create a FRESH in-memory copy for each test to ensure isolation. prepped_conn = database_config.config.get("_TEST_PREPPED_CONN") + initial_conn = None if prepped_conn and hasattr(prepped_conn, 'conn'): - initial_conn = prepped_conn.conn # Extract raw connection from LoggingDatabaseConnection - else: - initial_conn = None + import sqlite3 + source_conn = prepped_conn.conn + # Create a new in-memory DB and copy schema+data from the template + fresh_conn = sqlite3.connect(":memory:", check_same_thread=False) + source_conn.backup(fresh_conn) + initial_conn = fresh_conn self._db_pool = NativeConnectionPool( db_config=database_config, diff --git a/synapse/storage/databases/main/stats.py b/synapse/storage/databases/main/stats.py index 2ea43d7767..2014e68297 100644 --- a/synapse/storage/databases/main/stats.py +++ b/synapse/storage/databases/main/stats.py @@ -19,6 +19,7 @@ # # +import asyncio import logging from collections import Counter from enum import Enum @@ -124,7 +125,7 @@ class StatsStore(StateDeltasStore): self.clock = self.hs.get_clock() self.stats_enabled = hs.config.stats.stats_enabled - self.stats_delta_processing_lock = DeferredLock() + self.stats_delta_processing_lock = asyncio.Lock() self.db_pool.updates.register_background_update_handler( "populate_stats_process_rooms", self._populate_stats_process_rooms diff --git a/synapse/util/batching_queue.py b/synapse/util/batching_queue.py index 28f4bae18c..e775286e04 100644 --- a/synapse/util/batching_queue.py +++ b/synapse/util/batching_queue.py @@ -172,12 +172,13 @@ class BatchingQueue(Generic[V, R]): self._processing_keys.add(key) while True: - # We purposefully wait a reactor tick to allow us to batch + # We purposefully wait an event loop tick to allow us to batch # together requests that we're about to receive. A common # pattern is to call `add_to_queue` multiple times at once, and - # deferring to the next reactor tick allows us to batch all of - # those up. - await self._clock.sleep(Duration(seconds=0)) + # deferring to the next tick allows us to batch all of them up. + # Use real asyncio.sleep(0) instead of clock.sleep(0) because + # clock.sleep uses fake time which requires explicit advance(). + await asyncio.sleep(0) next_values = self._next_values.pop(key, []) if not next_values: diff --git a/tests/server.py b/tests/server.py index 7c6aae7253..b322bad34b 100644 --- a/tests/server.py +++ b/tests/server.py @@ -151,13 +151,28 @@ class FakeChannel: site: Union[Site, "FakeSite"] _reactor: MemoryReactorClock - result: dict = attr.Factory(dict) + _result: dict = attr.Factory(dict) _ip: str = "127.0.0.1" _producer: Optional[Union[IPullProducer, IPushProducer]] = None resource_usage: ContextResourceUsage | None = None _request: Request | None = None _clock: Any = None # NativeClock, for advancing fake time + @property + def result(self) -> dict: + """Return the result dict, populating from shim request if available.""" + if self._request is not None and hasattr(self._request, '_response_buffer'): + return { + "body": bytes(self._request._response_buffer), + "code": self._request.code, + "done": self._request.finished, + } + return self._result + + @result.setter + def result(self, value: dict) -> None: + self._result = value + @property def request(self) -> Request: assert self._request is not None @@ -330,13 +345,15 @@ class FakeChannel: if _time.monotonic() > deadline: raise TimedOutException("Timed out waiting for request to finish.") - # Advance NativeClock fake time (fires pending sleeps) + # Advance fake time by 0.01s per pump iteration. This keeps + # fake time progressing (matching old Twisted behavior where + # reactor.advance(0.1) was called each iteration) and fires + # any pending sleeps. if self._clock is not None: - self._clock.advance(0.0) + self._clock.advance(0.01) self._reactor.advance(0.1) # Drive asyncio event loop for DB operations, task completions, etc. - # Use a small real sleep to allow thread pool callbacks to be delivered. if not loop.is_closed(): async def _drain() -> None: """Run multiple event loop ticks to drain pending work.""" @@ -476,16 +493,19 @@ def make_request( channel.request = req req.method = method - req.path = path - # URI includes query string + # URI is the full path+query; path is just the path part (no query string). + # Twisted's Request.path was always without query string. req.uri = path + from urllib.parse import parse_qs, urlparse + path_str = path.decode("utf-8") if isinstance(path, bytes) else path + parsed = urlparse(path_str) + req.path = parsed.path.encode("utf-8") if isinstance(path, bytes) else parsed.path.encode("utf-8") + req.content = BytesIO(content) - req.content.seek(0, SEEK_END) + req.content.seek(0) req._client_ip = client_ip # 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 @@ -528,31 +548,55 @@ def make_request( # Initialize request metrics and logcontext before dispatch import asyncio from synapse.http.request_metrics import RequestMetrics - from synapse.logging.context import LoggingContext + from synapse.logging.context import ContextRequest, 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()) + + # Create a ContextRequest (NOT the SynapseRequest itself!) for the LoggingContext + context_request = ContextRequest( + request_id=req.get_request_id(), + ip_address=req.getClientIP(), + site_tag=getattr(site, 'site_tag', 'test'), + requester=None, + authenticated_entity=None, + method=req.get_method(), + url=req.get_redacted_uri(), + protocol="HTTP/1.1", + user_agent="", + ) req.logcontext = LoggingContext( name="test-%s-%s" % (req.get_method(), req.get_redacted_uri()), server_name=server_name, - request=req, + request=context_request, ) - # 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) + # Dispatch the request through the resource tree using the same + # logic as the production aiohttp handler. + from synapse.http.aiohttp_shim import _resolve_resource + + root_resource = getattr(site, 'resource', None) or getattr(site, '_resource', None) + if root_resource is not None: + target = _resolve_resource(root_resource, path) + + if hasattr(target, '_async_render_wrapper'): + req.render_deferred = asyncio.ensure_future( + target._async_render_wrapper(req) + ) else: - import sys - print(f"WARNING: No resource to dispatch to. site={type(site)}, resource={resource}", file=sys.stderr) + # Simple resource with render_GET/render_POST etc + method_str = req.method.decode('ascii') if isinstance(req.method, bytes) else req.method + method_name = 'render_' + method_str + handler = getattr(target, method_name, None) + if handler: + result = handler(req) + if asyncio.iscoroutine(result): + req.render_deferred = asyncio.ensure_future(result) + else: + from synapse.http.server import respond_with_json + respond_with_json(req, 404, {"errcode": "M_UNRECOGNIZED", "error": "Unrecognized request"}, send_cors=True) if await_result: channel.await_result() @@ -562,7 +606,7 @@ def make_request( # ISynapseReactor implies IReactorPluggableNameResolver, but explicitly # marking this as an implementer of the latter seems to keep mypy-zope happier. -@implementer(IReactorPluggableNameResolver, ISynapseReactor) +@implementer(IReactorPluggableNameResolver) class ThreadedMemoryReactorClock(MemoryReactorClock): """ A MemoryReactorClock that supports callFromThread. diff --git a/tests/unittest.py b/tests/unittest.py index 94af5a4ff8..a2edebd352 100644 --- a/tests/unittest.py +++ b/tests/unittest.py @@ -194,10 +194,12 @@ class TestCase(_stdlib_unittest.TestCase): @around(self) def setUp(orig: Callable[[], R]) -> R: - # Set up an asyncio event loop so asyncio primitives work + # Set up a fresh asyncio event loop for each test import asyncio as _asyncio + import nest_asyncio as _nest_asyncio self._asyncio_loop = _asyncio.new_event_loop() _asyncio.set_event_loop(self._asyncio_loop) + _nest_asyncio.apply(self._asyncio_loop) # if we're not starting in the sentinel logcontext, then to be honest # all future bets are off. @@ -240,6 +242,17 @@ class TestCase(_stdlib_unittest.TestCase): @around(self) def tearDown(orig: Callable[[], R]) -> R: ret = orig() + + # Cancel any remaining asyncio tasks from this test + import asyncio as _asyncio + loop = _asyncio.get_event_loop() + if not loop.is_closed(): + pending = [t for t in _asyncio.all_tasks(loop) if not t.done()] + for t in pending: + t.cancel() + if pending: + loop.run_until_complete(_asyncio.sleep(0)) + gc.collect(0) # Run a full GC every 50 gen-0 GCs. gen0_stats = gc.get_stats()[0]