Skip to content

PYTHON-5731 - Server selection deprioritization only for overload errors on replica sets #2710

New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

Open
NoahStapp wants to merge 4 commits into
base: master
Choose a base branch
from
+166 −2
Open

PYTHON-5731 - Server selection deprioritization only for overload errors on replica sets #2710

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
003796f
PYTHON-5731 - Server selection deprioritization only for overload err…
NoahStapp Feb 19, 2026
77464f0
Format tests
NoahStapp Feb 19, 2026
875d8e3
Require 4.4+
NoahStapp Feb 20, 2026
dff8159
Use require_secondaries_count
NoahStapp Feb 20, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion pymongo/asynchronous/mongo_client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2825,7 +2825,11 @@ async def run(self) -> T:
if self._last_error is None:
self._last_error = exc

if self._server is not None:
if (
self._server is not None
and self._client.topology_description.topology_type_name == "Sharded"
or exc.has_error_label("SystemOverloadedError")
):
self._deprioritized_servers.append(self._server)

def _is_not_eligible_for_retry(self) -> bool:
Expand Down
6 changes: 5 additions & 1 deletion pymongo/synchronous/mongo_client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2815,7 +2815,11 @@ def run(self) -> T:
if self._last_error is None:
self._last_error = exc

if self._server is not None:
if (
self._server is not None
and self._client.topology_description.topology_type_name == "Sharded"
or exc.has_error_label("SystemOverloadedError")
):
self._deprioritized_servers.append(self._server)

def _is_not_eligible_for_retry(self) -> bool:
Expand Down
78 changes: 78 additions & 0 deletions test/asynchronous/test_retryable_reads.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -261,6 +261,84 @@ async def test_retryable_reads_are_retried_on_the_same_implicit_session(self):
self.assertEqual(command_docs[0]["lsid"], command_docs[1]["lsid"])
self.assertIsNot(command_docs[0], command_docs[1])

@async_client_context.require_replica_set
@async_client_context.require_secondaries_count(1)
@async_client_context.require_failCommand_fail_point
@async_client_context.require_version_min(4, 4, 0)
Copy link
Member

@ShaneHarvey ShaneHarvey Feb 20, 2026 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

For these tests do we need to use require_secondaries_count?

Copy link
Contributor Author

@NoahStapp NoahStapp Feb 20, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good catch. These require at least one secondary.

Copy link
Contributor Author

@NoahStapp NoahStapp Feb 20, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What's the minimum definition of a replica set?

Copy link
Member

@ShaneHarvey ShaneHarvey Feb 20, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

A single member replica set is the min.

async def test_03_01_retryable_reads_caused_by_overload_errors_are_retried_on_a_different_replicaset_server_when_one_is_available(
self
):
listener = OvertCommandListener()

# 1. Create a client `client` with `retryReads=true`, `readPreference=primaryPreferred`, and command event monitoring enabled.
client = await self.async_rs_or_single_client(
event_listeners=[listener], retryReads=True, readPreference="primaryPreferred"
)

# 2. Configure a fail point with the RetryableError and SystemOverloadedError error labels.
command_args = {
"configureFailPoint": "failCommand",
"mode": {"times": 1},
"data": {
"failCommands": ["find"],
"errorLabels": ["RetryableError", "SystemOverloadedError"],
"errorCode": 6,
},
}
await async_set_fail_point(client, command_args)

# 3. Reset the command event monitor to clear the fail point command from its stored events.
listener.reset()

# 4. Execute a `find` command with `client`.
await client.t.t.find_one({})

# 5. Assert that one failed command event and one successful command event occurred.
self.assertEqual(len(listener.failed_events), 1)
self.assertEqual(len(listener.succeeded_events), 1)

# 6. Assert that both events occurred on different servers.
assert listener.failed_events[0].connection_id != listener.succeeded_events[0].connection_id

@async_client_context.require_replica_set
@async_client_context.require_secondaries_count(1)
@async_client_context.require_failCommand_fail_point
@async_client_context.require_version_min(4, 4, 0)
async def test_03_02_retryable_reads_caused_by_non_overload_errors_are_retried_on_the_same_replicaset_server(
self
):
listener = OvertCommandListener()

# 1. Create a client `client` with `retryReads=true`, `readPreference=primaryPreferred`, and command event monitoring enabled.
client = await self.async_rs_or_single_client(
event_listeners=[listener], retryReads=True, readPreference="primaryPreferred"
)

# 2. Configure a fail point with the RetryableError error label.
command_args = {
"configureFailPoint": "failCommand",
"mode": {"times": 1},
"data": {
"failCommands": ["find"],
"errorLabels": ["RetryableError"],
"errorCode": 6,
},
}
await async_set_fail_point(client, command_args)

# 3. Reset the command event monitor to clear the fail point command from its stored events.
listener.reset()

# 4. Execute a `find` command with `client`.
await client.t.t.find_one({})

# 5. Assert that one failed command event and one successful command event occurred.
self.assertEqual(len(listener.failed_events), 1)
self.assertEqual(len(listener.succeeded_events), 1)

# 6. Assert that both events occurred the same server.
assert listener.failed_events[0].connection_id == listener.succeeded_events[0].connection_id


if __name__ == "__main__":
unittest.main()
78 changes: 78 additions & 0 deletions test/test_retryable_reads.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -259,6 +259,84 @@ def test_retryable_reads_are_retried_on_the_same_implicit_session(self):
self.assertEqual(command_docs[0]["lsid"], command_docs[1]["lsid"])
self.assertIsNot(command_docs[0], command_docs[1])

@client_context.require_replica_set
@client_context.require_secondaries_count(1)
@client_context.require_failCommand_fail_point
@client_context.require_version_min(4, 4, 0)
def test_03_01_retryable_reads_caused_by_overload_errors_are_retried_on_a_different_replicaset_server_when_one_is_available(
self
):
listener = OvertCommandListener()

# 1. Create a client `client` with `retryReads=true`, `readPreference=primaryPreferred`, and command event monitoring enabled.
client = self.rs_or_single_client(
event_listeners=[listener], retryReads=True, readPreference="primaryPreferred"
)

# 2. Configure a fail point with the RetryableError and SystemOverloadedError error labels.
command_args = {
"configureFailPoint": "failCommand",
"mode": {"times": 1},
"data": {
"failCommands": ["find"],
"errorLabels": ["RetryableError", "SystemOverloadedError"],
"errorCode": 6,
},
}
set_fail_point(client, command_args)

# 3. Reset the command event monitor to clear the fail point command from its stored events.
listener.reset()

# 4. Execute a `find` command with `client`.
client.t.t.find_one({})

# 5. Assert that one failed command event and one successful command event occurred.
self.assertEqual(len(listener.failed_events), 1)
self.assertEqual(len(listener.succeeded_events), 1)

# 6. Assert that both events occurred on different servers.
assert listener.failed_events[0].connection_id != listener.succeeded_events[0].connection_id

@client_context.require_replica_set
@client_context.require_secondaries_count(1)
@client_context.require_failCommand_fail_point
@client_context.require_version_min(4, 4, 0)
def test_03_02_retryable_reads_caused_by_non_overload_errors_are_retried_on_the_same_replicaset_server(
self
):
listener = OvertCommandListener()

# 1. Create a client `client` with `retryReads=true`, `readPreference=primaryPreferred`, and command event monitoring enabled.
client = self.rs_or_single_client(
event_listeners=[listener], retryReads=True, readPreference="primaryPreferred"
)

# 2. Configure a fail point with the RetryableError error label.
command_args = {
"configureFailPoint": "failCommand",
"mode": {"times": 1},
"data": {
"failCommands": ["find"],
"errorLabels": ["RetryableError"],
"errorCode": 6,
},
}
set_fail_point(client, command_args)

# 3. Reset the command event monitor to clear the fail point command from its stored events.
listener.reset()

# 4. Execute a `find` command with `client`.
client.t.t.find_one({})

# 5. Assert that one failed command event and one successful command event occurred.
self.assertEqual(len(listener.failed_events), 1)
self.assertEqual(len(listener.succeeded_events), 1)

# 6. Assert that both events occurred the same server.
assert listener.failed_events[0].connection_id == listener.succeeded_events[0].connection_id


if __name__ == "__main__":
unittest.main()
Loading