mongodb/mongo-python-driver
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Revert "PYTHON-5536 Avoid clearing the connection pool when the server connec…"

Browse Source 
This reverts commit d267eb4833.
This commit is contained in:
Steven Silvester 2025-11-10 15:11:50 -06:00 committed by GitHub
parent 27785aede6
commit aeb6bcfe76
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
19 changed files with 63 additions and 345 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
.evergreen/run-tests.sh
View File

@ -26,9 +26,12 @@ else
fi
# List the packages.
uv sync ${UV_ARGS} --reinstall --quiet
uv sync ${UV_ARGS} --reinstall
uv pip list
# Ensure we go back to base environment after the test.
trap "uv sync" EXIT HUP
# Start the test runner.
uv run ${UV_ARGS} .evergreen/scripts/run_tests.py "$@"

37
justfile
View File

@ -1,11 +1,10 @@
# See https://just.systems/man/en/ for instructions
set shell := ["bash", "-c"]
# Do not modify the lock file when running justfile commands.
export UV_FROZEN := "1"
# Commonly used command segments.
typing_run := "uv run --group typing --extra aws --extra encryption --extra ocsp --extra snappy --extra test --extra zstd"
docs_run := "uv run --extra docs"
uv_run := "uv run --isolated --frozen "
typing_run := uv_run + "--group typing --extra aws --extra encryption --extra ocsp --extra snappy --extra test --extra zstd"
docs_run := uv_run + "--extra docs"
doc_build := "./doc/_build"
mypy_args := "--install-types --non-interactive"
@ -14,55 +13,51 @@ mypy_args := "--install-types --non-interactive"
default:
@just --list
[private]
resync:
@uv sync --quiet --frozen
install:
bash .evergreen/scripts/setup-dev-env.sh
[group('docs')]
docs: && resync
docs:
{{docs_run}} sphinx-build -W -b html doc {{doc_build}}/html
[group('docs')]
docs-serve: && resync
docs-serve:
{{docs_run}} sphinx-autobuild -W -b html doc --watch ./pymongo --watch ./bson --watch ./gridfs {{doc_build}}/serve
[group('docs')]
docs-linkcheck: && resync
docs-linkcheck:
{{docs_run}} sphinx-build -E -b linkcheck doc {{doc_build}}/linkcheck
[group('typing')]
typing: && resync
typing:
just typing-mypy
just typing-pyright
[group('typing')]
typing-mypy: && resync
typing-mypy:
{{typing_run}} mypy {{mypy_args}} bson gridfs tools pymongo
{{typing_run}} mypy {{mypy_args}} --config-file mypy_test.ini test
{{typing_run}} mypy {{mypy_args}} test/test_typing.py test/test_typing_strict.py
[group('typing')]
typing-pyright: && resync
typing-pyright:
{{typing_run}} pyright test/test_typing.py test/test_typing_strict.py
{{typing_run}} pyright -p strict_pyrightconfig.json test/test_typing_strict.py
[group('lint')]
lint: && resync
uv run pre-commit run --all-files
lint:
{{uv_run}} pre-commit run --all-files
[group('lint')]
lint-manual: && resync
uv run pre-commit run --all-files --hook-stage manual
lint-manual:
{{uv_run}} pre-commit run --all-files --hook-stage manual
[group('test')]
test *args="-v --durations=5 --maxfail=10": && resync
uv run --extra test pytest {{args}}
test *args="-v --durations=5 --maxfail=10":
{{uv_run}} --extra test pytest {{args}}
[group('test')]
run-tests *args: && resync
run-tests *args:
bash ./.evergreen/run-tests.sh {{args}}
[group('test')]

71
pymongo/asynchronous/pool.py
View File

@ -37,7 +37,7 @@ from typing import (
from bson import DEFAULT_CODEC_OPTIONS
from pymongo import _csot, helpers_shared
from pymongo.asynchronous.client_session import _validate_session_write_concern
from pymongo.asynchronous.helpers import _backoff, _handle_reauth
from pymongo.asynchronous.helpers import _handle_reauth
from pymongo.asynchronous.network import command
from pymongo.common import (
MAX_BSON_SIZE,
@ -788,9 +788,9 @@ class Pool:
# Enforces: maxConnecting
# Also used for: clearing the wait queue
self._max_connecting_cond = _async_create_condition(self.lock)
self._max_connecting = self.opts.max_connecting
self._pending = 0
self._client_id = client_id
self._backoff = 0
if self.enabled_for_cmap:
assert self.opts._event_listeners is not None
self.opts._event_listeners.publish_pool_created(
@ -846,8 +846,6 @@ class Pool:
async with self.size_cond:
if self.closed:
return
# Clear the backoff state.
self._backoff = 0
if self.opts.pause_enabled and pause and not self.opts.load_balanced:
old_state, self.state = self.state, PoolState.PAUSED
self.gen.inc(service_id)
@ -930,11 +928,6 @@ class Pool:
for conn in sockets:
await conn.close_conn(ConnectionClosedReason.STALE)
@property
def max_connecting(self) -> int:
"""The current max connecting limit for the pool."""
return 1 if self._backoff else self.opts.max_connecting
async def update_is_writable(self, is_writable: Optional[bool]) -> None:
"""Updates the is_writable attribute on all sockets currently in the
Pool.
@ -1001,7 +994,7 @@ class Pool:
async with self._max_connecting_cond:
# If maxConnecting connections are already being created
# by this pool then try again later instead of waiting.
if self._pending >= self.max_connecting:
if self._pending >= self._max_connecting:
return
self._pending += 1
incremented = True
@ -1029,30 +1022,6 @@ class Pool:
self.requests -= 1
self.size_cond.notify()
def _handle_connection_error(self, error: BaseException, phase: str, conn_id: int) -> None:
# Handle system overload condition for non-sdam pools.
# Look for an AutoReconnect error raised from a ConnectionResetError with
# errno == errno.ECONNRESET or raised from an OSError that we've created due to
# a closed connection.
# If found, set backoff and add error labels.
if self.is_sdam or type(error) != AutoReconnect:
return
self._backoff += 1
error._add_error_label("SystemOverloadedError")
error._add_error_label("RetryableError")
# Log the pool backoff message.
if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
_debug_log(
_CONNECTION_LOGGER,
message=_ConnectionStatusMessage.POOL_BACKOFF,
clientId=self._client_id,
serverHost=self.address[0],
serverPort=self.address[1],
driverConnectionId=conn_id,
reason=_verbose_connection_error_reason(ConnectionClosedReason.POOL_BACKOFF),
error=ConnectionClosedReason.POOL_BACKOFF,
)
async def connect(self, handler: Optional[_MongoClientErrorHandler] = None) -> AsyncConnection:
"""Connect to Mongo and return a new AsyncConnection.
@ -1082,17 +1051,8 @@ class Pool:
driverConnectionId=conn_id,
)
# Apply backoff if applicable.
if self._backoff:
await asyncio.sleep(_backoff(self._backoff))
# Pass a context to determine if we successfully create a configured socket.
context = dict(has_created_socket=False)
try:
networking_interface = await _configured_protocol_interface(
self.address, self.opts, context=context
)
networking_interface = await _configured_protocol_interface(self.address, self.opts)
# Catch KeyboardInterrupt, CancelledError, etc. and cleanup.
except BaseException as error:
async with self.lock:
@ -1113,11 +1073,10 @@ class Pool:
reason=_verbose_connection_error_reason(ConnectionClosedReason.ERROR),
error=ConnectionClosedReason.ERROR,
)
if context["has_created_socket"]:
self._handle_connection_error(error, "handshake", conn_id)
if isinstance(error, (IOError, OSError, *SSLErrors)):
details = _get_timeout_details(self.opts)
_raise_connection_failure(self.address, error, timeout_details=details)
raise
conn = AsyncConnection(networking_interface, self, self.address, conn_id, self.is_sdam) # type: ignore[arg-type]
@ -1135,18 +1094,15 @@ class Pool:
await conn.authenticate()
# Catch KeyboardInterrupt, CancelledError, etc. and cleanup.
except BaseException as e:
except BaseException:
async with self.lock:
self.active_contexts.discard(conn.cancel_context)
self._handle_connection_error(e, "hello", conn_id)
await conn.close_conn(ConnectionClosedReason.ERROR)
raise
if handler:
await handler.client._topology.receive_cluster_time(conn._cluster_time)
# Clear the backoff state.
self._backoff = 0
return conn
@contextlib.asynccontextmanager
@ -1323,12 +1279,12 @@ class Pool:
# to be checked back into the pool.
async with self._max_connecting_cond:
self._raise_if_not_ready(checkout_started_time, emit_event=False)
while not (self.conns or self._pending < self.max_connecting):
while not (self.conns or self._pending < self._max_connecting):
timeout = deadline - time.monotonic() if deadline else None
if not await _async_cond_wait(self._max_connecting_cond, timeout):
# Timed out, notify the next thread to ensure a
# timeout doesn't consume the condition.
if self.conns or self._pending < self.max_connecting:
if self.conns or self._pending < self._max_connecting:
self._max_connecting_cond.notify()
emitted_event = True
self._raise_wait_queue_timeout(checkout_started_time)
@ -1469,8 +1425,8 @@ class Pool:
:class:`~pymongo.errors.AutoReconnect` exceptions on server
hiccups, etc. We only check if the socket was closed by an external
error if it has been > 1 second since the socket was checked into the
pool, or we are in backoff mode, to keep performance reasonable -
we can't avoid AutoReconnects completely anyway.
pool, to keep performance reasonable - we can't avoid AutoReconnects
completely anyway.
"""
idle_time_seconds = conn.idle_time_seconds()
# If socket is idle, open a new one.
@ -1481,11 +1437,8 @@ class Pool:
await conn.close_conn(ConnectionClosedReason.IDLE)
return True
check_interval_seconds = self._check_interval_seconds
if self._backoff:
check_interval_seconds = 0
if check_interval_seconds is not None and (
check_interval_seconds == 0 or idle_time_seconds > check_interval_seconds
if self._check_interval_seconds is not None and (
self._check_interval_seconds == 0 or idle_time_seconds > self._check_interval_seconds
):
if conn.conn_closed():
await conn.close_conn(ConnectionClosedReason.ERROR)

4
pymongo/asynchronous/topology.py
View File

@ -890,9 +890,7 @@ class Topology:
# Clear the pool.
await server.reset(service_id)
elif isinstance(error, ConnectionFailure):
if isinstance(error, WaitQueueTimeoutError) or error.has_error_label(
"SystemOverloadedError"
):
if isinstance(error, WaitQueueTimeoutError):
return
# "Client MUST replace the server's description with type Unknown
# ... MUST NOT request an immediate check of the server."

2
pymongo/logger.py
View File

@ -42,7 +42,6 @@ class _ConnectionStatusMessage(str, enum.Enum):
POOL_READY = "Connection pool ready"
POOL_CLOSED = "Connection pool closed"
POOL_CLEARED = "Connection pool cleared"
POOL_BACKOFF = "Connection pool backoff"
CONN_CREATED = "Connection created"
CONN_READY = "Connection ready"
@ -89,7 +88,6 @@ _SDAM_LOGGER = logging.getLogger("pymongo.topology")
_VERBOSE_CONNECTION_ERROR_REASONS = {
ConnectionClosedReason.POOL_CLOSED: "Connection pool was closed",
ConnectionCheckOutFailedReason.POOL_CLOSED: "Connection pool was closed",
ConnectionClosedReason.POOL_BACKOFF: "Connection pool is in backoff",
ConnectionClosedReason.STALE: "Connection pool was stale",
ConnectionClosedReason.ERROR: "An error occurred while using the connection",
ConnectionCheckOutFailedReason.CONN_ERROR: "An error occurred while trying to establish a new connection",

3
pymongo/monitoring.py
View File

@ -934,9 +934,6 @@ class ConnectionClosedReason:
POOL_CLOSED = "poolClosed"
"""The pool was closed, making the connection no longer valid."""
POOL_BACKOFF = "poolBackoff"
"""The pool is in backoff mode."""
class ConnectionCheckOutFailedReason:
"""An enum that defines values for `reason` on a

11
pymongo/network_layer.py
View File

@ -256,7 +256,6 @@ class PyMongoBaseProtocol(Protocol):
self._timeout = timeout
self._closed = asyncio.get_running_loop().create_future()
self._connection_lost = False
self._closing_exception = None
def settimeout(self, timeout: float | None) -> None:
self._timeout = timeout
@ -270,11 +269,9 @@ class PyMongoBaseProtocol(Protocol):
self.transport.abort()
self._resolve_pending(exc)
self._connection_lost = True
self._closing_exception = exc # type:ignore[assignment]
def connection_lost(self, exc: Optional[Exception] = None) -> None:
self._resolve_pending(exc)
self._closing_exception = exc # type:ignore[assignment]
if not self._closed.done():
self._closed.set_result(None)
@ -338,11 +335,8 @@ class PyMongoProtocol(PyMongoBaseProtocol, BufferedProtocol):
if self._done_messages:
message = await self._done_messages.popleft()
else:
if self._closed.done():
if self._closing_exception:
raise self._closing_exception
else:
raise OSError("connection closed")
if self.transport and self.transport.is_closing():
raise OSError("connection is already closed")
read_waiter = asyncio.get_running_loop().create_future()
self._pending_messages.append(read_waiter)
try:
@ -480,7 +474,6 @@ class PyMongoProtocol(PyMongoBaseProtocol, BufferedProtocol):
else:
msg.set_exception(exc)
self._done_messages.append(msg)
self._pending_messages.clear()
class PyMongoKMSProtocol(PyMongoBaseProtocol):

11
pymongo/pool_shared.py
View File

@ -250,7 +250,6 @@ async def _configured_protocol_interface(
address: _Address,
options: PoolOptions,
protocol_kls: type[PyMongoBaseProtocol] = PyMongoProtocol,
context: dict[str, bool] | None = None,
) -> AsyncNetworkingInterface:
"""Given (host, port) and PoolOptions, return a configured AsyncNetworkingInterface.
@ -262,10 +261,6 @@ async def _configured_protocol_interface(
ssl_context = options._ssl_context
timeout = options.socket_timeout
# Signal that we have created the socket successfully.
if context:
context["has_created_socket"] = True
if ssl_context is None:
return AsyncNetworkingInterface(
await asyncio.get_running_loop().create_connection(
@ -379,7 +374,7 @@ def _create_connection(address: _Address, options: PoolOptions) -> socket.socket
def _configured_socket_interface(
address: _Address, options: PoolOptions, *args: Any, context: dict[str, bool] | None = None
address: _Address, options: PoolOptions, *args: Any
) -> NetworkingInterface:
"""Given (host, port) and PoolOptions, return a NetworkingInterface wrapping a configured socket.
@ -390,10 +385,6 @@ def _configured_socket_interface(
sock = _create_connection(address, options)
ssl_context = options._ssl_context
# Signal that we have created the socket successfully.
if context:
context["has_created_socket"] = True
if ssl_context is None:
sock.settimeout(options.socket_timeout)
return NetworkingInterface(sock)

71
pymongo/synchronous/pool.py
View File

@ -84,7 +84,7 @@ from pymongo.server_api import _add_to_command
from pymongo.server_type import SERVER_TYPE
from pymongo.socket_checker import SocketChecker
from pymongo.synchronous.client_session import _validate_session_write_concern
from pymongo.synchronous.helpers import _backoff, _handle_reauth
from pymongo.synchronous.helpers import _handle_reauth
from pymongo.synchronous.network import command
if TYPE_CHECKING:
@ -786,9 +786,9 @@ class Pool:
# Enforces: maxConnecting
# Also used for: clearing the wait queue
self._max_connecting_cond = _create_condition(self.lock)
self._max_connecting = self.opts.max_connecting
self._pending = 0
self._client_id = client_id
self._backoff = 0
if self.enabled_for_cmap:
assert self.opts._event_listeners is not None
self.opts._event_listeners.publish_pool_created(
@ -844,8 +844,6 @@ class Pool:
with self.size_cond:
if self.closed:
return
# Clear the backoff state.
self._backoff = 0
if self.opts.pause_enabled and pause and not self.opts.load_balanced:
old_state, self.state = self.state, PoolState.PAUSED
self.gen.inc(service_id)
@ -928,11 +926,6 @@ class Pool:
for conn in sockets:
conn.close_conn(ConnectionClosedReason.STALE)
@property
def max_connecting(self) -> int:
"""The current max connecting limit for the pool."""
return 1 if self._backoff else self.opts.max_connecting
def update_is_writable(self, is_writable: Optional[bool]) -> None:
"""Updates the is_writable attribute on all sockets currently in the
Pool.
@ -997,7 +990,7 @@ class Pool:
with self._max_connecting_cond:
# If maxConnecting connections are already being created
# by this pool then try again later instead of waiting.
if self._pending >= self.max_connecting:
if self._pending >= self._max_connecting:
return
self._pending += 1
incremented = True
@ -1025,30 +1018,6 @@ class Pool:
self.requests -= 1
self.size_cond.notify()
def _handle_connection_error(self, error: BaseException, phase: str, conn_id: int) -> None:
# Handle system overload condition for non-sdam pools.
# Look for an AutoReconnect error raised from a ConnectionResetError with
# errno == errno.ECONNRESET or raised from an OSError that we've created due to
# a closed connection.
# If found, set backoff and add error labels.
if self.is_sdam or type(error) != AutoReconnect:
return
self._backoff += 1
error._add_error_label("SystemOverloadedError")
error._add_error_label("RetryableError")
# Log the pool backoff message.
if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
_debug_log(
_CONNECTION_LOGGER,
message=_ConnectionStatusMessage.POOL_BACKOFF,
clientId=self._client_id,
serverHost=self.address[0],
serverPort=self.address[1],
driverConnectionId=conn_id,
reason=_verbose_connection_error_reason(ConnectionClosedReason.POOL_BACKOFF),
error=ConnectionClosedReason.POOL_BACKOFF,
)
def connect(self, handler: Optional[_MongoClientErrorHandler] = None) -> Connection:
"""Connect to Mongo and return a new Connection.
@ -1078,17 +1047,8 @@ class Pool:
driverConnectionId=conn_id,
)
# Apply backoff if applicable.
if self._backoff:
time.sleep(_backoff(self._backoff))
# Pass a context to determine if we successfully create a configured socket.
context = dict(has_created_socket=False)
try:
networking_interface = _configured_socket_interface(
self.address, self.opts, context=context
)
networking_interface = _configured_socket_interface(self.address, self.opts)
# Catch KeyboardInterrupt, CancelledError, etc. and cleanup.
except BaseException as error:
with self.lock:
@ -1109,11 +1069,10 @@ class Pool:
reason=_verbose_connection_error_reason(ConnectionClosedReason.ERROR),
error=ConnectionClosedReason.ERROR,
)
if context["has_created_socket"]:
self._handle_connection_error(error, "handshake", conn_id)
if isinstance(error, (IOError, OSError, *SSLErrors)):
details = _get_timeout_details(self.opts)
_raise_connection_failure(self.address, error, timeout_details=details)
raise
conn = Connection(networking_interface, self, self.address, conn_id, self.is_sdam) # type: ignore[arg-type]
@ -1131,18 +1090,15 @@ class Pool:
conn.authenticate()
# Catch KeyboardInterrupt, CancelledError, etc. and cleanup.
except BaseException as e:
except BaseException:
with self.lock:
self.active_contexts.discard(conn.cancel_context)
self._handle_connection_error(e, "hello", conn_id)
conn.close_conn(ConnectionClosedReason.ERROR)
raise
if handler:
handler.client._topology.receive_cluster_time(conn._cluster_time)
# Clear the backoff state.
self._backoff = 0
return conn
@contextlib.contextmanager
@ -1319,12 +1275,12 @@ class Pool:
# to be checked back into the pool.
with self._max_connecting_cond:
self._raise_if_not_ready(checkout_started_time, emit_event=False)
while not (self.conns or self._pending < self.max_connecting):
while not (self.conns or self._pending < self._max_connecting):
timeout = deadline - time.monotonic() if deadline else None
if not _cond_wait(self._max_connecting_cond, timeout):
# Timed out, notify the next thread to ensure a
# timeout doesn't consume the condition.
if self.conns or self._pending < self.max_connecting:
if self.conns or self._pending < self._max_connecting:
self._max_connecting_cond.notify()
emitted_event = True
self._raise_wait_queue_timeout(checkout_started_time)
@ -1465,8 +1421,8 @@ class Pool:
:class:`~pymongo.errors.AutoReconnect` exceptions on server
hiccups, etc. We only check if the socket was closed by an external
error if it has been > 1 second since the socket was checked into the
pool, or we are in backoff mode, to keep performance reasonable -
we can't avoid AutoReconnects completely anyway.
pool, to keep performance reasonable - we can't avoid AutoReconnects
completely anyway.
"""
idle_time_seconds = conn.idle_time_seconds()
# If socket is idle, open a new one.
@ -1477,11 +1433,8 @@ class Pool:
conn.close_conn(ConnectionClosedReason.IDLE)
return True
check_interval_seconds = self._check_interval_seconds
if self._backoff:
check_interval_seconds = 0
if check_interval_seconds is not None and (
check_interval_seconds == 0 or idle_time_seconds > check_interval_seconds
if self._check_interval_seconds is not None and (
self._check_interval_seconds == 0 or idle_time_seconds > self._check_interval_seconds
):
if conn.conn_closed():
conn.close_conn(ConnectionClosedReason.ERROR)

4
pymongo/synchronous/topology.py
View File

@ -888,9 +888,7 @@ class Topology:
# Clear the pool.
server.reset(service_id)
elif isinstance(error, ConnectionFailure):
if isinstance(error, WaitQueueTimeoutError) or error.has_error_label(
"SystemOverloadedError"
):
if isinstance(error, WaitQueueTimeoutError):
return
# "Client MUST replace the server's description with type Unknown
# ... MUST NOT request an immediate check of the server."

2
test/asynchronous/test_encryption.py
View File

@ -1276,7 +1276,7 @@ class TestBsonSizeBatches(AsyncEncryptionIntegrationTest):
with self.assertRaises(BulkWriteError) as ctx:
await self.coll_encrypted.bulk_write([InsertOne(doc)])
err = ctx.exception.details["writeErrors"][0]
self.assertIn(err["code"], [2, 10334])
self.assertEqual(2, err["code"])
self.assertIn("object to insert too large", err["errmsg"])

72
test/asynchronous/test_pooling.py
View File

@ -29,7 +29,6 @@ from pymongo import AsyncMongoClient, message, timeout
from pymongo.errors import AutoReconnect, ConnectionFailure, DuplicateKeyError
from pymongo.hello import HelloCompat
from pymongo.lock import _async_create_lock
from pymongo.read_preferences import ReadPreference
sys.path[0:0] = [""]
@ -514,77 +513,6 @@ class TestPooling(_TestPoolingBase):
str(error.exception),
)
async def test_pool_check_backoff(self):
# Test that Pool recovers from two connection failures in a row.
# This exercises code at the end of Pool._check().
cx_pool = await self.create_pool(max_pool_size=1, connect_timeout=1, wait_queue_timeout=1)
self.addAsyncCleanup(cx_pool.close)
async with cx_pool.checkout() as conn:
# Simulate a closed socket without telling the Connection it's
# closed.
await conn.conn.close()
# Enable backoff.
cx_pool._backoff = 1
# Swap pool's address with a bad one.
address, cx_pool.address = cx_pool.address, ("foo.com", 1234)
with self.assertRaises(AutoReconnect):
async with cx_pool.checkout():
pass
# Back to normal, semaphore was correctly released.
cx_pool.address = address
async with cx_pool.checkout():
pass
@async_client_context.require_failCommand_appName
async def test_pool_backoff_preserves_existing_connections(self):
client = await self.async_rs_or_single_client()
coll = self.db.t
pool = await async_get_pool(client)
await coll.insert_many([{"x": 1} for _ in range(10)])
t = SocketGetter(self.c, pool)
await t.start()
while t.state != "connection":
await asyncio.sleep(0.1)
assert not t.sock.conn_closed()
# Mock a session establishment overload.
mock_connection_fail = {
"configureFailPoint": "failCommand",
"mode": {"times": 1},
"data": {
"closeConnection": True,
},
}
async with self.fail_point(mock_connection_fail):
await coll.find_one({})
# Make sure the pool is out of backoff state.
assert pool._backoff == 0
# Make sure the existing socket was not affected.
assert not t.sock.conn_closed()
# Cleanup
await t.release_conn()
await t.join()
await pool.close()
async def test_pool_backoff_limits_maxConnecting(self):
client = await self.async_rs_or_single_client(maxConnecting=10)
pool = await async_get_pool(client)
assert pool.max_connecting == 10
pool._backoff = 1
assert pool.max_connecting == 1
pool._backoff = 0
assert pool.max_connecting == 10
await client.close()
class TestPoolMaxSize(_TestPoolingBase):
async def test_max_pool_size(self):

8
test/connection_logging/connection-logging.json
View File

@ -331,9 +331,7 @@
"uriOptions": {
"retryReads": false,
"appname": "clientAppName",
"heartbeatFrequencyMS": 10000,
"socketTimeoutMS": 500,
"connectTimeoutMS": 500
"heartbeatFrequencyMS": 10000
},
"observeLogMessages": {
"connection": "debug"
@ -357,9 +355,7 @@
"failCommands": [
"saslContinue"
],
"closeConnection": false,
"blockConnection": true,
"blockTimeMS": 1000,
"closeConnection": true,
"appName": "clientAppName"
}
}

10
test/connection_monitoring/pool-create-min-size-error.json
View File

@ -9,23 +9,21 @@
],
"failPoint": {
"configureFailPoint": "failCommand",
"mode": "alwaysOn",
"mode": {
"times": 50
},
"data": {
"failCommands": [
"isMaster",
"hello"
],
"closeConnection": false,
"blockConnection": true,
"blockTimeMS": 1000,
"closeConnection": true,
"appName": "poolCreateMinSizeErrorTest"
}
},
"poolOptions": {
"minPoolSize": 1,
"backgroundThreadIntervalMS": 50,
"socketTimeoutMS": 500,
"connectTimeoutMS": 500,
"appName": "poolCreateMinSizeErrorTest"
},
"operations": [

6
test/discovery_and_monitoring/unified/auth-network-error.json
View File

@ -53,9 +53,7 @@
"failCommands": [
"saslContinue"
],
"closeConnection": false,
"blockConnection": true,
"blockTimeMS": 1000,
"closeConnection": true,
"appName": "authNetworkErrorTest"
}
}
@ -77,8 +75,6 @@
],
"uriOptions": {
"retryWrites": false,
"socketTimeoutMS": 500,
"connectTimeoutMS": 500,
"appname": "authNetworkErrorTest"
}
}

15
test/load_balancer/sdam-error-handling.json
View File

@ -32,8 +32,6 @@
"useMultipleMongoses": false,
"uriOptions": {
"appname": "lbSDAMErrorTestClient",
"socketTimeoutMS": 500,
"connectTimeoutMS": 500,
"retryWrites": false
},
"observeEvents": [
@ -66,9 +64,7 @@
"id": "multiClient",
"useMultipleMongoses": true,
"uriOptions": {
"retryWrites": false,
"socketTimeoutMS": 500,
"connectTimeoutMS": 500
"retryWrites": false
},
"observeEvents": [
"connectionCreatedEvent",
@ -286,8 +282,7 @@
"isMaster",
"hello"
],
"blockConnection": true,
"blockTimeMS": 1000,
"closeConnection": true,
"appName": "lbSDAMErrorTestClient"
}
}
@ -350,8 +345,7 @@
"failCommands": [
"saslContinue"
],
"blockConnection": true,
"blockTimeMS": 1000,
"closeConnection": true,
"appName": "lbSDAMErrorTestClient"
}
}
@ -412,8 +406,7 @@
"failCommands": [
"getMore"
],
"closeConnection": true,
"appName": "lbSDAMErrorTestClient"
"closeConnection": true
}
}
}

2
test/test_encryption.py
View File

@ -1272,7 +1272,7 @@ class TestBsonSizeBatches(EncryptionIntegrationTest):
with self.assertRaises(BulkWriteError) as ctx:
self.coll_encrypted.bulk_write([InsertOne(doc)])
err = ctx.exception.details["writeErrors"][0]
self.assertIn(err["code"], [2, 10334])
self.assertEqual(2, err["code"])
self.assertIn("object to insert too large", err["errmsg"])

72
test/test_pooling.py
View File

@ -29,7 +29,6 @@ from pymongo import MongoClient, message, timeout
from pymongo.errors import AutoReconnect, ConnectionFailure, DuplicateKeyError
from pymongo.hello import HelloCompat
from pymongo.lock import _create_lock
from pymongo.read_preferences import ReadPreference
sys.path[0:0] = [""]
@ -512,77 +511,6 @@ class TestPooling(_TestPoolingBase):
str(error.exception),
)
def test_pool_check_backoff(self):
# Test that Pool recovers from two connection failures in a row.
# This exercises code at the end of Pool._check().
cx_pool = self.create_pool(max_pool_size=1, connect_timeout=1, wait_queue_timeout=1)
self.addCleanup(cx_pool.close)
with cx_pool.checkout() as conn:
# Simulate a closed socket without telling the Connection it's
# closed.
conn.conn.close()
# Enable backoff.
cx_pool._backoff = 1
# Swap pool's address with a bad one.
address, cx_pool.address = cx_pool.address, ("foo.com", 1234)
with self.assertRaises(AutoReconnect):
with cx_pool.checkout():
pass
# Back to normal, semaphore was correctly released.
cx_pool.address = address
with cx_pool.checkout():
pass
@client_context.require_failCommand_appName
def test_pool_backoff_preserves_existing_connections(self):
client = self.rs_or_single_client()
coll = self.db.t
pool = get_pool(client)
coll.insert_many([{"x": 1} for _ in range(10)])
t = SocketGetter(self.c, pool)
t.start()
while t.state != "connection":
time.sleep(0.1)
assert not t.sock.conn_closed()
# Mock a session establishment overload.
mock_connection_fail = {
"configureFailPoint": "failCommand",
"mode": {"times": 1},
"data": {
"closeConnection": True,
},
}
with self.fail_point(mock_connection_fail):
coll.find_one({})
# Make sure the pool is out of backoff state.
assert pool._backoff == 0
# Make sure the existing socket was not affected.
assert not t.sock.conn_closed()
# Cleanup
t.release_conn()
t.join()
pool.close()
def test_pool_backoff_limits_maxConnecting(self):
client = self.rs_or_single_client(maxConnecting=10)
pool = get_pool(client)
assert pool.max_connecting == 10
pool._backoff = 1
assert pool.max_connecting == 1
pool._backoff = 0
assert pool.max_connecting == 10
client.close()
class TestPoolMaxSize(_TestPoolingBase):
def test_max_pool_size(self):

2
tools/synchro.py
View File

@ -341,7 +341,7 @@ def translate_async_sleeps(lines: list[str]) -> list[str]:
sleeps = [line for line in lines if "asyncio.sleep" in line]
for line in sleeps:
res = re.search(r"asyncio\.sleep\(\s*(.*?)\)", line)
res = re.search(r"asyncio.sleep\(([^()]*)\)", line)
if res:
old = res[0]
index = lines.index(line)