From 8bb5376e2d2b202bb2bbff54aeea34612af7bc9d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Edgars=20Nem=C5=A1e?= Date: Mon, 14 Sep 2026 18:09:39 +0100 Subject: [PATCH 1/2] fix(rpc): prevent database pool waits from stalling Studio Run synchronous RPC handlers and public async read-path database checkouts outside the API event loop. Preserve worker-originated websocket notifications on the application loop. Bound Explorer queries to two per process, close read sessions inside the query worker, and coalesce count refreshes with a five-second cache. Overload returns 503 with Retry-After without opening another session. Hotfix for the v0.121 production line; no schema, dependency, pool-size, or runtime-version changes. Validated with 859 backend unit tests, 196 frontend unit tests, frontend type checks, and repository formatting/lint checks. Full stable-track E2E remains a landing gate. --- backend/protocol_rpc/app_lifespan.py | 2 + backend/protocol_rpc/endpoints.py | 17 +- backend/protocol_rpc/explorer/query_runner.py | 69 +++++ backend/protocol_rpc/explorer/router.py | 57 ++-- .../message_handler/fastapi_handler.py | 26 +- backend/protocol_rpc/rpc_endpoint_manager.py | 13 +- tests/unit/test_explorer_query_runner.py | 262 ++++++++++++++++++ tests/unit/test_rpc_worker_execution.py | 217 +++++++++++++++ 8 files changed, 631 insertions(+), 32 deletions(-) create mode 100644 backend/protocol_rpc/explorer/query_runner.py create mode 100644 tests/unit/test_explorer_query_runner.py create mode 100644 tests/unit/test_rpc_worker_execution.py diff --git a/backend/protocol_rpc/app_lifespan.py b/backend/protocol_rpc/app_lifespan.py index 08504c228..11ff8a6de 100644 --- a/backend/protocol_rpc/app_lifespan.py +++ b/backend/protocol_rpc/app_lifespan.py @@ -26,6 +26,7 @@ ) from backend.protocol_rpc.transactions_parser import TransactionParser from backend.protocol_rpc.configuration import GlobalConfiguration +from backend.protocol_rpc.explorer.query_runner import ExplorerQueryRunner from backend.protocol_rpc.fastapi_rpc_router import FastAPIRPCRouter from backend.protocol_rpc.message_handler.fastapi_handler import ( MessageHandler, @@ -234,6 +235,7 @@ async def rpc_app_lifespan(app, settings: RPCAppSettings) -> AsyncIterator[RPCAp ) db_manager = DatabaseSessionManager(settings.database_url) set_database_manager(db_manager) + app.state.explorer_query_runner = ExplorerQueryRunner(db_manager) logger.info("[STARTUP] Verifying database readiness and migrations") _verify_database_ready(db_manager) diff --git a/backend/protocol_rpc/endpoints.py b/backend/protocol_rpc/endpoints.py index a6254e337..fa4ed93ff 100644 --- a/backend/protocol_rpc/endpoints.py +++ b/backend/protocol_rpc/endpoints.py @@ -70,6 +70,7 @@ from backend.database_handler.snapshot_manager import SnapshotManager from backend.node.base import Manager as GenVMManager import asyncio +from starlette.concurrency import run_in_threadpool # Limit concurrent GenVM executions on the jsonrpc path to prevent uvloop fd # conflicts and DB pool exhaustion while calls hold request-scoped sessions. @@ -1076,7 +1077,9 @@ async def get_contract_schema( contract_address: str, ) -> dict: try: - contract_snapshot = ContractSnapshot(contract_address, session) + contract_snapshot = await run_in_threadpool( + ContractSnapshot, contract_address, session + ) except ContractNotFoundError: raise NotFoundError( message=f"Contract {contract_address} not found", @@ -1424,7 +1427,9 @@ async def _gen_call_with_validator( # Create validator node try: - contract_snapshot = ContractSnapshot(to_address, session) + contract_snapshot = await run_in_threadpool( + ContractSnapshot, to_address, session + ) except ContractNotFoundError: raise NotFoundError( message=f"Contract {to_address} not found", @@ -1630,8 +1635,8 @@ async def eth_call( # Check if this is a ConsensusData contract call that we should handle locally # This should happen before early return to allow interception even without 'from' - consensus_data_result = handle_consensus_data_call( - transactions_processor, to_address, data + consensus_data_result = await run_in_threadpool( + handle_consensus_data_call, transactions_processor, to_address, data ) if consensus_data_result is not None: return consensus_data_result @@ -1657,7 +1662,9 @@ async def eth_call( ) as_validator = snapshot.nodes[0].validator try: - target_contract_snapshot = ContractSnapshot(to_address, session) + target_contract_snapshot = await run_in_threadpool( + ContractSnapshot, to_address, session + ) except ContractNotFoundError: raise NotFoundError( message=f"Contract {to_address} not found", diff --git a/backend/protocol_rpc/explorer/query_runner.py b/backend/protocol_rpc/explorer/query_runner.py new file mode 100644 index 000000000..d7eb1d9fe --- /dev/null +++ b/backend/protocol_rpc/explorer/query_runner.py @@ -0,0 +1,69 @@ +"""Bound Explorer reads and close their sessions before returning to the API loop.""" + +import threading +import time +from typing import Any, Callable + +from fastapi import HTTPException + +from backend.database_handler.session_factory import DatabaseSessionManager + +from . import queries + + +class ExplorerQueryRunner: + def __init__( + self, + db_manager: DatabaseSessionManager, + *, + max_concurrent: int = 2, + counts_ttl: float = 5.0, + ) -> None: + self._db_manager = db_manager + self._slots = threading.BoundedSemaphore(max_concurrent) + self._counts_ttl = counts_ttl + self._counts_lock = threading.Lock() + self._cached_counts: tuple[float, dict] | None = None + + @staticmethod + def _busy() -> HTTPException: + return HTTPException( + status_code=503, + detail="Explorer busy; retry shortly", + headers={"Retry-After": "1"}, + ) + + def run(self, query: Callable[..., Any], *args: Any, **kwargs: Any) -> Any: + if not self._slots.acquire(blocking=False): + raise self._busy() + try: + # Query functions materialize their results. Closing here rolls back + # the read transaction in the same worker, without an event-loop hop. + with self._db_manager.open_session() as session: + return query(session, *args, **kwargs) + finally: + self._slots.release() + + def counts(self) -> dict: + cached = self._cached_counts + if cached is not None and cached[0] > time.monotonic(): + return dict(cached[1]) + + # Coalesce refreshes across Explorer server renders. A concurrent caller + # can use the previous counts while the single refresh is in progress. + if not self._counts_lock.acquire(blocking=False): + if cached is not None: + return dict(cached[1]) + raise self._busy() + try: + cached = self._cached_counts + if cached is not None and cached[0] > time.monotonic(): + return dict(cached[1]) + counts = self.run(queries.get_stats_counts) + self._cached_counts = ( + time.monotonic() + self._counts_ttl, + dict(counts), + ) + return counts + finally: + self._counts_lock.release() diff --git a/backend/protocol_rpc/explorer/router.py b/backend/protocol_rpc/explorer/router.py index 3fc369fa4..e24599397 100644 --- a/backend/protocol_rpc/explorer/router.py +++ b/backend/protocol_rpc/explorer/router.py @@ -2,29 +2,37 @@ from typing import Annotated, Literal, Optional -from fastapi import APIRouter, Depends, HTTPException, Query -from sqlalchemy.orm import Session - -from backend.protocol_rpc.dependencies import get_db_session +from fastapi import APIRouter, Depends, HTTPException, Query, Request from . import queries +from .query_runner import ExplorerQueryRunner explorer_router = APIRouter(prefix="/api/explorer", tags=["explorer"]) +def get_query_runner(request: Request) -> ExplorerQueryRunner: + runner = getattr(request.app.state, "explorer_query_runner", None) + if runner is None: + raise HTTPException(status_code=503, detail="Explorer not initialized") + return runner + + +QueryRunner = Annotated[ExplorerQueryRunner, Depends(get_query_runner)] + + # --------------------------------------------------------------------------- # Stats # --------------------------------------------------------------------------- @explorer_router.get("/stats") -def get_stats(session: Annotated[Session, Depends(get_db_session)]): - return queries.get_stats(session) +def get_stats(runner: QueryRunner): + return runner.run(queries.get_stats) @explorer_router.get("/stats/counts") -def get_stats_counts(session: Annotated[Session, Depends(get_db_session)]): - return queries.get_stats_counts(session) +def get_stats_counts(runner: QueryRunner): + return runner.counts() # --------------------------------------------------------------------------- @@ -34,7 +42,7 @@ def get_stats_counts(session: Annotated[Session, Depends(get_db_session)]): @explorer_router.get("/transactions") def get_transactions( - session: Annotated[Session, Depends(get_db_session)], + runner: QueryRunner, page: int = Query(1, ge=1), limit: int = Query(20, ge=1, le=100), status: Optional[str] = None, @@ -43,17 +51,24 @@ def get_transactions( to_date: Optional[str] = None, address: Optional[str] = None, ): - return queries.get_all_transactions_paginated( - session, page, limit, status, search, from_date, to_date, address + return runner.run( + queries.get_all_transactions_paginated, + page, + limit, + status, + search, + from_date, + to_date, + address, ) @explorer_router.get("/transactions/{tx_hash}") def get_transaction( tx_hash: str, - session: Annotated[Session, Depends(get_db_session)], + runner: QueryRunner, ): - result = queries.get_transaction_with_relations(session, tx_hash) + result = runner.run(queries.get_transaction_with_relations, tx_hash) if result is None: raise HTTPException(status_code=404, detail="Transaction not found") return result @@ -66,11 +81,11 @@ def get_transaction( @explorer_router.get("/validators") def get_validators( - session: Annotated[Session, Depends(get_db_session)], + runner: QueryRunner, search: Optional[str] = None, limit: Optional[int] = Query(None, ge=1, le=100), ): - return queries.get_all_validators(session, search=search, limit=limit) + return runner.run(queries.get_all_validators, search=search, limit=limit) # --------------------------------------------------------------------------- @@ -81,9 +96,9 @@ def get_validators( @explorer_router.get("/address/{address}") def get_address( address: str, - session: Annotated[Session, Depends(get_db_session)], + runner: QueryRunner, ): - result = queries.get_address_info(session, address) + result = runner.run(queries.get_address_info, address) if result is None: raise HTTPException(status_code=404, detail="Address not found") return result @@ -96,14 +111,14 @@ def get_address( @explorer_router.get("/contracts") def get_contracts( - session: Annotated[Session, Depends(get_db_session)], + runner: QueryRunner, search: Optional[str] = None, page: int = Query(1, ge=1), limit: int = Query(20, ge=1, le=100), sort_by: Optional[Literal["tx_count", "created_at", "updated_at"]] = None, sort_order: Literal["asc", "desc"] = "desc", ): - return queries.get_all_states(session, search, page, limit, sort_by, sort_order) + return runner.run(queries.get_all_states, search, page, limit, sort_by, sort_order) # --------------------------------------------------------------------------- @@ -112,5 +127,5 @@ def get_contracts( @explorer_router.get("/providers") -def get_providers(session: Annotated[Session, Depends(get_db_session)]): - return queries.get_all_providers(session) +def get_providers(runner: QueryRunner): + return runner.run(queries.get_all_providers) diff --git a/backend/protocol_rpc/message_handler/fastapi_handler.py b/backend/protocol_rpc/message_handler/fastapi_handler.py index 5e0af271d..ab3d63ee0 100644 --- a/backend/protocol_rpc/message_handler/fastapi_handler.py +++ b/backend/protocol_rpc/message_handler/fastapi_handler.py @@ -30,10 +30,15 @@ def __init__(self, broadcast: Broadcast, config: GlobalConfiguration): self.broadcast = broadcast self.config = config self.client_session_id = None + try: + self._loop = asyncio.get_running_loop() + except RuntimeError: + self._loop = None def with_client_session(self, client_session_id: str): new_msg_handler = MessageHandler(self.broadcast, self.config) new_msg_handler.client_session_id = client_session_id + new_msg_handler._loop = self._loop return new_msg_handler def log_endpoint_info(self, func): @@ -76,14 +81,27 @@ def _publish(self, channel: str, payload: dict[str, Any]) -> None: message = json.dumps(payload) try: - loop = asyncio.get_running_loop() + running_loop = asyncio.get_running_loop() except RuntimeError: - return + running_loop = None - if not loop.is_running(): + loop = self._loop or running_loop + if loop is None or not loop.is_running(): return + self._loop = loop - loop.create_task(self.broadcast.publish(channel=channel, message=message)) + def publish(): + loop.create_task(self.broadcast.publish(channel=channel, message=message)) + + if running_loop is loop: + publish() + else: + # Sync RPC handlers run in workers; Broadcast belongs to the app loop. + try: + loop.call_soon_threadsafe(publish) + except RuntimeError: + # The application may have shut down while a worker finished. + pass def _socket_emit(self, log_event: LogEvent) -> None: """Emit a log event via broadcast channels. diff --git a/backend/protocol_rpc/rpc_endpoint_manager.py b/backend/protocol_rpc/rpc_endpoint_manager.py index 502fa79f8..aaa2dace5 100644 --- a/backend/protocol_rpc/rpc_endpoint_manager.py +++ b/backend/protocol_rpc/rpc_endpoint_manager.py @@ -12,6 +12,7 @@ from fastapi.dependencies.utils import get_dependant, solve_dependencies from fastapi.requests import Request from pydantic import BaseModel, ConfigDict +from starlette.concurrency import run_in_threadpool from backend.protocol_rpc.exceptions import ( InternalError, @@ -305,7 +306,7 @@ async def _call_endpoint( call_kwargs = bound_arguments if "msg_handler" in call_kwargs: call_kwargs["msg_handler"] = session_logger - result = registered.dependant.call(**call_kwargs) + result = await self._invoke_handler(registered.dependant.call, call_kwargs) if inspect.isawaitable(result): result = await result return result @@ -366,11 +367,19 @@ async def _call_endpoint( if "msg_handler" in call_kwargs: call_kwargs["msg_handler"] = session_logger - result = registered.dependant.call(**call_kwargs) + result = await self._invoke_handler(registered.dependant.call, call_kwargs) if inspect.isawaitable(result): result = await result return result + @staticmethod + async def _invoke_handler(handler: Any, kwargs: Dict[str, Any]) -> Any: + # A synchronous pool checkout must not block the loop that schedules + # other requests' session cleanup (and therefore returns connections). + if inspect.iscoroutinefunction(handler): + return handler(**kwargs) + return await run_in_threadpool(handler, **kwargs) + def _bind_rpc_arguments( self, registered: RegisteredEndpoint, diff --git a/tests/unit/test_explorer_query_runner.py b/tests/unit/test_explorer_query_runner.py new file mode 100644 index 000000000..ee5a72fcf --- /dev/null +++ b/tests/unit/test_explorer_query_runner.py @@ -0,0 +1,262 @@ +from contextlib import contextmanager +from unittest.mock import Mock + +import pytest +from fastapi import FastAPI, HTTPException +from fastapi.testclient import TestClient +from sqlalchemy import text +from sqlalchemy.pool import QueuePool + +from backend.database_handler.session_factory import DatabaseSessionManager +from backend.protocol_rpc.explorer import queries, query_runner +from backend.protocol_rpc.explorer.query_runner import ExplorerQueryRunner +from backend.protocol_rpc.explorer.router import explorer_router + + +class DatabaseStub: + def __init__(self): + self.opened = 0 + self.closed = 0 + self.session = object() + + @contextmanager + def open_session(self): + self.opened += 1 + try: + yield self.session + finally: + self.closed += 1 + + +@pytest.mark.parametrize("fails", [False, True]) +def test_real_session_returns_connection_and_rolls_back(fails): + db = DatabaseSessionManager("sqlite://", poolclass=QueuePool) + runner = ExplorerQueryRunner(db) + try: + with db.engine.begin() as connection: + connection.execute(text("CREATE TABLE example (value INTEGER)")) + + def query(session): + session.execute(text("INSERT INTO example VALUES (1)")) + if fails: + raise ValueError("query failed") + return session.execute(text("SELECT count(*) FROM example")).scalar() + + if fails: + with pytest.raises(ValueError, match="query failed"): + runner.run(query) + else: + assert runner.run(query) == 1 + + assert db.engine.pool.checkedout() == 0 + # Explorer does not commit writes; ending the read session rolls back. + assert ( + runner.run( + lambda session: session.execute( + text("SELECT count(*) FROM example") + ).scalar() + ) + == 0 + ) + assert db.engine.pool.checkedout() == 0 + finally: + db.engine.dispose() + + +@pytest.mark.parametrize("fails", [False, True]) +def test_session_closes_before_return_and_admission_is_reusable(fails): + db = DatabaseStub() + runner = ExplorerQueryRunner(db, max_concurrent=1) + + def query(session, value, *, option): + assert session is db.session + assert db.closed == 0 + if fails: + raise ValueError("query failed") + return {"value": value, "option": option} + + if fails: + with pytest.raises(ValueError, match="query failed"): + runner.run(query, 1, option=2) + else: + assert runner.run(query, 1, option=2) == {"value": 1, "option": 2} + + assert db.opened == db.closed == 1 + assert runner.run(lambda session: "next") == "next" + assert db.opened == db.closed == 2 + + +def test_admission_rejects_before_opening_session(): + db = DatabaseStub() + runner = ExplorerQueryRunner(db, max_concurrent=2) + + def second_query(session): + with pytest.raises(HTTPException) as exc: + runner.run(lambda session: pytest.fail("must not run")) + assert exc.value.status_code == 503 + assert exc.value.headers == {"Retry-After": "1"} + assert db.opened == 2 + + runner.run(lambda session: runner.run(second_query)) + assert db.opened == db.closed == 2 + assert runner.run(lambda session: "next") == "next" + + +def test_admission_released_when_open_session_fails(): + db = Mock() + db.open_session.side_effect = RuntimeError("unavailable") + runner = ExplorerQueryRunner(db, max_concurrent=1) + for _ in range(2): + with pytest.raises(RuntimeError, match="unavailable"): + runner.run(lambda session: None) + assert db.open_session.call_count == 2 + + +def test_counts_cache_expires_and_does_not_share_mutable_results(monkeypatch): + db = DatabaseStub() + runner = ExplorerQueryRunner(db, counts_ttl=5) + now = [100.0] + monkeypatch.setattr(query_runner.time, "monotonic", lambda: now[0]) + count_query = Mock(side_effect=[{"transactions": 10}, {"transactions": 11}]) + monkeypatch.setattr(queries, "get_stats_counts", count_query) + + first = runner.counts() + first["transactions"] = -1 + assert runner.counts() == {"transactions": 10} + cached = runner.counts() + cached["transactions"] = -2 + assert runner.counts() == {"transactions": 10} + assert db.opened == db.closed == 1 + + now[0] = 105.0 + assert runner.counts() == {"transactions": 11} + assert db.opened == db.closed == 2 + + +@pytest.mark.parametrize("has_cached_counts", [False, True]) +def test_counts_refresh_is_coalesced(monkeypatch, has_cached_counts): + db = DatabaseStub() + runner = ExplorerQueryRunner(db, counts_ttl=0) + if has_cached_counts: + monkeypatch.setattr(queries, "get_stats_counts", lambda session: {"count": 1}) + assert runner.counts() == {"count": 1} + + def refresh(session): + opened = db.opened + if has_cached_counts: + assert runner.counts() == {"count": 1} + else: + with pytest.raises(HTTPException) as exc: + runner.counts() + assert exc.value.status_code == 503 + assert db.opened == opened + return {"count": 2} + + monkeypatch.setattr(queries, "get_stats_counts", refresh) + assert runner.counts() == {"count": 2} + assert db.opened == db.closed == (2 if has_cached_counts else 1) + + +def test_failed_counts_refresh_can_retry(monkeypatch): + db = DatabaseStub() + runner = ExplorerQueryRunner(db) + monkeypatch.setattr( + queries, + "get_stats_counts", + Mock(side_effect=[ValueError("failed"), {"count": 2}]), + ) + with pytest.raises(ValueError, match="failed"): + runner.counts() + assert runner.counts() == {"count": 2} + assert db.opened == db.closed == 2 + + +@pytest.mark.parametrize( + "path, query_name, args, kwargs", + [ + ("/stats", "get_stats", (), {}), + ("/stats/counts", "get_stats_counts", (), {}), + ( + "/transactions?page=2&limit=10&status=PENDING&search=x&from_date=a&to_date=b&address=c", + "get_all_transactions_paginated", + (2, 10, "PENDING", "x", "a", "b", "c"), + {}, + ), + ("/transactions/0xabc", "get_transaction_with_relations", ("0xabc",), {}), + ( + "/validators?search=x&limit=4", + "get_all_validators", + (), + {"search": "x", "limit": 4}, + ), + ("/address/0xabc", "get_address_info", ("0xabc",), {}), + ( + "/contracts?search=x&page=2&limit=4&sort_by=tx_count&sort_order=asc", + "get_all_states", + ("x", 2, 4, "tx_count", "asc"), + {}, + ), + ("/providers", "get_all_providers", (), {}), + ], +) +def test_routes_use_bounded_worker_sessions( + monkeypatch, path, query_name, args, kwargs +): + import threading + + db = DatabaseStub() + app = FastAPI() + app.state.explorer_query_runner = ExplorerQueryRunner(db) + app.include_router(explorer_router) + loop_threads = [] + query_threads = [] + + @app.middleware("http") + async def note_loop(request, call_next): + loop_threads.append(threading.get_ident()) + response = await call_next(request) + assert db.closed == 1 + return response + + def query(session, *query_args, **query_kwargs): + query_threads.append(threading.get_ident()) + assert session is db.session + assert query_args == args + assert query_kwargs == kwargs + return {"result": "ok"} + + monkeypatch.setattr(queries, query_name, query) + with TestClient(app) as client: + response = client.get("/api/explorer" + path) + assert response.status_code == 200 + assert response.json() == {"result": "ok"} + assert query_threads[0] != loop_threads[0] + + +def test_routes_preserve_not_found_validation_and_busy_responses(monkeypatch): + db = DatabaseStub() + app = FastAPI() + runner = ExplorerQueryRunner(db, max_concurrent=1) + app.state.explorer_query_runner = runner + app.include_router(explorer_router) + monkeypatch.setattr(queries, "get_transaction_with_relations", lambda *args: None) + monkeypatch.setattr(queries, "get_address_info", lambda *args: None) + with TestClient(app) as client: + for path in ("/transactions/missing", "/address/missing"): + assert client.get("/api/explorer" + path).status_code == 404 + assert db.opened == db.closed == 2 + for path in ("/transactions?limit=101", "/contracts?sort_by=invalid"): + assert client.get("/api/explorer" + path).status_code == 422 + assert db.opened == 2 + with runner._slots: + response = client.get("/api/explorer/stats") + assert response.status_code == 503 + assert response.headers["Retry-After"] == "1" + assert db.opened == 2 + + +def test_uninitialized_explorer_returns_503(): + app = FastAPI() + app.include_router(explorer_router) + with TestClient(app) as client: + assert client.get("/api/explorer/stats").status_code == 503 diff --git a/tests/unit/test_rpc_worker_execution.py b/tests/unit/test_rpc_worker_execution.py new file mode 100644 index 000000000..e2318e89b --- /dev/null +++ b/tests/unit/test_rpc_worker_execution.py @@ -0,0 +1,217 @@ +import asyncio +import contextvars +import json +import threading +from unittest.mock import MagicMock + +import pytest +from fastapi import Depends, FastAPI +from starlette.concurrency import run_in_threadpool +from starlette.requests import Request + +from backend.protocol_rpc import endpoints +from backend.protocol_rpc.dependencies import get_db_session +from backend.protocol_rpc.exceptions import JSONRPCError, NotFoundError +from backend.protocol_rpc.message_handler.fastapi_handler import MessageHandler +from backend.protocol_rpc.rpc_endpoint_manager import ( + JSONRPCRequest, + RPCEndpointDefinition, + RPCEndpointManager, +) + + +def make_request(app): + return Request( + { + "type": "http", + "method": "POST", + "headers": [], + "app": app, + "query_string": b"", + "path": "/api", + "root_path": "", + "scheme": "http", + } + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("with_dependency", [False, True]) +@pytest.mark.parametrize("fails", [False, True]) +async def test_sync_rpc_runs_off_loop_and_cleans_up_dependencies( + with_dependency, fails +): + app = FastAPI() + loop = asyncio.get_running_loop() + loop_thread = threading.get_ident() + request_context = contextvars.ContextVar("test_request_context") + token = request_context.set("client-session") + session = MagicMock() + app.state.db_manager = MagicMock() + app.state.db_manager.open_session.return_value = session + + def work(): + assert threading.get_ident() != loop_thread + assert request_context.get() == "client-session" + # A callback scheduled by the worker must run while it is still busy. + callback_ran = threading.Event() + loop.call_soon_threadsafe(callback_ran.set) + assert callback_ran.wait(timeout=2) + if fails: + raise JSONRPCError(code=123, message="query failed") + return 42 + + def handler_with_dependency(db=Depends(get_db_session)): + assert db is session + return work() + + handler = handler_with_dependency if with_dependency else work + manager = RPCEndpointManager(MagicMock(), dependency_overrides_provider=app) + manager.register(RPCEndpointDefinition(name="test", handler=handler)) + try: + response = await manager.invoke( + JSONRPCRequest(method="test", id=1), make_request(app) + ) + finally: + request_context.reset(token) + if fails: + assert response.error == {"code": 123, "message": "query failed"} + else: + assert response.result == 42 + if with_dependency: + session.close.assert_called_once() + if fails: + session.rollback.assert_called_once() + session.commit.assert_not_called() + else: + session.commit.assert_called_once() + session.rollback.assert_not_called() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("sync_wrapper", [False, True]) +async def test_async_rpc_result_stays_on_application_loop(sync_wrapper): + app = FastAPI() + loop = asyncio.get_running_loop() + + async def async_handler(): + assert asyncio.get_running_loop() is loop + return 42 + + def wrapper(): + return async_handler() + + manager = RPCEndpointManager(MagicMock(), dependency_overrides_provider=app) + manager.register( + RPCEndpointDefinition( + name="test", handler=wrapper if sync_wrapper else async_handler + ) + ) + response = await manager.invoke( + JSONRPCRequest(method="test", id=1), make_request(app) + ) + assert response.result == 42 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("clone_in_worker", [False, True]) +async def test_worker_notifications_publish_on_owner_loop(clone_in_worker): + loop = asyncio.get_running_loop() + published = asyncio.Event() + messages = [] + + async def publish(*, channel, message): + assert asyncio.get_running_loop() is loop + messages.append((channel, json.loads(message))) + published.set() + + broadcast = MagicMock() + broadcast.publish = publish + handler = MessageHandler(broadcast, MagicMock()) + + def send(): + current = ( + handler.with_client_session("client-1") if clone_in_worker else handler + ) + current.send_transaction_status_update("0xabc", "CANCELED") + + await run_in_threadpool(send) + await asyncio.wait_for(published.wait(), timeout=2) + assert messages[0][0] == "0xabc" + assert messages[0][1]["event"] == "transaction_status_updated" + assert messages[0][1]["data"]["data"]["status"] == "CANCELED" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("endpoint", ["schema", "gen_call", "eth_call"]) +async def test_contract_snapshot_checkout_is_off_loop(monkeypatch, endpoint): + loop_thread = threading.get_ident() + address = "0x" + "ab" * 20 + session = MagicMock() + observed = [] + + def snapshot(contract_address, db_session): + observed.append(threading.get_ident()) + assert threading.get_ident() != loop_thread + assert db_session is session + raise endpoints.ContractNotFoundError(contract_address) + + monkeypatch.setattr(endpoints, "ContractSnapshot", snapshot) + monkeypatch.setattr(endpoints, "_check_rate_limit", lambda *args: None) + monkeypatch.setattr(endpoints, "_genvm_admission_semaphore", asyncio.Semaphore(1)) + monkeypatch.setattr(endpoints, "handle_consensus_data_call", lambda *args: None) + params = {"to": address, "from": address, "type": "read", "data": "0x1234"} + validator_snapshot = MagicMock() + validator_snapshot.nodes = [MagicMock()] + validators = MagicMock() + validators.snapshot.return_value.__aenter__.return_value = validator_snapshot + + with pytest.raises(NotFoundError): + if endpoint == "schema": + await endpoints.get_contract_schema( + session, MagicMock(), MagicMock(), address + ) + elif endpoint == "gen_call": + await endpoints._gen_call_with_validator( + session, + MagicMock(), + MagicMock(), + MagicMock(), + MagicMock(), + validator_snapshot, + params, + ) + else: + await endpoints.eth_call( + session, + MagicMock(), + MagicMock(), + MagicMock(), + validators, + MagicMock(), + MagicMock(), + params, + ) + assert len(observed) == 1 + + +@pytest.mark.asyncio +async def test_eth_consensus_data_query_is_off_loop(monkeypatch): + loop_thread = threading.get_ident() + + def intercept(*args): + assert threading.get_ident() != loop_thread + return "0x1234" + + monkeypatch.setattr(endpoints, "handle_consensus_data_call", intercept) + result = await endpoints.eth_call( + MagicMock(), + MagicMock(), + MagicMock(), + MagicMock(), + MagicMock(), + MagicMock(), + MagicMock(), + {"to": "0x" + "ab" * 20, "data": "0x1234"}, + ) + assert result == "0x1234" From e8208389451cfe9ba392b4726bb0b56d928f24a7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Edgars=20Nem=C5=A1e?= Date: Mon, 14 Sep 2026 18:19:26 +0100 Subject: [PATCH 2/2] test(explorer): remove optional HTTP client requirement Exercise Explorer routes directly through ASGI so regression tests run with the repository dependencies even when Starlette TestClient requires httpx2. No runtime dependency or production-code changes. Validated all 859 backend tests using both the existing local environment and an isolated overlay matching CI FastAPI 0.135.1, Starlette 1.6.0, AnyIO 4.15.1 and typing_extensions 4.16.0. --- tests/unit/test_explorer_query_runner.py | 96 ++++++++++++++++++------ 1 file changed, 72 insertions(+), 24 deletions(-) diff --git a/tests/unit/test_explorer_query_runner.py b/tests/unit/test_explorer_query_runner.py index ee5a72fcf..0b30b5c72 100644 --- a/tests/unit/test_explorer_query_runner.py +++ b/tests/unit/test_explorer_query_runner.py @@ -1,9 +1,10 @@ +import asyncio +import json from contextlib import contextmanager from unittest.mock import Mock import pytest from fastapi import FastAPI, HTTPException -from fastapi.testclient import TestClient from sqlalchemy import text from sqlalchemy.pool import QueuePool @@ -13,6 +14,53 @@ from backend.protocol_rpc.explorer.router import explorer_router +async def asgi_get(app, url): + """Exercise the app without an optional HTTP test-client dependency.""" + path, _, query = url.partition("?") + messages = [] + request_sent = False + response_done = asyncio.Event() + + async def receive(): + nonlocal request_sent + if not request_sent: + request_sent = True + return {"type": "http.request", "body": b"", "more_body": False} + await response_done.wait() + return {"type": "http.disconnect"} + + async def send(message): + messages.append(message) + if message["type"] == "http.response.body" and not message.get("more_body"): + response_done.set() + + scope = { + "type": "http", + "asgi": {"version": "3.0", "spec_version": "2.4"}, + "http_version": "1.1", + "method": "GET", + "scheme": "http", + "path": path, + "raw_path": path.encode(), + "query_string": query.encode(), + "root_path": "", + "headers": [], + "server": ("test", 80), + "client": ("test", 1234), + } + await asyncio.wait_for(app(scope, receive, send), timeout=5) + start = next( + message for message in messages if message["type"] == "http.response.start" + ) + body = b"".join( + message.get("body", b"") + for message in messages + if message["type"] == "http.response.body" + ) + headers = {key.decode(): value.decode() for key, value in start["headers"]} + return start["status"], headers, json.loads(body) + + class DatabaseStub: def __init__(self): self.opened = 0 @@ -171,6 +219,7 @@ def test_failed_counts_refresh_can_retry(monkeypatch): assert db.opened == db.closed == 2 +@pytest.mark.asyncio @pytest.mark.parametrize( "path, query_name, args, kwargs", [ @@ -199,7 +248,7 @@ def test_failed_counts_refresh_can_retry(monkeypatch): ("/providers", "get_all_providers", (), {}), ], ) -def test_routes_use_bounded_worker_sessions( +async def test_routes_use_bounded_worker_sessions( monkeypatch, path, query_name, args, kwargs ): import threading @@ -226,14 +275,14 @@ def query(session, *query_args, **query_kwargs): return {"result": "ok"} monkeypatch.setattr(queries, query_name, query) - with TestClient(app) as client: - response = client.get("/api/explorer" + path) - assert response.status_code == 200 - assert response.json() == {"result": "ok"} + status, _, body = await asgi_get(app, "/api/explorer" + path) + assert status == 200 + assert body == {"result": "ok"} assert query_threads[0] != loop_threads[0] -def test_routes_preserve_not_found_validation_and_busy_responses(monkeypatch): +@pytest.mark.asyncio +async def test_routes_preserve_not_found_validation_and_busy_responses(monkeypatch): db = DatabaseStub() app = FastAPI() runner = ExplorerQueryRunner(db, max_concurrent=1) @@ -241,22 +290,21 @@ def test_routes_preserve_not_found_validation_and_busy_responses(monkeypatch): app.include_router(explorer_router) monkeypatch.setattr(queries, "get_transaction_with_relations", lambda *args: None) monkeypatch.setattr(queries, "get_address_info", lambda *args: None) - with TestClient(app) as client: - for path in ("/transactions/missing", "/address/missing"): - assert client.get("/api/explorer" + path).status_code == 404 - assert db.opened == db.closed == 2 - for path in ("/transactions?limit=101", "/contracts?sort_by=invalid"): - assert client.get("/api/explorer" + path).status_code == 422 - assert db.opened == 2 - with runner._slots: - response = client.get("/api/explorer/stats") - assert response.status_code == 503 - assert response.headers["Retry-After"] == "1" - assert db.opened == 2 - - -def test_uninitialized_explorer_returns_503(): + for path in ("/transactions/missing", "/address/missing"): + assert (await asgi_get(app, "/api/explorer" + path))[0] == 404 + assert db.opened == db.closed == 2 + for path in ("/transactions?limit=101", "/contracts?sort_by=invalid"): + assert (await asgi_get(app, "/api/explorer" + path))[0] == 422 + assert db.opened == 2 + with runner._slots: + status, headers, _ = await asgi_get(app, "/api/explorer/stats") + assert status == 503 + assert headers["retry-after"] == "1" + assert db.opened == 2 + + +@pytest.mark.asyncio +async def test_uninitialized_explorer_returns_503(): app = FastAPI() app.include_router(explorer_router) - with TestClient(app) as client: - assert client.get("/api/explorer/stats").status_code == 503 + assert (await asgi_get(app, "/api/explorer/stats"))[0] == 503