Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Patch the db conn pool sooner in tests #17017

Merged
merged 3 commits into from
Mar 21, 2024
Merged
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Next Next commit
Patch the db conn pool sooner in tests
When running unit tests, we patch the database connection pool so that it runs
queries "synchronously". This is ok, except that if any queries are launched
before we do the patching, those queries get left in limbo and never complete.

To fix this, let's change the way we do the switcheroo, by patching out the
method which creates the connection pool in the first place.
richvdh committed Mar 21, 2024

Verified

This commit was signed with the committer’s verified signature. The key has expired.
zalegrala Zach Leslie
commit 7461649a64b68e0631b7d135b04c6738dc29eaef
113 changes: 60 additions & 53 deletions tests/server.py
Original file line number Diff line number Diff line change
@@ -47,14 +47,15 @@
Union,
cast,
)
from unittest.mock import Mock
from unittest.mock import Mock, patch

import attr
from incremental import Version
from typing_extensions import ParamSpec
from zope.interface import implementer

import twisted
from twisted.enterprise import adbapi
from twisted.internet import address, tcp, threads, udp
from twisted.internet._resolver import SimpleResolverComplexifier
from twisted.internet.defer import Deferred, fail, maybeDeferred, succeed
@@ -69,6 +70,7 @@
IProtocol,
IPullProducer,
IPushProducer,
IReactorCore,
IReactorPluggableNameResolver,
IReactorTime,
IResolverSimple,
@@ -94,8 +96,8 @@
)
from synapse.server import HomeServer
from synapse.storage import DataStore
from synapse.storage.database import LoggingDatabaseConnection
from synapse.storage.engines import create_engine
from synapse.storage.database import LoggingDatabaseConnection, make_pool
from synapse.storage.engines import BaseDatabaseEngine, create_engine
from synapse.storage.prepare_database import prepare_database
from synapse.types import ISynapseReactor, JsonDict
from synapse.util import Clock
@@ -670,6 +672,53 @@ def validate_connector(connector: tcp.Connector, expected_ip: str) -> None:
)


def make_fake_db_pool(
reactor: IReactorCore,
db_config: DatabaseConnectionConfig,
engine: BaseDatabaseEngine,
) -> adbapi.ConnectionPool:
"""Wrapper for `make_pool` which builds a pool which runs db queries synchronously.

For more deterministic testing, we don't use a regular db connection pool: instead
we run all db queries synchronously on the test reactor's main thread. This function
is a drop-in replacement for the normal `make_pool` which builds such a connection
pool.
"""
pool = make_pool(reactor, db_config, engine)

def runWithConnection(
func: Callable[..., R], *args: Any, **kwargs: Any
) -> Awaitable[R]:
return threads.deferToThreadPool(
pool._reactor,
pool.threadpool,
pool._runWithConnection,
func,
*args,
**kwargs,
)

def runInteraction(
desc: str, func: Callable[..., R], *args: Any, **kwargs: Any
) -> Awaitable[R]:
return threads.deferToThreadPool(
pool._reactor,
pool.threadpool,
pool._runInteraction,
desc,
func,
*args,
**kwargs,
)

pool.runWithConnection = runWithConnection # type: ignore[method-assign]
pool.runInteraction = runInteraction # type: ignore[assignment]
# Replace the thread pool with a threadless 'thread' pool
pool.threadpool = ThreadPool(reactor)
pool.running = True
return pool


class ThreadPool:
"""
Threadless thread pool.
@@ -706,52 +755,6 @@ def _(res: Any) -> None:
return d


def _make_test_homeserver_synchronous(server: HomeServer) -> None:
"""
Make the given test homeserver's database interactions synchronous.
"""

clock = server.get_clock()

for database in server.get_datastores().databases:
pool = database._db_pool

def runWithConnection(
func: Callable[..., R], *args: Any, **kwargs: Any
) -> Awaitable[R]:
return threads.deferToThreadPool(
pool._reactor,
pool.threadpool,
pool._runWithConnection,
func,
*args,
**kwargs,
)

def runInteraction(
desc: str, func: Callable[..., R], *args: Any, **kwargs: Any
) -> Awaitable[R]:
return threads.deferToThreadPool(
pool._reactor,
pool.threadpool,
pool._runInteraction,
desc,
func,
*args,
**kwargs,
)

pool.runWithConnection = runWithConnection # type: ignore[method-assign]
pool.runInteraction = runInteraction # type: ignore[assignment]
# Replace the thread pool with a threadless 'thread' pool
pool.threadpool = ThreadPool(clock._reactor)
pool.running = True

# We've just changed the Databases to run DB transactions on the same
# thread, so we need to disable the dedicated thread behaviour.
server.get_datastores().main.USE_DEDICATED_DB_THREADS_FOR_EVENT_FETCHING = False


def get_clock() -> Tuple[ThreadedMemoryReactorClock, Clock]:
clock = ThreadedMemoryReactorClock()
hs_clock = Clock(clock)
@@ -1067,7 +1070,14 @@ def setup_test_homeserver(
# Mock TLS
hs.tls_server_context_factory = Mock()

hs.setup()
# Patch `make_pool` before initialising the database, to make database transactions
# synchronous for testing.
with patch("synapse.storage.database.make_pool", side_effect=make_fake_db_pool):
hs.setup()

# Since we've changed the databases to run DB transactions on the same
# thread, we need to stop the event fetcher hogging that one thread.
hs.get_datastores().main.USE_DEDICATED_DB_THREADS_FOR_EVENT_FETCHING = False

if USE_POSTGRES_FOR_TESTS:
database_pool = hs.get_datastores().databases[0]
@@ -1137,9 +1147,6 @@ async def validate_hash(p: str, h: str) -> bool:

hs.get_auth_handler().validate_hash = validate_hash # type: ignore[assignment]

# Make the threadpool and database transactions synchronous for testing.
_make_test_homeserver_synchronous(hs)

# Load any configured modules into the homeserver
module_api = hs.get_module_api()
for module, module_config in hs.config.modules.loaded_modules: