PYTHON-5528 & PYTHON-5651 Add exponential backoff to operation retry loop for server overloaded errors (#2635)
Co-authored-by: Kevin Albertson <kevin.albertson@mongodb.com> Co-authored-by: Casey Clements <caseyclements@users.noreply.github.com>
This commit is contained in:
Steven Silvester
GitHub
parent
27a9f477a9
commit
8dbf90372b
@ -94,6 +94,9 @@ do
|
||||
change-streams|change_streams)
|
||||
cpjson change-streams/tests/ change_streams/
|
||||
;;
|
||||
client-backpressure|client_backpressure)
|
||||
cpjson client-backpressure/tests client-backpressure
|
||||
;;
|
||||
client-side-encryption|csfle|fle)
|
||||
cpjson client-side-encryption/tests/ client-side-encryption/spec
|
||||
cpjson client-side-encryption/corpus/ client-side-encryption/corpus
|
||||
|
||||
16
justfile
16
justfile
@ -1,7 +1,5 @@
|
||||
# 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 --with numpy --extra ocsp --extra snappy --extra test --extra zstd"
|
||||
@ -16,7 +14,7 @@ default:
|
||||
|
||||
[private]
|
||||
resync:
|
||||
@uv sync --quiet --frozen
|
||||
@uv sync --quiet
|
||||
|
||||
install:
|
||||
bash .evergreen/scripts/setup-dev-env.sh
|
||||
@ -50,12 +48,12 @@ typing-pyright: && resync
|
||||
{{typing_run}} python -m pyright -p strict_pyrightconfig.json test/test_typing_strict.py
|
||||
|
||||
[group('lint')]
|
||||
lint: && resync
|
||||
uv run pre-commit run --all-files
|
||||
lint *args="": && resync
|
||||
uvx pre-commit run --all-files {{args}}
|
||||
|
||||
[group('lint')]
|
||||
lint-manual: && resync
|
||||
uv run pre-commit run --all-files --hook-stage manual
|
||||
lint-manual *args="": && resync
|
||||
uvx pre-commit run --all-files --hook-stage manual {{args}}
|
||||
|
||||
[group('test')]
|
||||
test *args="-v --durations=5 --maxfail=10": && resync
|
||||
@ -77,6 +75,10 @@ setup-tests *args="":
|
||||
teardown-tests:
|
||||
bash .evergreen/scripts/teardown-tests.sh
|
||||
|
||||
[group('test')]
|
||||
integration-tests:
|
||||
bash integration_tests/run.sh
|
||||
|
||||
[group('server')]
|
||||
run-server *args="":
|
||||
bash .evergreen/scripts/run-server.sh {{args}}
|
||||
|
||||
@ -563,9 +563,22 @@ class _AsyncClientBulk:
|
||||
error, ConnectionFailure
|
||||
) and not isinstance(error, (NotPrimaryError, WaitQueueTimeoutError))
|
||||
|
||||
retryable_label_error = (
|
||||
hasattr(error, "details")
|
||||
and isinstance(error.details, dict)
|
||||
and "errorLabels" in error.details
|
||||
and isinstance(error.details["errorLabels"], list)
|
||||
and "RetryableError" in error.details["errorLabels"]
|
||||
and "SystemOverloadedError" in error.details["errorLabels"]
|
||||
)
|
||||
|
||||
# Synthesize the full bulk result without modifying the
|
||||
# current one because this write operation may be retried.
|
||||
if retryable and (retryable_top_level_error or retryable_network_error):
|
||||
if retryable and (
|
||||
retryable_top_level_error
|
||||
or retryable_network_error
|
||||
or retryable_label_error
|
||||
):
|
||||
full = copy.deepcopy(full_result)
|
||||
_merge_command(self.ops, self.idx_offset, full, result)
|
||||
_throw_client_bulk_write_exception(full, self.verbose_results)
|
||||
|
||||
@ -406,6 +406,7 @@ class _Transaction:
|
||||
self.recovery_token = None
|
||||
self.attempt = 0
|
||||
self.client = client
|
||||
self.has_completed_command = False
|
||||
|
||||
def active(self) -> bool:
|
||||
return self.state in (_TxnState.STARTING, _TxnState.IN_PROGRESS)
|
||||
@ -413,6 +414,9 @@ class _Transaction:
|
||||
def starting(self) -> bool:
|
||||
return self.state == _TxnState.STARTING
|
||||
|
||||
def set_starting(self) -> None:
|
||||
self.state = _TxnState.STARTING
|
||||
|
||||
@property
|
||||
def pinned_conn(self) -> Optional[AsyncConnection]:
|
||||
if self.active() and self.conn_mgr:
|
||||
|
||||
@ -20,7 +20,6 @@ from collections import abc
|
||||
from typing import (
|
||||
TYPE_CHECKING,
|
||||
Any,
|
||||
AsyncContextManager,
|
||||
Callable,
|
||||
Coroutine,
|
||||
Generic,
|
||||
@ -58,7 +57,6 @@ from pymongo.asynchronous.cursor import (
|
||||
AsyncCursor,
|
||||
AsyncRawBatchCursor,
|
||||
)
|
||||
from pymongo.asynchronous.helpers import _retry_overload
|
||||
from pymongo.collation import validate_collation_or_none
|
||||
from pymongo.common import _ecoc_coll_name, _esc_coll_name
|
||||
from pymongo.errors import (
|
||||
@ -573,11 +571,6 @@ class AsyncCollection(common.BaseObject, Generic[_DocumentType]):
|
||||
await change_stream._initialize_cursor()
|
||||
return change_stream
|
||||
|
||||
async def _conn_for_writes(
|
||||
self, session: Optional[AsyncClientSession], operation: str
|
||||
) -> AsyncContextManager[AsyncConnection]:
|
||||
return await self._database.client._conn_for_writes(session, operation)
|
||||
|
||||
async def _command(
|
||||
self,
|
||||
conn: AsyncConnection,
|
||||
@ -654,7 +647,10 @@ class AsyncCollection(common.BaseObject, Generic[_DocumentType]):
|
||||
if "size" in options:
|
||||
options["size"] = float(options["size"])
|
||||
cmd.update(options)
|
||||
async with await self._conn_for_writes(session, operation=_Op.CREATE) as conn:
|
||||
|
||||
async def inner(
|
||||
session: Optional[AsyncClientSession], conn: AsyncConnection, _retryable_write: bool
|
||||
) -> None:
|
||||
if qev2_required and conn.max_wire_version < 21:
|
||||
raise ConfigurationError(
|
||||
"Driver support of Queryable Encryption is incompatible with server. "
|
||||
@ -671,6 +667,8 @@ class AsyncCollection(common.BaseObject, Generic[_DocumentType]):
|
||||
session=session,
|
||||
)
|
||||
|
||||
await self.database.client._retryable_write(False, inner, session, _Op.CREATE)
|
||||
|
||||
async def _create(
|
||||
self,
|
||||
options: MutableMapping[str, Any],
|
||||
@ -2229,7 +2227,6 @@ class AsyncCollection(common.BaseObject, Generic[_DocumentType]):
|
||||
return await self._create_indexes(indexes, session, **kwargs)
|
||||
|
||||
@_csot.apply
|
||||
@_retry_overload
|
||||
async def _create_indexes(
|
||||
self, indexes: Sequence[IndexModel], session: Optional[AsyncClientSession], **kwargs: Any
|
||||
) -> list[str]:
|
||||
@ -2243,7 +2240,10 @@ class AsyncCollection(common.BaseObject, Generic[_DocumentType]):
|
||||
command (like maxTimeMS) can be passed as keyword arguments.
|
||||
"""
|
||||
names = []
|
||||
async with await self._conn_for_writes(session, operation=_Op.CREATE_INDEXES) as conn:
|
||||
|
||||
async def inner(
|
||||
session: Optional[AsyncClientSession], conn: AsyncConnection, _retryable_write: bool
|
||||
) -> list[str]:
|
||||
supports_quorum = conn.max_wire_version >= 9
|
||||
|
||||
def gen_indexes() -> Iterator[Mapping[str, Any]]:
|
||||
@ -2272,7 +2272,11 @@ class AsyncCollection(common.BaseObject, Generic[_DocumentType]):
|
||||
write_concern=self._write_concern_for(session),
|
||||
session=session,
|
||||
)
|
||||
return names
|
||||
return names
|
||||
|
||||
return await self.database.client._retryable_write(
|
||||
False, inner, session, _Op.CREATE_INDEXES
|
||||
)
|
||||
|
||||
async def create_index(
|
||||
self,
|
||||
@ -2474,7 +2478,6 @@ class AsyncCollection(common.BaseObject, Generic[_DocumentType]):
|
||||
await self._drop_index(index_or_name, session, comment, **kwargs)
|
||||
|
||||
@_csot.apply
|
||||
@_retry_overload
|
||||
async def _drop_index(
|
||||
self,
|
||||
index_or_name: _IndexKeyHint,
|
||||
@ -2493,7 +2496,10 @@ class AsyncCollection(common.BaseObject, Generic[_DocumentType]):
|
||||
cmd.update(kwargs)
|
||||
if comment is not None:
|
||||
cmd["comment"] = comment
|
||||
async with await self._conn_for_writes(session, operation=_Op.DROP_INDEXES) as conn:
|
||||
|
||||
async def inner(
|
||||
session: Optional[AsyncClientSession], conn: AsyncConnection, _retryable_write: bool
|
||||
) -> None:
|
||||
await self._command(
|
||||
conn,
|
||||
cmd,
|
||||
@ -2503,6 +2509,8 @@ class AsyncCollection(common.BaseObject, Generic[_DocumentType]):
|
||||
session=session,
|
||||
)
|
||||
|
||||
await self.database.client._retryable_write(False, inner, session, _Op.DROP_INDEXES)
|
||||
|
||||
async def list_indexes(
|
||||
self,
|
||||
session: Optional[AsyncClientSession] = None,
|
||||
@ -2766,17 +2774,22 @@ class AsyncCollection(common.BaseObject, Generic[_DocumentType]):
|
||||
cmd = {"createSearchIndexes": self.name, "indexes": list(gen_indexes())}
|
||||
cmd.update(kwargs)
|
||||
|
||||
async with await self._conn_for_writes(
|
||||
session, operation=_Op.CREATE_SEARCH_INDEXES
|
||||
) as conn:
|
||||
async def inner(
|
||||
session: Optional[AsyncClientSession], conn: AsyncConnection, _retryable_write: bool
|
||||
) -> list[str]:
|
||||
resp = await self._command(
|
||||
conn,
|
||||
cmd,
|
||||
read_preference=ReadPreference.PRIMARY,
|
||||
codec_options=_UNICODE_REPLACE_CODEC_OPTIONS,
|
||||
session=session,
|
||||
)
|
||||
return [index["name"] for index in resp["indexesCreated"]]
|
||||
|
||||
return await self.database.client._retryable_write(
|
||||
False, inner, session, _Op.CREATE_SEARCH_INDEXES
|
||||
)
|
||||
|
||||
async def drop_search_index(
|
||||
self,
|
||||
name: str,
|
||||
@ -2802,15 +2815,21 @@ class AsyncCollection(common.BaseObject, Generic[_DocumentType]):
|
||||
cmd.update(kwargs)
|
||||
if comment is not None:
|
||||
cmd["comment"] = comment
|
||||
async with await self._conn_for_writes(session, operation=_Op.DROP_SEARCH_INDEXES) as conn:
|
||||
|
||||
async def inner(
|
||||
session: Optional[AsyncClientSession], conn: AsyncConnection, _retryable_write: bool
|
||||
) -> None:
|
||||
await self._command(
|
||||
conn,
|
||||
cmd,
|
||||
read_preference=ReadPreference.PRIMARY,
|
||||
allowable_errors=["ns not found", 26],
|
||||
codec_options=_UNICODE_REPLACE_CODEC_OPTIONS,
|
||||
session=session,
|
||||
)
|
||||
|
||||
await self.database.client._retryable_write(False, inner, session, _Op.DROP_SEARCH_INDEXES)
|
||||
|
||||
async def update_search_index(
|
||||
self,
|
||||
name: str,
|
||||
@ -2838,15 +2857,21 @@ class AsyncCollection(common.BaseObject, Generic[_DocumentType]):
|
||||
cmd.update(kwargs)
|
||||
if comment is not None:
|
||||
cmd["comment"] = comment
|
||||
async with await self._conn_for_writes(session, operation=_Op.UPDATE_SEARCH_INDEX) as conn:
|
||||
|
||||
async def inner(
|
||||
session: Optional[AsyncClientSession], conn: AsyncConnection, _retryable_write: bool
|
||||
) -> None:
|
||||
await self._command(
|
||||
conn,
|
||||
cmd,
|
||||
read_preference=ReadPreference.PRIMARY,
|
||||
allowable_errors=["ns not found", 26],
|
||||
codec_options=_UNICODE_REPLACE_CODEC_OPTIONS,
|
||||
session=session,
|
||||
)
|
||||
|
||||
await self.database.client._retryable_write(False, inner, session, _Op.UPDATE_SEARCH_INDEX)
|
||||
|
||||
async def options(
|
||||
self,
|
||||
session: Optional[AsyncClientSession] = None,
|
||||
@ -3075,7 +3100,6 @@ class AsyncCollection(common.BaseObject, Generic[_DocumentType]):
|
||||
)
|
||||
|
||||
@_csot.apply
|
||||
@_retry_overload
|
||||
async def rename(
|
||||
self,
|
||||
new_name: str,
|
||||
@ -3127,17 +3151,21 @@ class AsyncCollection(common.BaseObject, Generic[_DocumentType]):
|
||||
if comment is not None:
|
||||
cmd["comment"] = comment
|
||||
write_concern = self._write_concern_for_cmd(cmd, session)
|
||||
client = self._database.client
|
||||
|
||||
async with await self._conn_for_writes(session, operation=_Op.RENAME) as conn:
|
||||
async with self._database.client._tmp_session(session) as s:
|
||||
return await conn.command(
|
||||
"admin",
|
||||
cmd,
|
||||
write_concern=write_concern,
|
||||
parse_write_concern_error=True,
|
||||
session=s,
|
||||
client=self._database.client,
|
||||
)
|
||||
async def inner(
|
||||
session: Optional[AsyncClientSession], conn: AsyncConnection, _retryable_write: bool
|
||||
) -> MutableMapping[str, Any]:
|
||||
return await conn.command(
|
||||
"admin",
|
||||
cmd,
|
||||
write_concern=write_concern,
|
||||
parse_write_concern_error=True,
|
||||
session=session,
|
||||
client=client,
|
||||
)
|
||||
|
||||
return await client._retryable_write(False, inner, session, _Op.RENAME)
|
||||
|
||||
async def distinct(
|
||||
self,
|
||||
|
||||
@ -38,7 +38,6 @@ from pymongo.asynchronous.aggregation import _DatabaseAggregationCommand
|
||||
from pymongo.asynchronous.change_stream import AsyncDatabaseChangeStream
|
||||
from pymongo.asynchronous.collection import AsyncCollection
|
||||
from pymongo.asynchronous.command_cursor import AsyncCommandCursor
|
||||
from pymongo.asynchronous.helpers import _retry_overload
|
||||
from pymongo.common import _ecoc_coll_name, _esc_coll_name
|
||||
from pymongo.database_shared import _check_name, _CodecDocumentType
|
||||
from pymongo.errors import CollectionInvalid, InvalidOperation
|
||||
@ -479,7 +478,6 @@ class AsyncDatabase(common.BaseObject, Generic[_DocumentType]):
|
||||
return change_stream
|
||||
|
||||
@_csot.apply
|
||||
@_retry_overload
|
||||
async def create_collection(
|
||||
self,
|
||||
name: str,
|
||||
@ -822,7 +820,6 @@ class AsyncDatabase(common.BaseObject, Generic[_DocumentType]):
|
||||
...
|
||||
|
||||
@_csot.apply
|
||||
@_retry_overload
|
||||
async def command(
|
||||
self,
|
||||
command: Union[str, MutableMapping[str, Any]],
|
||||
@ -935,14 +932,15 @@ class AsyncDatabase(common.BaseObject, Generic[_DocumentType]):
|
||||
|
||||
if read_preference is None:
|
||||
read_preference = (session and session._txn_read_preference()) or ReadPreference.PRIMARY
|
||||
async with await self._client._conn_for_reads(
|
||||
read_preference, session, operation=command_name
|
||||
) as (
|
||||
connection,
|
||||
read_preference,
|
||||
):
|
||||
|
||||
async def inner(
|
||||
session: Optional[AsyncClientSession],
|
||||
_server: Server,
|
||||
conn: AsyncConnection,
|
||||
read_preference: _ServerMode,
|
||||
) -> Union[dict[str, Any], _CodecDocumentType]:
|
||||
return await self._command(
|
||||
connection,
|
||||
conn,
|
||||
command,
|
||||
value,
|
||||
check,
|
||||
@ -953,8 +951,11 @@ class AsyncDatabase(common.BaseObject, Generic[_DocumentType]):
|
||||
**kwargs,
|
||||
)
|
||||
|
||||
return await self._client._retryable_read(
|
||||
inner, read_preference, session, command_name, None, False
|
||||
)
|
||||
|
||||
@_csot.apply
|
||||
@_retry_overload
|
||||
async def cursor_command(
|
||||
self,
|
||||
command: Union[str, MutableMapping[str, Any]],
|
||||
@ -1021,17 +1022,17 @@ class AsyncDatabase(common.BaseObject, Generic[_DocumentType]):
|
||||
|
||||
async with self._client._tmp_session(session) as tmp_session:
|
||||
opts = codec_options or DEFAULT_CODEC_OPTIONS
|
||||
|
||||
if read_preference is None:
|
||||
read_preference = (
|
||||
tmp_session and tmp_session._txn_read_preference()
|
||||
) or ReadPreference.PRIMARY
|
||||
async with await self._client._conn_for_reads(
|
||||
read_preference, tmp_session, command_name
|
||||
) as (
|
||||
conn,
|
||||
read_preference,
|
||||
):
|
||||
|
||||
async def inner(
|
||||
session: Optional[AsyncClientSession],
|
||||
_server: Server,
|
||||
conn: AsyncConnection,
|
||||
read_preference: _ServerMode,
|
||||
) -> AsyncCommandCursor[_DocumentType]:
|
||||
response = await self._command(
|
||||
conn,
|
||||
command,
|
||||
@ -1040,7 +1041,7 @@ class AsyncDatabase(common.BaseObject, Generic[_DocumentType]):
|
||||
None,
|
||||
read_preference,
|
||||
opts,
|
||||
session=tmp_session,
|
||||
session=session,
|
||||
**kwargs,
|
||||
)
|
||||
coll = self.get_collection("$cmd", read_preference=read_preference)
|
||||
@ -1050,7 +1051,7 @@ class AsyncDatabase(common.BaseObject, Generic[_DocumentType]):
|
||||
response["cursor"],
|
||||
conn.address,
|
||||
max_await_time_ms=max_await_time_ms,
|
||||
session=tmp_session,
|
||||
session=session,
|
||||
comment=comment,
|
||||
)
|
||||
await cmd_cursor._maybe_pin_connection(conn)
|
||||
@ -1058,6 +1059,10 @@ class AsyncDatabase(common.BaseObject, Generic[_DocumentType]):
|
||||
else:
|
||||
raise InvalidOperation("Command does not return a cursor.")
|
||||
|
||||
return await self.client._retryable_read(
|
||||
inner, read_preference, tmp_session, command_name, None, False
|
||||
)
|
||||
|
||||
async def _retryable_read_command(
|
||||
self,
|
||||
command: Union[str, MutableMapping[str, Any]],
|
||||
@ -1259,9 +1264,11 @@ class AsyncDatabase(common.BaseObject, Generic[_DocumentType]):
|
||||
if comment is not None:
|
||||
command["comment"] = comment
|
||||
|
||||
async with await self._client._conn_for_writes(session, operation=_Op.DROP) as connection:
|
||||
async def inner(
|
||||
session: Optional[AsyncClientSession], conn: AsyncConnection, _retryable_write: bool
|
||||
) -> dict[str, Any]:
|
||||
return await self._command(
|
||||
connection,
|
||||
conn,
|
||||
command,
|
||||
allowable_errors=["ns not found", 26],
|
||||
write_concern=self._write_concern_for(session),
|
||||
@ -1269,8 +1276,9 @@ class AsyncDatabase(common.BaseObject, Generic[_DocumentType]):
|
||||
session=session,
|
||||
)
|
||||
|
||||
return await self.client._retryable_write(False, inner, session, _Op.DROP)
|
||||
|
||||
@_csot.apply
|
||||
@_retry_overload
|
||||
async def drop_collection(
|
||||
self,
|
||||
name_or_collection: Union[str, AsyncCollection[_DocumentTypeArg]],
|
||||
|
||||
@ -32,7 +32,6 @@ from typing import (
|
||||
from pymongo import _csot
|
||||
from pymongo.errors import (
|
||||
OperationFailure,
|
||||
PyMongoError,
|
||||
)
|
||||
from pymongo.helpers_shared import _REAUTHENTICATION_REQUIRED_CODE
|
||||
from pymongo.lock import _async_create_lock
|
||||
@ -166,34 +165,6 @@ class _RetryPolicy:
|
||||
return True
|
||||
|
||||
|
||||
def _retry_overload(func: F) -> F:
|
||||
@functools.wraps(func)
|
||||
async def inner(self: Any, *args: Any, **kwargs: Any) -> Any:
|
||||
retry_policy = self._retry_policy
|
||||
attempt = 0
|
||||
while True:
|
||||
try:
|
||||
res = await func(self, *args, **kwargs)
|
||||
await retry_policy.record_success(retry=attempt > 0)
|
||||
return res
|
||||
except PyMongoError as exc:
|
||||
if not exc.has_error_label("RetryableError"):
|
||||
raise
|
||||
attempt += 1
|
||||
delay = 0
|
||||
if exc.has_error_label("SystemOverloadedError"):
|
||||
delay = retry_policy.backoff(attempt)
|
||||
if not await retry_policy.should_retry(attempt, delay):
|
||||
raise
|
||||
|
||||
# Implement exponential backoff on retry.
|
||||
if delay:
|
||||
await asyncio.sleep(delay)
|
||||
continue
|
||||
|
||||
return cast(F, inner)
|
||||
|
||||
|
||||
async def _getaddrinfo(
|
||||
host: Any, port: Any, **kwargs: Any
|
||||
) -> list[
|
||||
|
||||
@ -69,7 +69,6 @@ from pymongo.asynchronous.client_bulk import _AsyncClientBulk
|
||||
from pymongo.asynchronous.client_session import _EmptyServerSession
|
||||
from pymongo.asynchronous.command_cursor import AsyncCommandCursor
|
||||
from pymongo.asynchronous.helpers import (
|
||||
_retry_overload,
|
||||
_RetryPolicy,
|
||||
_TokenBucket,
|
||||
)
|
||||
@ -2403,7 +2402,6 @@ class AsyncMongoClient(common.BaseObject, Generic[_DocumentType]):
|
||||
return [doc["name"] async for doc in res]
|
||||
|
||||
@_csot.apply
|
||||
@_retry_overload
|
||||
async def drop_database(
|
||||
self,
|
||||
name_or_database: Union[str, database.AsyncDatabase[_DocumentTypeArg]],
|
||||
@ -2446,15 +2444,13 @@ class AsyncMongoClient(common.BaseObject, Generic[_DocumentType]):
|
||||
f"name_or_database must be an instance of str or a AsyncDatabase, not {type(name)}"
|
||||
)
|
||||
|
||||
async with await self._conn_for_writes(session, operation=_Op.DROP_DATABASE) as conn:
|
||||
await self[name]._command(
|
||||
conn,
|
||||
{"dropDatabase": 1, "comment": comment},
|
||||
read_preference=ReadPreference.PRIMARY,
|
||||
write_concern=self._write_concern_for(session),
|
||||
parse_write_concern_error=True,
|
||||
session=session,
|
||||
)
|
||||
await self[name].command(
|
||||
{"dropDatabase": 1, "comment": comment},
|
||||
read_preference=ReadPreference.PRIMARY,
|
||||
write_concern=self._write_concern_for(session),
|
||||
parse_write_concern_error=True,
|
||||
session=session,
|
||||
)
|
||||
|
||||
@_csot.apply
|
||||
async def bulk_write(
|
||||
@ -2781,6 +2777,11 @@ class _ClientConnectionRetryable(Generic[T]):
|
||||
try:
|
||||
res = await self._read() if self._is_read else await self._write()
|
||||
await self._retry_policy.record_success(self._attempt_number > 0)
|
||||
# Track whether the transaction has completed a command.
|
||||
# If we need to apply backpressure to the first command,
|
||||
# we will need to revert back to starting state.
|
||||
if self._session is not None and self._session.in_transaction:
|
||||
self._session._transaction.has_completed_command = True
|
||||
return res
|
||||
except ServerSelectionTimeoutError:
|
||||
# The application may think the write was never attempted
|
||||
@ -2800,8 +2801,8 @@ class _ClientConnectionRetryable(Generic[T]):
|
||||
if isinstance(exc, (ConnectionFailure, OperationFailure)):
|
||||
# ConnectionFailures do not supply a code property
|
||||
exc_code = getattr(exc, "code", None)
|
||||
always_retryable = exc.has_error_label("RetryableError")
|
||||
overloaded = exc.has_error_label("SystemOverloadedError")
|
||||
always_retryable = exc.has_error_label("RetryableError") and overloaded
|
||||
if not always_retryable and (
|
||||
self._is_not_eligible_for_retry()
|
||||
or (
|
||||
@ -2813,6 +2814,18 @@ class _ClientConnectionRetryable(Generic[T]):
|
||||
self._retrying = True
|
||||
self._last_error = exc
|
||||
self._attempt_number += 1
|
||||
|
||||
# Revert back to starting state if we're in a transaction but haven't completed the first
|
||||
# command.
|
||||
if (
|
||||
overloaded
|
||||
and self._session is not None
|
||||
and self._session.in_transaction
|
||||
):
|
||||
transaction = self._session._transaction
|
||||
if not transaction.has_completed_command:
|
||||
transaction.set_starting()
|
||||
transaction.attempt = 0
|
||||
else:
|
||||
raise
|
||||
|
||||
@ -2823,8 +2836,8 @@ class _ClientConnectionRetryable(Generic[T]):
|
||||
):
|
||||
exc_to_check = exc.error
|
||||
retryable_write_label = exc_to_check.has_error_label("RetryableWriteError")
|
||||
always_retryable = exc_to_check.has_error_label("RetryableError")
|
||||
overloaded = exc_to_check.has_error_label("SystemOverloadedError")
|
||||
always_retryable = exc_to_check.has_error_label("RetryableError") and overloaded
|
||||
if not self._retryable and not always_retryable:
|
||||
raise
|
||||
if retryable_write_label or always_retryable:
|
||||
@ -2846,20 +2859,26 @@ class _ClientConnectionRetryable(Generic[T]):
|
||||
self._last_error = exc
|
||||
if self._last_error is None:
|
||||
self._last_error = exc
|
||||
# Revert back to starting state if we're in a transaction but haven't completed the first
|
||||
# command.
|
||||
if overloaded and self._session is not None and self._session.in_transaction:
|
||||
transaction = self._session._transaction
|
||||
if not transaction.has_completed_command:
|
||||
transaction.set_starting()
|
||||
transaction.attempt = 0
|
||||
|
||||
if self._server is not None:
|
||||
self._deprioritized_servers.append(self._server)
|
||||
|
||||
self._always_retryable = always_retryable
|
||||
if always_retryable:
|
||||
if overloaded:
|
||||
delay = self._retry_policy.backoff(self._attempt_number) if overloaded else 0
|
||||
if not await self._retry_policy.should_retry(self._attempt_number, delay):
|
||||
if exc_to_check.has_error_label("NoWritesPerformed") and self._last_error:
|
||||
raise self._last_error from exc
|
||||
else:
|
||||
raise
|
||||
if overloaded:
|
||||
await asyncio.sleep(delay)
|
||||
await asyncio.sleep(delay)
|
||||
|
||||
def _is_not_eligible_for_retry(self) -> bool:
|
||||
"""Checks if the exchange is not eligible for retry"""
|
||||
|
||||
@ -254,6 +254,7 @@ class AsyncConnection:
|
||||
cmd = self.hello_cmd()
|
||||
performing_handshake = not self.performed_handshake
|
||||
awaitable = False
|
||||
cmd["backpressure"] = True
|
||||
if performing_handshake:
|
||||
self.performed_handshake = True
|
||||
cmd["client"] = self.opts.metadata
|
||||
|
||||
@ -561,9 +561,22 @@ class _ClientBulk:
|
||||
error, ConnectionFailure
|
||||
) and not isinstance(error, (NotPrimaryError, WaitQueueTimeoutError))
|
||||
|
||||
retryable_label_error = (
|
||||
hasattr(error, "details")
|
||||
and isinstance(error.details, dict)
|
||||
and "errorLabels" in error.details
|
||||
and isinstance(error.details["errorLabels"], list)
|
||||
and "RetryableError" in error.details["errorLabels"]
|
||||
and "SystemOverloadedError" in error.details["errorLabels"]
|
||||
)
|
||||
|
||||
# Synthesize the full bulk result without modifying the
|
||||
# current one because this write operation may be retried.
|
||||
if retryable and (retryable_top_level_error or retryable_network_error):
|
||||
if retryable and (
|
||||
retryable_top_level_error
|
||||
or retryable_network_error
|
||||
or retryable_label_error
|
||||
):
|
||||
full = copy.deepcopy(full_result)
|
||||
_merge_command(self.ops, self.idx_offset, full, result)
|
||||
_throw_client_bulk_write_exception(full, self.verbose_results)
|
||||
|
||||
@ -404,6 +404,7 @@ class _Transaction:
|
||||
self.recovery_token = None
|
||||
self.attempt = 0
|
||||
self.client = client
|
||||
self.has_completed_command = False
|
||||
|
||||
def active(self) -> bool:
|
||||
return self.state in (_TxnState.STARTING, _TxnState.IN_PROGRESS)
|
||||
@ -411,6 +412,9 @@ class _Transaction:
|
||||
def starting(self) -> bool:
|
||||
return self.state == _TxnState.STARTING
|
||||
|
||||
def set_starting(self) -> None:
|
||||
self.state = _TxnState.STARTING
|
||||
|
||||
@property
|
||||
def pinned_conn(self) -> Optional[Connection]:
|
||||
if self.active() and self.conn_mgr:
|
||||
|
||||
@ -21,7 +21,6 @@ from typing import (
|
||||
TYPE_CHECKING,
|
||||
Any,
|
||||
Callable,
|
||||
ContextManager,
|
||||
Generic,
|
||||
Iterable,
|
||||
Iterator,
|
||||
@ -89,7 +88,6 @@ from pymongo.synchronous.cursor import (
|
||||
Cursor,
|
||||
RawBatchCursor,
|
||||
)
|
||||
from pymongo.synchronous.helpers import _retry_overload
|
||||
from pymongo.typings import _CollationIn, _DocumentType, _DocumentTypeArg, _Pipeline
|
||||
from pymongo.write_concern import DEFAULT_WRITE_CONCERN, WriteConcern, validate_boolean
|
||||
|
||||
@ -574,11 +572,6 @@ class Collection(common.BaseObject, Generic[_DocumentType]):
|
||||
change_stream._initialize_cursor()
|
||||
return change_stream
|
||||
|
||||
def _conn_for_writes(
|
||||
self, session: Optional[ClientSession], operation: str
|
||||
) -> ContextManager[Connection]:
|
||||
return self._database.client._conn_for_writes(session, operation)
|
||||
|
||||
def _command(
|
||||
self,
|
||||
conn: Connection,
|
||||
@ -655,7 +648,10 @@ class Collection(common.BaseObject, Generic[_DocumentType]):
|
||||
if "size" in options:
|
||||
options["size"] = float(options["size"])
|
||||
cmd.update(options)
|
||||
with self._conn_for_writes(session, operation=_Op.CREATE) as conn:
|
||||
|
||||
def inner(
|
||||
session: Optional[ClientSession], conn: Connection, _retryable_write: bool
|
||||
) -> None:
|
||||
if qev2_required and conn.max_wire_version < 21:
|
||||
raise ConfigurationError(
|
||||
"Driver support of Queryable Encryption is incompatible with server. "
|
||||
@ -672,6 +668,8 @@ class Collection(common.BaseObject, Generic[_DocumentType]):
|
||||
session=session,
|
||||
)
|
||||
|
||||
self.database.client._retryable_write(False, inner, session, _Op.CREATE)
|
||||
|
||||
def _create(
|
||||
self,
|
||||
options: MutableMapping[str, Any],
|
||||
@ -2226,7 +2224,6 @@ class Collection(common.BaseObject, Generic[_DocumentType]):
|
||||
return self._create_indexes(indexes, session, **kwargs)
|
||||
|
||||
@_csot.apply
|
||||
@_retry_overload
|
||||
def _create_indexes(
|
||||
self, indexes: Sequence[IndexModel], session: Optional[ClientSession], **kwargs: Any
|
||||
) -> list[str]:
|
||||
@ -2240,7 +2237,10 @@ class Collection(common.BaseObject, Generic[_DocumentType]):
|
||||
command (like maxTimeMS) can be passed as keyword arguments.
|
||||
"""
|
||||
names = []
|
||||
with self._conn_for_writes(session, operation=_Op.CREATE_INDEXES) as conn:
|
||||
|
||||
def inner(
|
||||
session: Optional[ClientSession], conn: Connection, _retryable_write: bool
|
||||
) -> list[str]:
|
||||
supports_quorum = conn.max_wire_version >= 9
|
||||
|
||||
def gen_indexes() -> Iterator[Mapping[str, Any]]:
|
||||
@ -2269,7 +2269,9 @@ class Collection(common.BaseObject, Generic[_DocumentType]):
|
||||
write_concern=self._write_concern_for(session),
|
||||
session=session,
|
||||
)
|
||||
return names
|
||||
return names
|
||||
|
||||
return self.database.client._retryable_write(False, inner, session, _Op.CREATE_INDEXES)
|
||||
|
||||
def create_index(
|
||||
self,
|
||||
@ -2471,7 +2473,6 @@ class Collection(common.BaseObject, Generic[_DocumentType]):
|
||||
self._drop_index(index_or_name, session, comment, **kwargs)
|
||||
|
||||
@_csot.apply
|
||||
@_retry_overload
|
||||
def _drop_index(
|
||||
self,
|
||||
index_or_name: _IndexKeyHint,
|
||||
@ -2490,7 +2491,10 @@ class Collection(common.BaseObject, Generic[_DocumentType]):
|
||||
cmd.update(kwargs)
|
||||
if comment is not None:
|
||||
cmd["comment"] = comment
|
||||
with self._conn_for_writes(session, operation=_Op.DROP_INDEXES) as conn:
|
||||
|
||||
def inner(
|
||||
session: Optional[ClientSession], conn: Connection, _retryable_write: bool
|
||||
) -> None:
|
||||
self._command(
|
||||
conn,
|
||||
cmd,
|
||||
@ -2500,6 +2504,8 @@ class Collection(common.BaseObject, Generic[_DocumentType]):
|
||||
session=session,
|
||||
)
|
||||
|
||||
self.database.client._retryable_write(False, inner, session, _Op.DROP_INDEXES)
|
||||
|
||||
def list_indexes(
|
||||
self,
|
||||
session: Optional[ClientSession] = None,
|
||||
@ -2763,15 +2769,22 @@ class Collection(common.BaseObject, Generic[_DocumentType]):
|
||||
cmd = {"createSearchIndexes": self.name, "indexes": list(gen_indexes())}
|
||||
cmd.update(kwargs)
|
||||
|
||||
with self._conn_for_writes(session, operation=_Op.CREATE_SEARCH_INDEXES) as conn:
|
||||
def inner(
|
||||
session: Optional[ClientSession], conn: Connection, _retryable_write: bool
|
||||
) -> list[str]:
|
||||
resp = self._command(
|
||||
conn,
|
||||
cmd,
|
||||
read_preference=ReadPreference.PRIMARY,
|
||||
codec_options=_UNICODE_REPLACE_CODEC_OPTIONS,
|
||||
session=session,
|
||||
)
|
||||
return [index["name"] for index in resp["indexesCreated"]]
|
||||
|
||||
return self.database.client._retryable_write(
|
||||
False, inner, session, _Op.CREATE_SEARCH_INDEXES
|
||||
)
|
||||
|
||||
def drop_search_index(
|
||||
self,
|
||||
name: str,
|
||||
@ -2797,15 +2810,21 @@ class Collection(common.BaseObject, Generic[_DocumentType]):
|
||||
cmd.update(kwargs)
|
||||
if comment is not None:
|
||||
cmd["comment"] = comment
|
||||
with self._conn_for_writes(session, operation=_Op.DROP_SEARCH_INDEXES) as conn:
|
||||
|
||||
def inner(
|
||||
session: Optional[ClientSession], conn: Connection, _retryable_write: bool
|
||||
) -> None:
|
||||
self._command(
|
||||
conn,
|
||||
cmd,
|
||||
read_preference=ReadPreference.PRIMARY,
|
||||
allowable_errors=["ns not found", 26],
|
||||
codec_options=_UNICODE_REPLACE_CODEC_OPTIONS,
|
||||
session=session,
|
||||
)
|
||||
|
||||
self.database.client._retryable_write(False, inner, session, _Op.DROP_SEARCH_INDEXES)
|
||||
|
||||
def update_search_index(
|
||||
self,
|
||||
name: str,
|
||||
@ -2833,15 +2852,21 @@ class Collection(common.BaseObject, Generic[_DocumentType]):
|
||||
cmd.update(kwargs)
|
||||
if comment is not None:
|
||||
cmd["comment"] = comment
|
||||
with self._conn_for_writes(session, operation=_Op.UPDATE_SEARCH_INDEX) as conn:
|
||||
|
||||
def inner(
|
||||
session: Optional[ClientSession], conn: Connection, _retryable_write: bool
|
||||
) -> None:
|
||||
self._command(
|
||||
conn,
|
||||
cmd,
|
||||
read_preference=ReadPreference.PRIMARY,
|
||||
allowable_errors=["ns not found", 26],
|
||||
codec_options=_UNICODE_REPLACE_CODEC_OPTIONS,
|
||||
session=session,
|
||||
)
|
||||
|
||||
self.database.client._retryable_write(False, inner, session, _Op.UPDATE_SEARCH_INDEX)
|
||||
|
||||
def options(
|
||||
self,
|
||||
session: Optional[ClientSession] = None,
|
||||
@ -3068,7 +3093,6 @@ class Collection(common.BaseObject, Generic[_DocumentType]):
|
||||
)
|
||||
|
||||
@_csot.apply
|
||||
@_retry_overload
|
||||
def rename(
|
||||
self,
|
||||
new_name: str,
|
||||
@ -3120,17 +3144,21 @@ class Collection(common.BaseObject, Generic[_DocumentType]):
|
||||
if comment is not None:
|
||||
cmd["comment"] = comment
|
||||
write_concern = self._write_concern_for_cmd(cmd, session)
|
||||
client = self._database.client
|
||||
|
||||
with self._conn_for_writes(session, operation=_Op.RENAME) as conn:
|
||||
with self._database.client._tmp_session(session) as s:
|
||||
return conn.command(
|
||||
"admin",
|
||||
cmd,
|
||||
write_concern=write_concern,
|
||||
parse_write_concern_error=True,
|
||||
session=s,
|
||||
client=self._database.client,
|
||||
)
|
||||
def inner(
|
||||
session: Optional[ClientSession], conn: Connection, _retryable_write: bool
|
||||
) -> MutableMapping[str, Any]:
|
||||
return conn.command(
|
||||
"admin",
|
||||
cmd,
|
||||
write_concern=write_concern,
|
||||
parse_write_concern_error=True,
|
||||
session=session,
|
||||
client=client,
|
||||
)
|
||||
|
||||
return client._retryable_write(False, inner, session, _Op.RENAME)
|
||||
|
||||
def distinct(
|
||||
self,
|
||||
|
||||
@ -43,7 +43,6 @@ from pymongo.synchronous.aggregation import _DatabaseAggregationCommand
|
||||
from pymongo.synchronous.change_stream import DatabaseChangeStream
|
||||
from pymongo.synchronous.collection import Collection
|
||||
from pymongo.synchronous.command_cursor import CommandCursor
|
||||
from pymongo.synchronous.helpers import _retry_overload
|
||||
from pymongo.typings import _CollationIn, _DocumentType, _DocumentTypeArg, _Pipeline
|
||||
|
||||
if TYPE_CHECKING:
|
||||
@ -479,7 +478,6 @@ class Database(common.BaseObject, Generic[_DocumentType]):
|
||||
return change_stream
|
||||
|
||||
@_csot.apply
|
||||
@_retry_overload
|
||||
def create_collection(
|
||||
self,
|
||||
name: str,
|
||||
@ -822,7 +820,6 @@ class Database(common.BaseObject, Generic[_DocumentType]):
|
||||
...
|
||||
|
||||
@_csot.apply
|
||||
@_retry_overload
|
||||
def command(
|
||||
self,
|
||||
command: Union[str, MutableMapping[str, Any]],
|
||||
@ -935,12 +932,15 @@ class Database(common.BaseObject, Generic[_DocumentType]):
|
||||
|
||||
if read_preference is None:
|
||||
read_preference = (session and session._txn_read_preference()) or ReadPreference.PRIMARY
|
||||
with self._client._conn_for_reads(read_preference, session, operation=command_name) as (
|
||||
connection,
|
||||
read_preference,
|
||||
):
|
||||
|
||||
def inner(
|
||||
session: Optional[ClientSession],
|
||||
_server: Server,
|
||||
conn: Connection,
|
||||
read_preference: _ServerMode,
|
||||
) -> Union[dict[str, Any], _CodecDocumentType]:
|
||||
return self._command(
|
||||
connection,
|
||||
conn,
|
||||
command,
|
||||
value,
|
||||
check,
|
||||
@ -951,8 +951,11 @@ class Database(common.BaseObject, Generic[_DocumentType]):
|
||||
**kwargs,
|
||||
)
|
||||
|
||||
return self._client._retryable_read(
|
||||
inner, read_preference, session, command_name, None, False
|
||||
)
|
||||
|
||||
@_csot.apply
|
||||
@_retry_overload
|
||||
def cursor_command(
|
||||
self,
|
||||
command: Union[str, MutableMapping[str, Any]],
|
||||
@ -1019,15 +1022,17 @@ class Database(common.BaseObject, Generic[_DocumentType]):
|
||||
|
||||
with self._client._tmp_session(session) as tmp_session:
|
||||
opts = codec_options or DEFAULT_CODEC_OPTIONS
|
||||
|
||||
if read_preference is None:
|
||||
read_preference = (
|
||||
tmp_session and tmp_session._txn_read_preference()
|
||||
) or ReadPreference.PRIMARY
|
||||
with self._client._conn_for_reads(read_preference, tmp_session, command_name) as (
|
||||
conn,
|
||||
read_preference,
|
||||
):
|
||||
|
||||
def inner(
|
||||
session: Optional[ClientSession],
|
||||
_server: Server,
|
||||
conn: Connection,
|
||||
read_preference: _ServerMode,
|
||||
) -> CommandCursor[_DocumentType]:
|
||||
response = self._command(
|
||||
conn,
|
||||
command,
|
||||
@ -1036,7 +1041,7 @@ class Database(common.BaseObject, Generic[_DocumentType]):
|
||||
None,
|
||||
read_preference,
|
||||
opts,
|
||||
session=tmp_session,
|
||||
session=session,
|
||||
**kwargs,
|
||||
)
|
||||
coll = self.get_collection("$cmd", read_preference=read_preference)
|
||||
@ -1046,7 +1051,7 @@ class Database(common.BaseObject, Generic[_DocumentType]):
|
||||
response["cursor"],
|
||||
conn.address,
|
||||
max_await_time_ms=max_await_time_ms,
|
||||
session=tmp_session,
|
||||
session=session,
|
||||
comment=comment,
|
||||
)
|
||||
cmd_cursor._maybe_pin_connection(conn)
|
||||
@ -1054,6 +1059,10 @@ class Database(common.BaseObject, Generic[_DocumentType]):
|
||||
else:
|
||||
raise InvalidOperation("Command does not return a cursor.")
|
||||
|
||||
return self.client._retryable_read(
|
||||
inner, read_preference, tmp_session, command_name, None, False
|
||||
)
|
||||
|
||||
def _retryable_read_command(
|
||||
self,
|
||||
command: Union[str, MutableMapping[str, Any]],
|
||||
@ -1252,9 +1261,11 @@ class Database(common.BaseObject, Generic[_DocumentType]):
|
||||
if comment is not None:
|
||||
command["comment"] = comment
|
||||
|
||||
with self._client._conn_for_writes(session, operation=_Op.DROP) as connection:
|
||||
def inner(
|
||||
session: Optional[ClientSession], conn: Connection, _retryable_write: bool
|
||||
) -> dict[str, Any]:
|
||||
return self._command(
|
||||
connection,
|
||||
conn,
|
||||
command,
|
||||
allowable_errors=["ns not found", 26],
|
||||
write_concern=self._write_concern_for(session),
|
||||
@ -1262,8 +1273,9 @@ class Database(common.BaseObject, Generic[_DocumentType]):
|
||||
session=session,
|
||||
)
|
||||
|
||||
return self.client._retryable_write(False, inner, session, _Op.DROP)
|
||||
|
||||
@_csot.apply
|
||||
@_retry_overload
|
||||
def drop_collection(
|
||||
self,
|
||||
name_or_collection: Union[str, Collection[_DocumentTypeArg]],
|
||||
|
||||
@ -32,7 +32,6 @@ from typing import (
|
||||
from pymongo import _csot
|
||||
from pymongo.errors import (
|
||||
OperationFailure,
|
||||
PyMongoError,
|
||||
)
|
||||
from pymongo.helpers_shared import _REAUTHENTICATION_REQUIRED_CODE
|
||||
from pymongo.lock import _create_lock
|
||||
@ -166,34 +165,6 @@ class _RetryPolicy:
|
||||
return True
|
||||
|
||||
|
||||
def _retry_overload(func: F) -> F:
|
||||
@functools.wraps(func)
|
||||
def inner(self: Any, *args: Any, **kwargs: Any) -> Any:
|
||||
retry_policy = self._retry_policy
|
||||
attempt = 0
|
||||
while True:
|
||||
try:
|
||||
res = func(self, *args, **kwargs)
|
||||
retry_policy.record_success(retry=attempt > 0)
|
||||
return res
|
||||
except PyMongoError as exc:
|
||||
if not exc.has_error_label("RetryableError"):
|
||||
raise
|
||||
attempt += 1
|
||||
delay = 0
|
||||
if exc.has_error_label("SystemOverloadedError"):
|
||||
delay = retry_policy.backoff(attempt)
|
||||
if not retry_policy.should_retry(attempt, delay):
|
||||
raise
|
||||
|
||||
# Implement exponential backoff on retry.
|
||||
if delay:
|
||||
time.sleep(delay)
|
||||
continue
|
||||
|
||||
return cast(F, inner)
|
||||
|
||||
|
||||
def _getaddrinfo(
|
||||
host: Any, port: Any, **kwargs: Any
|
||||
) -> list[
|
||||
|
||||
@ -112,7 +112,6 @@ from pymongo.synchronous.client_bulk import _ClientBulk
|
||||
from pymongo.synchronous.client_session import _EmptyServerSession
|
||||
from pymongo.synchronous.command_cursor import CommandCursor
|
||||
from pymongo.synchronous.helpers import (
|
||||
_retry_overload,
|
||||
_RetryPolicy,
|
||||
_TokenBucket,
|
||||
)
|
||||
@ -2393,7 +2392,6 @@ class MongoClient(common.BaseObject, Generic[_DocumentType]):
|
||||
return [doc["name"] for doc in res]
|
||||
|
||||
@_csot.apply
|
||||
@_retry_overload
|
||||
def drop_database(
|
||||
self,
|
||||
name_or_database: Union[str, database.Database[_DocumentTypeArg]],
|
||||
@ -2436,15 +2434,13 @@ class MongoClient(common.BaseObject, Generic[_DocumentType]):
|
||||
f"name_or_database must be an instance of str or a Database, not {type(name)}"
|
||||
)
|
||||
|
||||
with self._conn_for_writes(session, operation=_Op.DROP_DATABASE) as conn:
|
||||
self[name]._command(
|
||||
conn,
|
||||
{"dropDatabase": 1, "comment": comment},
|
||||
read_preference=ReadPreference.PRIMARY,
|
||||
write_concern=self._write_concern_for(session),
|
||||
parse_write_concern_error=True,
|
||||
session=session,
|
||||
)
|
||||
self[name].command(
|
||||
{"dropDatabase": 1, "comment": comment},
|
||||
read_preference=ReadPreference.PRIMARY,
|
||||
write_concern=self._write_concern_for(session),
|
||||
parse_write_concern_error=True,
|
||||
session=session,
|
||||
)
|
||||
|
||||
@_csot.apply
|
||||
def bulk_write(
|
||||
@ -2771,6 +2767,11 @@ class _ClientConnectionRetryable(Generic[T]):
|
||||
try:
|
||||
res = self._read() if self._is_read else self._write()
|
||||
self._retry_policy.record_success(self._attempt_number > 0)
|
||||
# Track whether the transaction has completed a command.
|
||||
# If we need to apply backpressure to the first command,
|
||||
# we will need to revert back to starting state.
|
||||
if self._session is not None and self._session.in_transaction:
|
||||
self._session._transaction.has_completed_command = True
|
||||
return res
|
||||
except ServerSelectionTimeoutError:
|
||||
# The application may think the write was never attempted
|
||||
@ -2790,8 +2791,8 @@ class _ClientConnectionRetryable(Generic[T]):
|
||||
if isinstance(exc, (ConnectionFailure, OperationFailure)):
|
||||
# ConnectionFailures do not supply a code property
|
||||
exc_code = getattr(exc, "code", None)
|
||||
always_retryable = exc.has_error_label("RetryableError")
|
||||
overloaded = exc.has_error_label("SystemOverloadedError")
|
||||
always_retryable = exc.has_error_label("RetryableError") and overloaded
|
||||
if not always_retryable and (
|
||||
self._is_not_eligible_for_retry()
|
||||
or (
|
||||
@ -2803,6 +2804,18 @@ class _ClientConnectionRetryable(Generic[T]):
|
||||
self._retrying = True
|
||||
self._last_error = exc
|
||||
self._attempt_number += 1
|
||||
|
||||
# Revert back to starting state if we're in a transaction but haven't completed the first
|
||||
# command.
|
||||
if (
|
||||
overloaded
|
||||
and self._session is not None
|
||||
and self._session.in_transaction
|
||||
):
|
||||
transaction = self._session._transaction
|
||||
if not transaction.has_completed_command:
|
||||
transaction.set_starting()
|
||||
transaction.attempt = 0
|
||||
else:
|
||||
raise
|
||||
|
||||
@ -2813,8 +2826,8 @@ class _ClientConnectionRetryable(Generic[T]):
|
||||
):
|
||||
exc_to_check = exc.error
|
||||
retryable_write_label = exc_to_check.has_error_label("RetryableWriteError")
|
||||
always_retryable = exc_to_check.has_error_label("RetryableError")
|
||||
overloaded = exc_to_check.has_error_label("SystemOverloadedError")
|
||||
always_retryable = exc_to_check.has_error_label("RetryableError") and overloaded
|
||||
if not self._retryable and not always_retryable:
|
||||
raise
|
||||
if retryable_write_label or always_retryable:
|
||||
@ -2836,20 +2849,26 @@ class _ClientConnectionRetryable(Generic[T]):
|
||||
self._last_error = exc
|
||||
if self._last_error is None:
|
||||
self._last_error = exc
|
||||
# Revert back to starting state if we're in a transaction but haven't completed the first
|
||||
# command.
|
||||
if overloaded and self._session is not None and self._session.in_transaction:
|
||||
transaction = self._session._transaction
|
||||
if not transaction.has_completed_command:
|
||||
transaction.set_starting()
|
||||
transaction.attempt = 0
|
||||
|
||||
if self._server is not None:
|
||||
self._deprioritized_servers.append(self._server)
|
||||
|
||||
self._always_retryable = always_retryable
|
||||
if always_retryable:
|
||||
if overloaded:
|
||||
delay = self._retry_policy.backoff(self._attempt_number) if overloaded else 0
|
||||
if not self._retry_policy.should_retry(self._attempt_number, delay):
|
||||
if exc_to_check.has_error_label("NoWritesPerformed") and self._last_error:
|
||||
raise self._last_error from exc
|
||||
else:
|
||||
raise
|
||||
if overloaded:
|
||||
time.sleep(delay)
|
||||
time.sleep(delay)
|
||||
|
||||
def _is_not_eligible_for_retry(self) -> bool:
|
||||
"""Checks if the exchange is not eligible for retry"""
|
||||
|
||||
@ -254,6 +254,7 @@ class Connection:
|
||||
cmd = self.hello_cmd()
|
||||
performing_handshake = not self.performed_handshake
|
||||
awaitable = False
|
||||
cmd["backpressure"] = True
|
||||
if performing_handshake:
|
||||
self.performed_handshake = True
|
||||
cmd["client"] = self.opts.metadata
|
||||
|
||||
@ -15,10 +15,11 @@
|
||||
"""Test Client Backpressure spec."""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import os
|
||||
import pathlib
|
||||
import sys
|
||||
|
||||
import pymongo
|
||||
from time import perf_counter
|
||||
from unittest.mock import patch
|
||||
|
||||
sys.path[0:0] = [""]
|
||||
|
||||
@ -28,10 +29,13 @@ from test.asynchronous import (
|
||||
async_client_context,
|
||||
unittest,
|
||||
)
|
||||
from test.asynchronous.unified_format import generate_test_classes
|
||||
from test.utils_shared import EventListener, OvertCommandListener
|
||||
|
||||
import pymongo
|
||||
from pymongo.asynchronous import helpers
|
||||
from pymongo.asynchronous.helpers import _MAX_RETRIES, _RetryPolicy, _TokenBucket
|
||||
from pymongo.errors import PyMongoError
|
||||
from pymongo.errors import OperationFailure, PyMongoError
|
||||
|
||||
_IS_SYNC = False
|
||||
|
||||
@ -42,7 +46,7 @@ mock_overload_error = {
|
||||
"data": {
|
||||
"failCommands": ["find", "insert", "update"],
|
||||
"errorCode": 462, # IngressRequestRateLimitExceeded
|
||||
"errorLabels": ["RetryableError"],
|
||||
"errorLabels": ["RetryableError", "SystemOverloadedError"],
|
||||
},
|
||||
}
|
||||
|
||||
@ -68,6 +72,7 @@ class TestBackpressure(AsyncIntegrationTest):
|
||||
await self.db.command("find", "t")
|
||||
|
||||
self.assertIn("RetryableError", str(error.exception))
|
||||
self.assertIn("SystemOverloadedError", str(error.exception))
|
||||
|
||||
@async_client_context.require_failCommand_appName
|
||||
async def test_retry_overload_error_find(self):
|
||||
@ -87,6 +92,7 @@ class TestBackpressure(AsyncIntegrationTest):
|
||||
await self.db.t.find_one()
|
||||
|
||||
self.assertIn("RetryableError", str(error.exception))
|
||||
self.assertIn("SystemOverloadedError", str(error.exception))
|
||||
|
||||
@async_client_context.require_failCommand_appName
|
||||
async def test_retry_overload_error_insert_one(self):
|
||||
@ -106,6 +112,7 @@ class TestBackpressure(AsyncIntegrationTest):
|
||||
await self.db.t.find_one()
|
||||
|
||||
self.assertIn("RetryableError", str(error.exception))
|
||||
self.assertIn("SystemOverloadedError", str(error.exception))
|
||||
|
||||
@async_client_context.require_failCommand_appName
|
||||
async def test_retry_overload_error_update_many(self):
|
||||
@ -127,6 +134,7 @@ class TestBackpressure(AsyncIntegrationTest):
|
||||
await self.db.t.update_many({}, {"$set": {"x": 2}})
|
||||
|
||||
self.assertIn("RetryableError", str(error.exception))
|
||||
self.assertIn("SystemOverloadedError", str(error.exception))
|
||||
|
||||
@async_client_context.require_failCommand_appName
|
||||
async def test_retry_overload_error_getMore(self):
|
||||
@ -140,7 +148,7 @@ class TestBackpressure(AsyncIntegrationTest):
|
||||
"data": {
|
||||
"failCommands": ["getMore"],
|
||||
"errorCode": 462, # IngressRequestRateLimitExceeded
|
||||
"errorLabels": ["RetryableError"],
|
||||
"errorLabels": ["RetryableError", "SystemOverloadedError"],
|
||||
},
|
||||
}
|
||||
cursor = coll.find(batch_size=2)
|
||||
@ -158,6 +166,7 @@ class TestBackpressure(AsyncIntegrationTest):
|
||||
await cursor.to_list()
|
||||
|
||||
self.assertIn("RetryableError", str(error.exception))
|
||||
self.assertIn("SystemOverloadedError", str(error.exception))
|
||||
|
||||
@async_client_context.require_failCommand_appName
|
||||
async def test_limit_retry_command(self):
|
||||
@ -180,6 +189,7 @@ class TestBackpressure(AsyncIntegrationTest):
|
||||
await db.command("find", "t")
|
||||
|
||||
self.assertIn("RetryableError", str(error.exception))
|
||||
self.assertIn("SystemOverloadedError", str(error.exception))
|
||||
|
||||
|
||||
class TestRetryPolicy(AsyncPyMongoTestCase):
|
||||
@ -226,5 +236,83 @@ class TestRetryPolicy(AsyncPyMongoTestCase):
|
||||
self.assertTrue(await retry_policy.should_retry(1, 1.0))
|
||||
|
||||
|
||||
# Prose tests.
|
||||
class AsyncTestClientBackpressure(AsyncIntegrationTest):
|
||||
listener: EventListener
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls) -> None:
|
||||
cls.listener = OvertCommandListener()
|
||||
|
||||
@async_client_context.require_connection
|
||||
async def asyncSetUp(self) -> None:
|
||||
await super().asyncSetUp()
|
||||
self.listener.reset()
|
||||
self.app_name = self.__class__.__name__.lower()
|
||||
self.client = await self.async_rs_or_single_client(
|
||||
event_listeners=[self.listener], retryWrites=False, appName=self.app_name
|
||||
)
|
||||
|
||||
@patch("random.random")
|
||||
@async_client_context.require_failCommand_appName
|
||||
async def test_01_operation_retry_uses_exponential_backoff(self, random_func):
|
||||
# Drivers should test that retries do not occur immediately when a SystemOverloadedError is encountered.
|
||||
|
||||
# 1. let `client` be a `MongoClient`
|
||||
client = self.client
|
||||
|
||||
# 2. let `collection` be a collection
|
||||
collection = client.test.test
|
||||
|
||||
# 3. Now, run transactions without backoff:
|
||||
|
||||
# a. Configure the random number generator used for jitter to always return `0` -- this effectively disables backoff.
|
||||
random_func.return_value = 0
|
||||
|
||||
# b. Configure the following failPoint:
|
||||
fail_point = dict(
|
||||
mode="alwaysOn",
|
||||
data=dict(
|
||||
failCommands=["insert"],
|
||||
errorCode=2,
|
||||
errorLabels=["SystemOverloadedError", "RetryableError"],
|
||||
appName=self.app_name,
|
||||
),
|
||||
)
|
||||
async with self.fail_point(fail_point):
|
||||
# c. Execute the following command. Expect that the command errors. Measure the duration of the command execution.
|
||||
start0 = perf_counter()
|
||||
with self.assertRaises(OperationFailure):
|
||||
await collection.insert_one({"a": 1})
|
||||
end0 = perf_counter()
|
||||
|
||||
# d. Configure the random number generator used for jitter to always return `1`.
|
||||
random_func.return_value = 1
|
||||
|
||||
# e. Execute step c again.
|
||||
start1 = perf_counter()
|
||||
with self.assertRaises(OperationFailure):
|
||||
await collection.insert_one({"a": 1})
|
||||
end1 = perf_counter()
|
||||
|
||||
# f. Compare the two time between the two runs.
|
||||
# The sum of 5 backoffs is 3.1 seconds. There is a 1-second window to account for potential variance between the two
|
||||
# runs.
|
||||
self.assertTrue(abs((end1 - start1) - (end0 - start0 + 3.1)) < 1)
|
||||
|
||||
|
||||
# Location of JSON test specifications.
|
||||
if _IS_SYNC:
|
||||
_TEST_PATH = os.path.join(pathlib.Path(__file__).resolve().parent, "client-backpressure")
|
||||
else:
|
||||
_TEST_PATH = os.path.join(pathlib.Path(__file__).resolve().parent.parent, "client-backpressure")
|
||||
|
||||
globals().update(
|
||||
generate_test_classes(
|
||||
_TEST_PATH,
|
||||
module=__name__,
|
||||
)
|
||||
)
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@ -219,6 +219,19 @@ class TestClientMetadataProse(AsyncIntegrationTest):
|
||||
# add same metadata again
|
||||
await self.check_metadata_added(client, "Framework", None, None)
|
||||
|
||||
async def test_handshake_documents_include_backpressure(self):
|
||||
# Create a `MongoClient` that is configured to record all handshake documents sent to the server as a part of
|
||||
# connection establishment.
|
||||
client = await self.async_rs_or_single_client("mongodb://" + self.server.address_string)
|
||||
|
||||
# Send a `ping` command to the server and verify that the command succeeds. This ensure that a connection is
|
||||
# established on all topologies. Note: MockupDB only supports standalone servers.
|
||||
await client.admin.command("ping")
|
||||
|
||||
# Assert that for every handshake document intercepted:
|
||||
# the document has a field `backpressure` whose value is `true`.
|
||||
self.assertEqual(self.handshake_req["backpressure"], True)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
2759
test/client-backpressure/backpressure-retry-loop.json
Normal file
2759
test/client-backpressure/backpressure-retry-loop.json
Normal file
File diff suppressed because it is too large
Load Diff
3448
test/client-backpressure/backpressure-retry-max-attempts.json
Normal file
3448
test/client-backpressure/backpressure-retry-max-attempts.json
Normal file
File diff suppressed because it is too large
Load Diff
291
test/client-backpressure/getMore-retried.json
Normal file
291
test/client-backpressure/getMore-retried.json
Normal file
@ -0,0 +1,291 @@
|
||||
{
|
||||
"description": "getMore-retries-backpressure",
|
||||
"schemaVersion": "1.3",
|
||||
"runOnRequirements": [
|
||||
{
|
||||
"minServerVersion": "4.4"
|
||||
}
|
||||
],
|
||||
"createEntities": [
|
||||
{
|
||||
"client": {
|
||||
"id": "client0",
|
||||
"observeEvents": [
|
||||
"commandStartedEvent",
|
||||
"commandFailedEvent",
|
||||
"commandSucceededEvent"
|
||||
]
|
||||
}
|
||||
},
|
||||
{
|
||||
"client": {
|
||||
"id": "failPointClient"
|
||||
}
|
||||
},
|
||||
{
|
||||
"database": {
|
||||
"id": "db",
|
||||
"client": "client0",
|
||||
"databaseName": "default"
|
||||
}
|
||||
},
|
||||
{
|
||||
"collection": {
|
||||
"id": "coll",
|
||||
"database": "db",
|
||||
"collectionName": "default"
|
||||
}
|
||||
}
|
||||
],
|
||||
"initialData": [
|
||||
{
|
||||
"databaseName": "default",
|
||||
"collectionName": "default",
|
||||
"documents": [
|
||||
{
|
||||
"a": 1
|
||||
},
|
||||
{
|
||||
"a": 2
|
||||
},
|
||||
{
|
||||
"a": 3
|
||||
}
|
||||
]
|
||||
}
|
||||
],
|
||||
"tests": [
|
||||
{
|
||||
"description": "getMores are retried",
|
||||
"operations": [
|
||||
{
|
||||
"name": "failPoint",
|
||||
"object": "testRunner",
|
||||
"arguments": {
|
||||
"client": "failPointClient",
|
||||
"failPoint": {
|
||||
"configureFailPoint": "failCommand",
|
||||
"mode": {
|
||||
"times": 3
|
||||
},
|
||||
"data": {
|
||||
"failCommands": [
|
||||
"getMore"
|
||||
],
|
||||
"errorLabels": [
|
||||
"RetryableError",
|
||||
"SystemOverloadedError"
|
||||
],
|
||||
"errorCode": 2
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "find",
|
||||
"arguments": {
|
||||
"batchSize": 2,
|
||||
"filter": {},
|
||||
"sort": {
|
||||
"a": 1
|
||||
}
|
||||
},
|
||||
"object": "coll",
|
||||
"expectResult": [
|
||||
{
|
||||
"a": 1
|
||||
},
|
||||
{
|
||||
"a": 2
|
||||
},
|
||||
{
|
||||
"a": 3
|
||||
}
|
||||
]
|
||||
}
|
||||
],
|
||||
"expectEvents": [
|
||||
{
|
||||
"client": "client0",
|
||||
"events": [
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"commandName": "find"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandSucceededEvent": {
|
||||
"commandName": "find"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"commandName": "getMore"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandFailedEvent": {
|
||||
"commandName": "getMore"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"commandName": "getMore"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandFailedEvent": {
|
||||
"commandName": "getMore"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"commandName": "getMore"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandFailedEvent": {
|
||||
"commandName": "getMore"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"commandName": "getMore"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandSucceededEvent": {
|
||||
"commandName": "getMore"
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"description": "getMores are retried maxAttempts=5 times",
|
||||
"operations": [
|
||||
{
|
||||
"name": "failPoint",
|
||||
"object": "testRunner",
|
||||
"arguments": {
|
||||
"client": "failPointClient",
|
||||
"failPoint": {
|
||||
"configureFailPoint": "failCommand",
|
||||
"mode": "alwaysOn",
|
||||
"data": {
|
||||
"failCommands": [
|
||||
"getMore"
|
||||
],
|
||||
"errorLabels": [
|
||||
"RetryableError",
|
||||
"SystemOverloadedError"
|
||||
],
|
||||
"errorCode": 2
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "find",
|
||||
"arguments": {
|
||||
"batchSize": 2,
|
||||
"filter": {}
|
||||
},
|
||||
"object": "coll",
|
||||
"expectError": {
|
||||
"isError": true,
|
||||
"isClientError": false
|
||||
}
|
||||
}
|
||||
],
|
||||
"expectEvents": [
|
||||
{
|
||||
"client": "client0",
|
||||
"events": [
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"commandName": "find"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandSucceededEvent": {
|
||||
"commandName": "find"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"commandName": "getMore"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandFailedEvent": {
|
||||
"commandName": "getMore"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"commandName": "getMore"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandFailedEvent": {
|
||||
"commandName": "getMore"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"commandName": "getMore"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandFailedEvent": {
|
||||
"commandName": "getMore"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"commandName": "getMore"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandFailedEvent": {
|
||||
"commandName": "getMore"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"commandName": "getMore"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandFailedEvent": {
|
||||
"commandName": "getMore"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"commandName": "getMore"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandFailedEvent": {
|
||||
"commandName": "getMore"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"commandName": "killCursors"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandSucceededEvent": {
|
||||
"commandName": "killCursors"
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
@ -15,10 +15,11 @@
|
||||
"""Test Client Backpressure spec."""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import os
|
||||
import pathlib
|
||||
import sys
|
||||
|
||||
import pymongo
|
||||
from time import perf_counter
|
||||
from unittest.mock import patch
|
||||
|
||||
sys.path[0:0] = [""]
|
||||
|
||||
@ -28,8 +29,11 @@ from test import (
|
||||
client_context,
|
||||
unittest,
|
||||
)
|
||||
from test.unified_format import generate_test_classes
|
||||
from test.utils_shared import EventListener, OvertCommandListener
|
||||
|
||||
from pymongo.errors import PyMongoError
|
||||
import pymongo
|
||||
from pymongo.errors import OperationFailure, PyMongoError
|
||||
from pymongo.synchronous import helpers
|
||||
from pymongo.synchronous.helpers import _MAX_RETRIES, _RetryPolicy, _TokenBucket
|
||||
|
||||
@ -42,7 +46,7 @@ mock_overload_error = {
|
||||
"data": {
|
||||
"failCommands": ["find", "insert", "update"],
|
||||
"errorCode": 462, # IngressRequestRateLimitExceeded
|
||||
"errorLabels": ["RetryableError"],
|
||||
"errorLabels": ["RetryableError", "SystemOverloadedError"],
|
||||
},
|
||||
}
|
||||
|
||||
@ -68,6 +72,7 @@ class TestBackpressure(IntegrationTest):
|
||||
self.db.command("find", "t")
|
||||
|
||||
self.assertIn("RetryableError", str(error.exception))
|
||||
self.assertIn("SystemOverloadedError", str(error.exception))
|
||||
|
||||
@client_context.require_failCommand_appName
|
||||
def test_retry_overload_error_find(self):
|
||||
@ -87,6 +92,7 @@ class TestBackpressure(IntegrationTest):
|
||||
self.db.t.find_one()
|
||||
|
||||
self.assertIn("RetryableError", str(error.exception))
|
||||
self.assertIn("SystemOverloadedError", str(error.exception))
|
||||
|
||||
@client_context.require_failCommand_appName
|
||||
def test_retry_overload_error_insert_one(self):
|
||||
@ -106,6 +112,7 @@ class TestBackpressure(IntegrationTest):
|
||||
self.db.t.find_one()
|
||||
|
||||
self.assertIn("RetryableError", str(error.exception))
|
||||
self.assertIn("SystemOverloadedError", str(error.exception))
|
||||
|
||||
@client_context.require_failCommand_appName
|
||||
def test_retry_overload_error_update_many(self):
|
||||
@ -127,6 +134,7 @@ class TestBackpressure(IntegrationTest):
|
||||
self.db.t.update_many({}, {"$set": {"x": 2}})
|
||||
|
||||
self.assertIn("RetryableError", str(error.exception))
|
||||
self.assertIn("SystemOverloadedError", str(error.exception))
|
||||
|
||||
@client_context.require_failCommand_appName
|
||||
def test_retry_overload_error_getMore(self):
|
||||
@ -140,7 +148,7 @@ class TestBackpressure(IntegrationTest):
|
||||
"data": {
|
||||
"failCommands": ["getMore"],
|
||||
"errorCode": 462, # IngressRequestRateLimitExceeded
|
||||
"errorLabels": ["RetryableError"],
|
||||
"errorLabels": ["RetryableError", "SystemOverloadedError"],
|
||||
},
|
||||
}
|
||||
cursor = coll.find(batch_size=2)
|
||||
@ -158,6 +166,7 @@ class TestBackpressure(IntegrationTest):
|
||||
cursor.to_list()
|
||||
|
||||
self.assertIn("RetryableError", str(error.exception))
|
||||
self.assertIn("SystemOverloadedError", str(error.exception))
|
||||
|
||||
@client_context.require_failCommand_appName
|
||||
def test_limit_retry_command(self):
|
||||
@ -180,6 +189,7 @@ class TestBackpressure(IntegrationTest):
|
||||
db.command("find", "t")
|
||||
|
||||
self.assertIn("RetryableError", str(error.exception))
|
||||
self.assertIn("SystemOverloadedError", str(error.exception))
|
||||
|
||||
|
||||
class TestRetryPolicy(PyMongoTestCase):
|
||||
@ -226,5 +236,83 @@ class TestRetryPolicy(PyMongoTestCase):
|
||||
self.assertTrue(retry_policy.should_retry(1, 1.0))
|
||||
|
||||
|
||||
# Prose tests.
|
||||
class TestClientBackpressure(IntegrationTest):
|
||||
listener: EventListener
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls) -> None:
|
||||
cls.listener = OvertCommandListener()
|
||||
|
||||
@client_context.require_connection
|
||||
def setUp(self) -> None:
|
||||
super().setUp()
|
||||
self.listener.reset()
|
||||
self.app_name = self.__class__.__name__.lower()
|
||||
self.client = self.rs_or_single_client(
|
||||
event_listeners=[self.listener], retryWrites=False, appName=self.app_name
|
||||
)
|
||||
|
||||
@patch("random.random")
|
||||
@client_context.require_failCommand_appName
|
||||
def test_01_operation_retry_uses_exponential_backoff(self, random_func):
|
||||
# Drivers should test that retries do not occur immediately when a SystemOverloadedError is encountered.
|
||||
|
||||
# 1. let `client` be a `MongoClient`
|
||||
client = self.client
|
||||
|
||||
# 2. let `collection` be a collection
|
||||
collection = client.test.test
|
||||
|
||||
# 3. Now, run transactions without backoff:
|
||||
|
||||
# a. Configure the random number generator used for jitter to always return `0` -- this effectively disables backoff.
|
||||
random_func.return_value = 0
|
||||
|
||||
# b. Configure the following failPoint:
|
||||
fail_point = dict(
|
||||
mode="alwaysOn",
|
||||
data=dict(
|
||||
failCommands=["insert"],
|
||||
errorCode=2,
|
||||
errorLabels=["SystemOverloadedError", "RetryableError"],
|
||||
appName=self.app_name,
|
||||
),
|
||||
)
|
||||
with self.fail_point(fail_point):
|
||||
# c. Execute the following command. Expect that the command errors. Measure the duration of the command execution.
|
||||
start0 = perf_counter()
|
||||
with self.assertRaises(OperationFailure):
|
||||
collection.insert_one({"a": 1})
|
||||
end0 = perf_counter()
|
||||
|
||||
# d. Configure the random number generator used for jitter to always return `1`.
|
||||
random_func.return_value = 1
|
||||
|
||||
# e. Execute step c again.
|
||||
start1 = perf_counter()
|
||||
with self.assertRaises(OperationFailure):
|
||||
collection.insert_one({"a": 1})
|
||||
end1 = perf_counter()
|
||||
|
||||
# f. Compare the two time between the two runs.
|
||||
# The sum of 5 backoffs is 3.1 seconds. There is a 1-second window to account for potential variance between the two
|
||||
# runs.
|
||||
self.assertTrue(abs((end1 - start1) - (end0 - start0 + 3.1)) < 1)
|
||||
|
||||
|
||||
# Location of JSON test specifications.
|
||||
if _IS_SYNC:
|
||||
_TEST_PATH = os.path.join(pathlib.Path(__file__).resolve().parent, "client-backpressure")
|
||||
else:
|
||||
_TEST_PATH = os.path.join(pathlib.Path(__file__).resolve().parent.parent, "client-backpressure")
|
||||
|
||||
globals().update(
|
||||
generate_test_classes(
|
||||
_TEST_PATH,
|
||||
module=__name__,
|
||||
)
|
||||
)
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@ -219,6 +219,19 @@ class TestClientMetadataProse(IntegrationTest):
|
||||
# add same metadata again
|
||||
self.check_metadata_added(client, "Framework", None, None)
|
||||
|
||||
def test_handshake_documents_include_backpressure(self):
|
||||
# Create a `MongoClient` that is configured to record all handshake documents sent to the server as a part of
|
||||
# connection establishment.
|
||||
client = self.rs_or_single_client("mongodb://" + self.server.address_string)
|
||||
|
||||
# Send a `ping` command to the server and verify that the command succeeds. This ensure that a connection is
|
||||
# established on all topologies. Note: MockupDB only supports standalone servers.
|
||||
client.admin.command("ping")
|
||||
|
||||
# Assert that for every handshake document intercepted:
|
||||
# the document has a field `backpressure` whose value is `true`.
|
||||
self.assertEqual(self.handshake_req["backpressure"], True)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
357
test/transactions/unified/backpressure-retryable-abort.json
Normal file
357
test/transactions/unified/backpressure-retryable-abort.json
Normal file
@ -0,0 +1,357 @@
|
||||
{
|
||||
"description": "backpressure-retryable-abort",
|
||||
"schemaVersion": "1.3",
|
||||
"runOnRequirements": [
|
||||
{
|
||||
"minServerVersion": "4.4",
|
||||
"topologies": [
|
||||
"replicaset",
|
||||
"sharded",
|
||||
"load-balanced"
|
||||
]
|
||||
}
|
||||
],
|
||||
"createEntities": [
|
||||
{
|
||||
"client": {
|
||||
"id": "client0",
|
||||
"useMultipleMongoses": false,
|
||||
"observeEvents": [
|
||||
"commandStartedEvent"
|
||||
]
|
||||
}
|
||||
},
|
||||
{
|
||||
"database": {
|
||||
"id": "database0",
|
||||
"client": "client0",
|
||||
"databaseName": "transaction-tests"
|
||||
}
|
||||
},
|
||||
{
|
||||
"collection": {
|
||||
"id": "collection0",
|
||||
"database": "database0",
|
||||
"collectionName": "test"
|
||||
}
|
||||
},
|
||||
{
|
||||
"session": {
|
||||
"id": "session0",
|
||||
"client": "client0"
|
||||
}
|
||||
}
|
||||
],
|
||||
"initialData": [
|
||||
{
|
||||
"collectionName": "test",
|
||||
"databaseName": "transaction-tests",
|
||||
"documents": []
|
||||
}
|
||||
],
|
||||
"tests": [
|
||||
{
|
||||
"description": "abortTransaction retries if backpressure labels are added",
|
||||
"operations": [
|
||||
{
|
||||
"object": "testRunner",
|
||||
"name": "failPoint",
|
||||
"arguments": {
|
||||
"client": "client0",
|
||||
"failPoint": {
|
||||
"configureFailPoint": "failCommand",
|
||||
"mode": {
|
||||
"times": 2
|
||||
},
|
||||
"data": {
|
||||
"failCommands": [
|
||||
"abortTransaction"
|
||||
],
|
||||
"errorLabels": [
|
||||
"RetryableError",
|
||||
"SystemOverloadedError"
|
||||
],
|
||||
"errorCode": 112
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
"object": "session0",
|
||||
"name": "startTransaction"
|
||||
},
|
||||
{
|
||||
"object": "collection0",
|
||||
"name": "insertOne",
|
||||
"arguments": {
|
||||
"session": "session0",
|
||||
"document": {
|
||||
"_id": 1
|
||||
}
|
||||
},
|
||||
"expectResult": {
|
||||
"$$unsetOrMatches": {
|
||||
"insertedId": {
|
||||
"$$unsetOrMatches": 1
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
"object": "session0",
|
||||
"name": "abortTransaction"
|
||||
}
|
||||
],
|
||||
"expectEvents": [
|
||||
{
|
||||
"client": "client0",
|
||||
"events": [
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"command": {
|
||||
"insert": "test",
|
||||
"documents": [
|
||||
{
|
||||
"_id": 1
|
||||
}
|
||||
],
|
||||
"ordered": true,
|
||||
"readConcern": {
|
||||
"$$exists": false
|
||||
},
|
||||
"lsid": {
|
||||
"$$sessionLsid": "session0"
|
||||
},
|
||||
"txnNumber": {
|
||||
"$numberLong": "1"
|
||||
},
|
||||
"startTransaction": true,
|
||||
"autocommit": false,
|
||||
"writeConcern": {
|
||||
"$$exists": false
|
||||
}
|
||||
},
|
||||
"commandName": "insert",
|
||||
"databaseName": "transaction-tests"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"command": {
|
||||
"abortTransaction": 1,
|
||||
"lsid": {
|
||||
"$$sessionLsid": "session0"
|
||||
},
|
||||
"txnNumber": {
|
||||
"$numberLong": "1"
|
||||
},
|
||||
"startTransaction": {
|
||||
"$$exists": false
|
||||
},
|
||||
"autocommit": false,
|
||||
"writeConcern": {
|
||||
"$$exists": false
|
||||
}
|
||||
},
|
||||
"commandName": "abortTransaction",
|
||||
"databaseName": "admin"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"command": {
|
||||
"abortTransaction": 1,
|
||||
"lsid": {
|
||||
"$$sessionLsid": "session0"
|
||||
},
|
||||
"txnNumber": {
|
||||
"$numberLong": "1"
|
||||
},
|
||||
"startTransaction": {
|
||||
"$$exists": false
|
||||
},
|
||||
"autocommit": false,
|
||||
"writeConcern": {
|
||||
"$$exists": false
|
||||
}
|
||||
},
|
||||
"commandName": "abortTransaction",
|
||||
"databaseName": "admin"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"command": {
|
||||
"abortTransaction": 1,
|
||||
"lsid": {
|
||||
"$$sessionLsid": "session0"
|
||||
},
|
||||
"txnNumber": {
|
||||
"$numberLong": "1"
|
||||
},
|
||||
"startTransaction": {
|
||||
"$$exists": false
|
||||
},
|
||||
"autocommit": false,
|
||||
"writeConcern": {
|
||||
"$$exists": false
|
||||
}
|
||||
},
|
||||
"commandName": "abortTransaction",
|
||||
"databaseName": "admin"
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
],
|
||||
"outcome": [
|
||||
{
|
||||
"collectionName": "test",
|
||||
"databaseName": "transaction-tests",
|
||||
"documents": []
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"description": "abortTransaction is retried maxAttempts=5 times if backpressure labels are added",
|
||||
"operations": [
|
||||
{
|
||||
"object": "testRunner",
|
||||
"name": "failPoint",
|
||||
"arguments": {
|
||||
"client": "client0",
|
||||
"failPoint": {
|
||||
"configureFailPoint": "failCommand",
|
||||
"mode": "alwaysOn",
|
||||
"data": {
|
||||
"failCommands": [
|
||||
"abortTransaction"
|
||||
],
|
||||
"errorLabels": [
|
||||
"RetryableError",
|
||||
"SystemOverloadedError"
|
||||
],
|
||||
"errorCode": 112
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
"object": "session0",
|
||||
"name": "startTransaction"
|
||||
},
|
||||
{
|
||||
"object": "collection0",
|
||||
"name": "insertOne",
|
||||
"arguments": {
|
||||
"session": "session0",
|
||||
"document": {
|
||||
"_id": 1
|
||||
}
|
||||
},
|
||||
"expectResult": {
|
||||
"$$unsetOrMatches": {
|
||||
"insertedId": {
|
||||
"$$unsetOrMatches": 1
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
"object": "session0",
|
||||
"name": "abortTransaction"
|
||||
}
|
||||
],
|
||||
"expectEvents": [
|
||||
{
|
||||
"client": "client0",
|
||||
"events": [
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"command": {
|
||||
"insert": "test",
|
||||
"documents": [
|
||||
{
|
||||
"_id": 1
|
||||
}
|
||||
],
|
||||
"ordered": true,
|
||||
"readConcern": {
|
||||
"$$exists": false
|
||||
},
|
||||
"lsid": {
|
||||
"$$sessionLsid": "session0"
|
||||
},
|
||||
"txnNumber": {
|
||||
"$numberLong": "1"
|
||||
},
|
||||
"startTransaction": true,
|
||||
"autocommit": false,
|
||||
"writeConcern": {
|
||||
"$$exists": false
|
||||
}
|
||||
},
|
||||
"commandName": "insert",
|
||||
"databaseName": "transaction-tests"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"command": {
|
||||
"abortTransaction": 1,
|
||||
"lsid": {
|
||||
"$$sessionLsid": "session0"
|
||||
},
|
||||
"txnNumber": {
|
||||
"$numberLong": "1"
|
||||
},
|
||||
"startTransaction": {
|
||||
"$$exists": false
|
||||
},
|
||||
"autocommit": false,
|
||||
"writeConcern": {
|
||||
"$$exists": false
|
||||
}
|
||||
},
|
||||
"commandName": "abortTransaction",
|
||||
"databaseName": "admin"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"commandName": "abortTransaction"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"commandName": "abortTransaction"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"commandName": "abortTransaction"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"commandName": "abortTransaction"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"commandName": "abortTransaction"
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
],
|
||||
"outcome": [
|
||||
{
|
||||
"collectionName": "test",
|
||||
"databaseName": "transaction-tests",
|
||||
"documents": []
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
374
test/transactions/unified/backpressure-retryable-commit.json
Normal file
374
test/transactions/unified/backpressure-retryable-commit.json
Normal file
@ -0,0 +1,374 @@
|
||||
{
|
||||
"description": "backpressure-retryable-commit",
|
||||
"schemaVersion": "1.4",
|
||||
"runOnRequirements": [
|
||||
{
|
||||
"minServerVersion": "4.4",
|
||||
"topologies": [
|
||||
"sharded",
|
||||
"replicaset",
|
||||
"load-balanced"
|
||||
]
|
||||
}
|
||||
],
|
||||
"createEntities": [
|
||||
{
|
||||
"client": {
|
||||
"id": "client0",
|
||||
"useMultipleMongoses": false,
|
||||
"observeEvents": [
|
||||
"commandStartedEvent"
|
||||
]
|
||||
}
|
||||
},
|
||||
{
|
||||
"database": {
|
||||
"id": "database0",
|
||||
"client": "client0",
|
||||
"databaseName": "transaction-tests"
|
||||
}
|
||||
},
|
||||
{
|
||||
"collection": {
|
||||
"id": "collection0",
|
||||
"database": "database0",
|
||||
"collectionName": "test"
|
||||
}
|
||||
},
|
||||
{
|
||||
"session": {
|
||||
"id": "session0",
|
||||
"client": "client0"
|
||||
}
|
||||
}
|
||||
],
|
||||
"initialData": [
|
||||
{
|
||||
"collectionName": "test",
|
||||
"databaseName": "transaction-tests",
|
||||
"documents": []
|
||||
}
|
||||
],
|
||||
"tests": [
|
||||
{
|
||||
"description": "commitTransaction retries if backpressure labels are added",
|
||||
"runOnRequirements": [
|
||||
{
|
||||
"serverless": "forbid"
|
||||
}
|
||||
],
|
||||
"operations": [
|
||||
{
|
||||
"object": "testRunner",
|
||||
"name": "failPoint",
|
||||
"arguments": {
|
||||
"client": "client0",
|
||||
"failPoint": {
|
||||
"configureFailPoint": "failCommand",
|
||||
"mode": {
|
||||
"times": 2
|
||||
},
|
||||
"data": {
|
||||
"failCommands": [
|
||||
"commitTransaction"
|
||||
],
|
||||
"errorLabels": [
|
||||
"RetryableError",
|
||||
"SystemOverloadedError"
|
||||
],
|
||||
"errorCode": 112
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
"object": "session0",
|
||||
"name": "startTransaction"
|
||||
},
|
||||
{
|
||||
"object": "collection0",
|
||||
"name": "insertOne",
|
||||
"arguments": {
|
||||
"session": "session0",
|
||||
"document": {
|
||||
"_id": 1
|
||||
}
|
||||
},
|
||||
"expectResult": {
|
||||
"$$unsetOrMatches": {
|
||||
"insertedId": {
|
||||
"$$unsetOrMatches": 1
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
"object": "session0",
|
||||
"name": "commitTransaction"
|
||||
}
|
||||
],
|
||||
"expectEvents": [
|
||||
{
|
||||
"client": "client0",
|
||||
"events": [
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"command": {
|
||||
"insert": "test",
|
||||
"documents": [
|
||||
{
|
||||
"_id": 1
|
||||
}
|
||||
],
|
||||
"ordered": true,
|
||||
"readConcern": {
|
||||
"$$exists": false
|
||||
},
|
||||
"lsid": {
|
||||
"$$sessionLsid": "session0"
|
||||
},
|
||||
"txnNumber": {
|
||||
"$numberLong": "1"
|
||||
},
|
||||
"startTransaction": true,
|
||||
"autocommit": false,
|
||||
"writeConcern": {
|
||||
"$$exists": false
|
||||
}
|
||||
},
|
||||
"commandName": "insert",
|
||||
"databaseName": "transaction-tests"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"command": {
|
||||
"commitTransaction": 1,
|
||||
"lsid": {
|
||||
"$$sessionLsid": "session0"
|
||||
},
|
||||
"txnNumber": {
|
||||
"$numberLong": "1"
|
||||
},
|
||||
"startTransaction": {
|
||||
"$$exists": false
|
||||
},
|
||||
"autocommit": false,
|
||||
"writeConcern": {
|
||||
"$$exists": false
|
||||
}
|
||||
},
|
||||
"commandName": "commitTransaction",
|
||||
"databaseName": "admin"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"command": {
|
||||
"commitTransaction": 1,
|
||||
"lsid": {
|
||||
"$$sessionLsid": "session0"
|
||||
},
|
||||
"txnNumber": {
|
||||
"$numberLong": "1"
|
||||
},
|
||||
"startTransaction": {
|
||||
"$$exists": false
|
||||
},
|
||||
"autocommit": false,
|
||||
"writeConcern": {
|
||||
"$$exists": false
|
||||
}
|
||||
},
|
||||
"commandName": "commitTransaction",
|
||||
"databaseName": "admin"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"command": {
|
||||
"commitTransaction": 1,
|
||||
"lsid": {
|
||||
"$$sessionLsid": "session0"
|
||||
},
|
||||
"txnNumber": {
|
||||
"$numberLong": "1"
|
||||
},
|
||||
"startTransaction": {
|
||||
"$$exists": false
|
||||
},
|
||||
"autocommit": false,
|
||||
"writeConcern": {
|
||||
"$$exists": false
|
||||
}
|
||||
},
|
||||
"commandName": "commitTransaction",
|
||||
"databaseName": "admin"
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
],
|
||||
"outcome": [
|
||||
{
|
||||
"collectionName": "test",
|
||||
"databaseName": "transaction-tests",
|
||||
"documents": [
|
||||
{
|
||||
"_id": 1
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"description": "commitTransaction is retried maxAttempts=5 times if backpressure labels are added",
|
||||
"runOnRequirements": [
|
||||
{
|
||||
"serverless": "forbid"
|
||||
}
|
||||
],
|
||||
"operations": [
|
||||
{
|
||||
"object": "testRunner",
|
||||
"name": "failPoint",
|
||||
"arguments": {
|
||||
"client": "client0",
|
||||
"failPoint": {
|
||||
"configureFailPoint": "failCommand",
|
||||
"mode": "alwaysOn",
|
||||
"data": {
|
||||
"failCommands": [
|
||||
"commitTransaction"
|
||||
],
|
||||
"errorLabels": [
|
||||
"RetryableError",
|
||||
"SystemOverloadedError"
|
||||
],
|
||||
"errorCode": 112
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
"object": "session0",
|
||||
"name": "startTransaction"
|
||||
},
|
||||
{
|
||||
"object": "collection0",
|
||||
"name": "insertOne",
|
||||
"arguments": {
|
||||
"session": "session0",
|
||||
"document": {
|
||||
"_id": 1
|
||||
}
|
||||
},
|
||||
"expectResult": {
|
||||
"$$unsetOrMatches": {
|
||||
"insertedId": {
|
||||
"$$unsetOrMatches": 1
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
"object": "session0",
|
||||
"name": "commitTransaction",
|
||||
"expectError": {
|
||||
"isError": true
|
||||
}
|
||||
}
|
||||
],
|
||||
"expectEvents": [
|
||||
{
|
||||
"client": "client0",
|
||||
"events": [
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"command": {
|
||||
"insert": "test",
|
||||
"documents": [
|
||||
{
|
||||
"_id": 1
|
||||
}
|
||||
],
|
||||
"ordered": true,
|
||||
"readConcern": {
|
||||
"$$exists": false
|
||||
},
|
||||
"lsid": {
|
||||
"$$sessionLsid": "session0"
|
||||
},
|
||||
"txnNumber": {
|
||||
"$numberLong": "1"
|
||||
},
|
||||
"startTransaction": true,
|
||||
"autocommit": false,
|
||||
"writeConcern": {
|
||||
"$$exists": false
|
||||
}
|
||||
},
|
||||
"commandName": "insert",
|
||||
"databaseName": "transaction-tests"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"command": {
|
||||
"commitTransaction": 1,
|
||||
"lsid": {
|
||||
"$$sessionLsid": "session0"
|
||||
},
|
||||
"txnNumber": {
|
||||
"$numberLong": "1"
|
||||
},
|
||||
"startTransaction": {
|
||||
"$$exists": false
|
||||
},
|
||||
"autocommit": false,
|
||||
"writeConcern": {
|
||||
"$$exists": false
|
||||
}
|
||||
},
|
||||
"commandName": "commitTransaction",
|
||||
"databaseName": "admin"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"commandName": "commitTransaction"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"commandName": "commitTransaction"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"commandName": "commitTransaction"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"commandName": "commitTransaction"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"commandName": "commitTransaction"
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
],
|
||||
"outcome": [
|
||||
{
|
||||
"collectionName": "test",
|
||||
"databaseName": "transaction-tests",
|
||||
"documents": []
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
328
test/transactions/unified/backpressure-retryable-reads.json
Normal file
328
test/transactions/unified/backpressure-retryable-reads.json
Normal file
@ -0,0 +1,328 @@
|
||||
{
|
||||
"description": "backpressure-retryable-reads",
|
||||
"schemaVersion": "1.3",
|
||||
"runOnRequirements": [
|
||||
{
|
||||
"minServerVersion": "4.4",
|
||||
"topologies": [
|
||||
"replicaset",
|
||||
"sharded",
|
||||
"load-balanced"
|
||||
]
|
||||
}
|
||||
],
|
||||
"createEntities": [
|
||||
{
|
||||
"client": {
|
||||
"id": "client0",
|
||||
"useMultipleMongoses": false,
|
||||
"observeEvents": [
|
||||
"commandStartedEvent"
|
||||
]
|
||||
}
|
||||
},
|
||||
{
|
||||
"database": {
|
||||
"id": "database0",
|
||||
"client": "client0",
|
||||
"databaseName": "transaction-tests"
|
||||
}
|
||||
},
|
||||
{
|
||||
"collection": {
|
||||
"id": "collection0",
|
||||
"database": "database0",
|
||||
"collectionName": "test"
|
||||
}
|
||||
},
|
||||
{
|
||||
"session": {
|
||||
"id": "session0",
|
||||
"client": "client0"
|
||||
}
|
||||
}
|
||||
],
|
||||
"initialData": [
|
||||
{
|
||||
"collectionName": "test",
|
||||
"databaseName": "transaction-tests",
|
||||
"documents": []
|
||||
}
|
||||
],
|
||||
"tests": [
|
||||
{
|
||||
"description": "reads are retried if backpressure labels are added",
|
||||
"operations": [
|
||||
{
|
||||
"object": "session0",
|
||||
"name": "startTransaction"
|
||||
},
|
||||
{
|
||||
"object": "collection0",
|
||||
"name": "insertOne",
|
||||
"arguments": {
|
||||
"session": "session0",
|
||||
"document": {
|
||||
"_id": 1
|
||||
}
|
||||
},
|
||||
"expectResult": {
|
||||
"$$unsetOrMatches": {
|
||||
"insertedId": {
|
||||
"$$unsetOrMatches": 1
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
"object": "testRunner",
|
||||
"name": "failPoint",
|
||||
"arguments": {
|
||||
"client": "client0",
|
||||
"failPoint": {
|
||||
"configureFailPoint": "failCommand",
|
||||
"mode": {
|
||||
"times": 1
|
||||
},
|
||||
"data": {
|
||||
"failCommands": [
|
||||
"find"
|
||||
],
|
||||
"errorLabels": [
|
||||
"RetryableError",
|
||||
"SystemOverloadedError"
|
||||
],
|
||||
"errorCode": 112
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
"object": "collection0",
|
||||
"name": "find",
|
||||
"arguments": {
|
||||
"filter": {},
|
||||
"session": "session0"
|
||||
}
|
||||
},
|
||||
{
|
||||
"object": "session0",
|
||||
"name": "commitTransaction"
|
||||
}
|
||||
],
|
||||
"expectEvents": [
|
||||
{
|
||||
"client": "client0",
|
||||
"events": [
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"command": {
|
||||
"insert": "test",
|
||||
"documents": [
|
||||
{
|
||||
"_id": 1
|
||||
}
|
||||
],
|
||||
"ordered": true,
|
||||
"readConcern": {
|
||||
"$$exists": false
|
||||
},
|
||||
"lsid": {
|
||||
"$$sessionLsid": "session0"
|
||||
},
|
||||
"txnNumber": {
|
||||
"$numberLong": "1"
|
||||
},
|
||||
"startTransaction": true,
|
||||
"autocommit": false,
|
||||
"writeConcern": {
|
||||
"$$exists": false
|
||||
}
|
||||
},
|
||||
"commandName": "insert",
|
||||
"databaseName": "transaction-tests"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"command": {
|
||||
"find": "test",
|
||||
"readConcern": {
|
||||
"$$exists": false
|
||||
},
|
||||
"lsid": {
|
||||
"$$sessionLsid": "session0"
|
||||
},
|
||||
"txnNumber": {
|
||||
"$numberLong": "1"
|
||||
},
|
||||
"autocommit": false,
|
||||
"writeConcern": {
|
||||
"$$exists": false
|
||||
}
|
||||
},
|
||||
"commandName": "find",
|
||||
"databaseName": "transaction-tests"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"command": {
|
||||
"find": "test",
|
||||
"readConcern": {
|
||||
"$$exists": false
|
||||
},
|
||||
"lsid": {
|
||||
"$$sessionLsid": "session0"
|
||||
},
|
||||
"txnNumber": {
|
||||
"$numberLong": "1"
|
||||
},
|
||||
"autocommit": false,
|
||||
"writeConcern": {
|
||||
"$$exists": false
|
||||
}
|
||||
},
|
||||
"commandName": "find",
|
||||
"databaseName": "transaction-tests"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"command": {
|
||||
"abortTransaction": {
|
||||
"$$exists": false
|
||||
},
|
||||
"lsid": {
|
||||
"$$sessionLsid": "session0"
|
||||
},
|
||||
"txnNumber": {
|
||||
"$numberLong": "1"
|
||||
},
|
||||
"startTransaction": {
|
||||
"$$exists": false
|
||||
},
|
||||
"autocommit": false,
|
||||
"writeConcern": {
|
||||
"$$exists": false
|
||||
}
|
||||
},
|
||||
"commandName": "commitTransaction",
|
||||
"databaseName": "admin"
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"description": "reads are retried maxAttempts=5 times if backpressure labels are added",
|
||||
"operations": [
|
||||
{
|
||||
"object": "session0",
|
||||
"name": "startTransaction"
|
||||
},
|
||||
{
|
||||
"object": "collection0",
|
||||
"name": "insertOne",
|
||||
"arguments": {
|
||||
"session": "session0",
|
||||
"document": {
|
||||
"_id": 1
|
||||
}
|
||||
},
|
||||
"expectResult": {
|
||||
"$$unsetOrMatches": {
|
||||
"insertedId": {
|
||||
"$$unsetOrMatches": 1
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
"object": "testRunner",
|
||||
"name": "failPoint",
|
||||
"arguments": {
|
||||
"client": "client0",
|
||||
"failPoint": {
|
||||
"configureFailPoint": "failCommand",
|
||||
"mode": "alwaysOn",
|
||||
"data": {
|
||||
"failCommands": [
|
||||
"find"
|
||||
],
|
||||
"errorLabels": [
|
||||
"RetryableError",
|
||||
"SystemOverloadedError"
|
||||
],
|
||||
"errorCode": 112
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
"object": "collection0",
|
||||
"name": "find",
|
||||
"arguments": {
|
||||
"filter": {},
|
||||
"session": "session0"
|
||||
},
|
||||
"expectError": {
|
||||
"isError": true
|
||||
}
|
||||
},
|
||||
{
|
||||
"object": "session0",
|
||||
"name": "abortTransaction"
|
||||
}
|
||||
],
|
||||
"expectEvents": [
|
||||
{
|
||||
"client": "client0",
|
||||
"events": [
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"commandName": "insert"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"commandName": "find"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"commandName": "find"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"commandName": "find"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"commandName": "find"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"commandName": "find"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"commandName": "find"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"commandName": "abortTransaction"
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
440
test/transactions/unified/backpressure-retryable-writes.json
Normal file
440
test/transactions/unified/backpressure-retryable-writes.json
Normal file
@ -0,0 +1,440 @@
|
||||
{
|
||||
"description": "backpressure-retryable-writes",
|
||||
"schemaVersion": "1.3",
|
||||
"runOnRequirements": [
|
||||
{
|
||||
"minServerVersion": "4.4",
|
||||
"topologies": [
|
||||
"replicaset",
|
||||
"sharded",
|
||||
"load-balanced"
|
||||
]
|
||||
}
|
||||
],
|
||||
"createEntities": [
|
||||
{
|
||||
"client": {
|
||||
"id": "client0",
|
||||
"useMultipleMongoses": false,
|
||||
"observeEvents": [
|
||||
"commandStartedEvent"
|
||||
]
|
||||
}
|
||||
},
|
||||
{
|
||||
"database": {
|
||||
"id": "database0",
|
||||
"client": "client0",
|
||||
"databaseName": "transaction-tests"
|
||||
}
|
||||
},
|
||||
{
|
||||
"collection": {
|
||||
"id": "collection0",
|
||||
"database": "database0",
|
||||
"collectionName": "test"
|
||||
}
|
||||
},
|
||||
{
|
||||
"session": {
|
||||
"id": "session0",
|
||||
"client": "client0"
|
||||
}
|
||||
}
|
||||
],
|
||||
"initialData": [
|
||||
{
|
||||
"collectionName": "test",
|
||||
"databaseName": "transaction-tests",
|
||||
"documents": []
|
||||
}
|
||||
],
|
||||
"tests": [
|
||||
{
|
||||
"description": "writes are retried if backpressure labels are added",
|
||||
"operations": [
|
||||
{
|
||||
"object": "session0",
|
||||
"name": "startTransaction"
|
||||
},
|
||||
{
|
||||
"object": "collection0",
|
||||
"name": "insertOne",
|
||||
"arguments": {
|
||||
"session": "session0",
|
||||
"document": {
|
||||
"_id": 1
|
||||
}
|
||||
},
|
||||
"expectResult": {
|
||||
"$$unsetOrMatches": {
|
||||
"insertedId": {
|
||||
"$$unsetOrMatches": 1
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
"object": "testRunner",
|
||||
"name": "failPoint",
|
||||
"arguments": {
|
||||
"client": "client0",
|
||||
"failPoint": {
|
||||
"configureFailPoint": "failCommand",
|
||||
"mode": {
|
||||
"times": 1
|
||||
},
|
||||
"data": {
|
||||
"failCommands": [
|
||||
"insert"
|
||||
],
|
||||
"errorLabels": [
|
||||
"RetryableError",
|
||||
"SystemOverloadedError"
|
||||
],
|
||||
"errorCode": 112
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
"object": "collection0",
|
||||
"name": "insertOne",
|
||||
"arguments": {
|
||||
"session": "session0",
|
||||
"document": {
|
||||
"_id": 2
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
"object": "session0",
|
||||
"name": "commitTransaction"
|
||||
}
|
||||
],
|
||||
"expectEvents": [
|
||||
{
|
||||
"client": "client0",
|
||||
"events": [
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"command": {
|
||||
"insert": "test",
|
||||
"documents": [
|
||||
{
|
||||
"_id": 1
|
||||
}
|
||||
],
|
||||
"ordered": true,
|
||||
"readConcern": {
|
||||
"$$exists": false
|
||||
},
|
||||
"lsid": {
|
||||
"$$sessionLsid": "session0"
|
||||
},
|
||||
"txnNumber": {
|
||||
"$numberLong": "1"
|
||||
},
|
||||
"startTransaction": true,
|
||||
"autocommit": false,
|
||||
"writeConcern": {
|
||||
"$$exists": false
|
||||
}
|
||||
},
|
||||
"commandName": "insert",
|
||||
"databaseName": "transaction-tests"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"command": {
|
||||
"insert": "test",
|
||||
"documents": [
|
||||
{
|
||||
"_id": 2
|
||||
}
|
||||
],
|
||||
"ordered": true,
|
||||
"readConcern": {
|
||||
"$$exists": false
|
||||
},
|
||||
"lsid": {
|
||||
"$$sessionLsid": "session0"
|
||||
},
|
||||
"txnNumber": {
|
||||
"$numberLong": "1"
|
||||
},
|
||||
"autocommit": false,
|
||||
"writeConcern": {
|
||||
"$$exists": false
|
||||
}
|
||||
},
|
||||
"commandName": "insert",
|
||||
"databaseName": "transaction-tests"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"command": {
|
||||
"insert": "test",
|
||||
"documents": [
|
||||
{
|
||||
"_id": 2
|
||||
}
|
||||
],
|
||||
"ordered": true,
|
||||
"readConcern": {
|
||||
"$$exists": false
|
||||
},
|
||||
"lsid": {
|
||||
"$$sessionLsid": "session0"
|
||||
},
|
||||
"txnNumber": {
|
||||
"$numberLong": "1"
|
||||
},
|
||||
"autocommit": false,
|
||||
"writeConcern": {
|
||||
"$$exists": false
|
||||
}
|
||||
},
|
||||
"commandName": "insert",
|
||||
"databaseName": "transaction-tests"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"command": {
|
||||
"abortTransaction": {
|
||||
"$$exists": false
|
||||
},
|
||||
"lsid": {
|
||||
"$$sessionLsid": "session0"
|
||||
},
|
||||
"txnNumber": {
|
||||
"$numberLong": "1"
|
||||
},
|
||||
"startTransaction": {
|
||||
"$$exists": false
|
||||
},
|
||||
"autocommit": false,
|
||||
"writeConcern": {
|
||||
"$$exists": false
|
||||
}
|
||||
},
|
||||
"commandName": "commitTransaction",
|
||||
"databaseName": "admin"
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
],
|
||||
"outcome": [
|
||||
{
|
||||
"collectionName": "test",
|
||||
"databaseName": "transaction-tests",
|
||||
"documents": [
|
||||
{
|
||||
"_id": 1
|
||||
},
|
||||
{
|
||||
"_id": 2
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"description": "writes are retried maxAttempts=5 times if backpressure labels are added",
|
||||
"operations": [
|
||||
{
|
||||
"object": "session0",
|
||||
"name": "startTransaction"
|
||||
},
|
||||
{
|
||||
"object": "collection0",
|
||||
"name": "insertOne",
|
||||
"arguments": {
|
||||
"session": "session0",
|
||||
"document": {
|
||||
"_id": 1
|
||||
}
|
||||
},
|
||||
"expectResult": {
|
||||
"$$unsetOrMatches": {
|
||||
"insertedId": {
|
||||
"$$unsetOrMatches": 1
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
"object": "testRunner",
|
||||
"name": "failPoint",
|
||||
"arguments": {
|
||||
"client": "client0",
|
||||
"failPoint": {
|
||||
"configureFailPoint": "failCommand",
|
||||
"mode": "alwaysOn",
|
||||
"data": {
|
||||
"failCommands": [
|
||||
"insert"
|
||||
],
|
||||
"errorLabels": [
|
||||
"RetryableError",
|
||||
"SystemOverloadedError"
|
||||
],
|
||||
"errorCode": 112
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
"object": "collection0",
|
||||
"name": "insertOne",
|
||||
"arguments": {
|
||||
"session": "session0",
|
||||
"document": {
|
||||
"_id": 2
|
||||
}
|
||||
},
|
||||
"expectError": {
|
||||
"isError": true
|
||||
}
|
||||
},
|
||||
{
|
||||
"object": "session0",
|
||||
"name": "abortTransaction"
|
||||
}
|
||||
],
|
||||
"expectEvents": [
|
||||
{
|
||||
"client": "client0",
|
||||
"events": [
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"commandName": "insert"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"commandName": "insert"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"commandName": "insert"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"commandName": "insert"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"commandName": "insert"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"commandName": "insert"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"commandName": "insert"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"commandName": "abortTransaction"
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
],
|
||||
"outcome": [
|
||||
{
|
||||
"collectionName": "test",
|
||||
"databaseName": "transaction-tests",
|
||||
"documents": []
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"description": "retry succeeds if backpressure labels are added to the first operation in a transaction",
|
||||
"operations": [
|
||||
{
|
||||
"object": "session0",
|
||||
"name": "startTransaction"
|
||||
},
|
||||
{
|
||||
"object": "testRunner",
|
||||
"name": "failPoint",
|
||||
"arguments": {
|
||||
"client": "client0",
|
||||
"failPoint": {
|
||||
"configureFailPoint": "failCommand",
|
||||
"mode": {
|
||||
"times": 1
|
||||
},
|
||||
"data": {
|
||||
"failCommands": [
|
||||
"insert"
|
||||
],
|
||||
"errorLabels": [
|
||||
"RetryableError",
|
||||
"SystemOverloadedError"
|
||||
],
|
||||
"errorCode": 112
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
"object": "collection0",
|
||||
"name": "insertOne",
|
||||
"arguments": {
|
||||
"session": "session0",
|
||||
"document": {
|
||||
"_id": 2
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
"object": "session0",
|
||||
"name": "abortTransaction"
|
||||
}
|
||||
],
|
||||
"expectEvents": [
|
||||
{
|
||||
"client": "client0",
|
||||
"events": [
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"commandName": "insert"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"commandName": "insert"
|
||||
}
|
||||
},
|
||||
{
|
||||
"commandStartedEvent": {
|
||||
"commandName": "abortTransaction"
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
],
|
||||
"outcome": [
|
||||
{
|
||||
"collectionName": "test",
|
||||
"databaseName": "transaction-tests",
|
||||
"documents": []
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
@ -209,9 +209,9 @@ converted_tests = [
|
||||
"test_auth_oidc.py",
|
||||
"test_auth_spec.py",
|
||||
"test_bulk.py",
|
||||
"test_backpressure.py",
|
||||
"test_change_stream.py",
|
||||
"test_client.py",
|
||||
"test_client_backpressure.py",
|
||||
"test_client_bulk_write.py",
|
||||
"test_client_context.py",
|
||||
"test_client_metadata.py",
|
||||
|
||||
Loading…
Reference in New Issue
Block a user