PYTHON-5536 Avoid clearing the connection pool when the server connection rate limiter triggers

[blink1073]

Currently testing with this script for async:

import asyncio
from pymongo import AsyncMongoClient

client = AsyncMongoClient(maxConnecting=100)

async def target():
    await client.admin.command("ping")

async def main():
    await client.admin.command('setParameter', 1, ingressConnectionEstablishmentRateLimiterEnabled=True)
    await client.admin.command('setParameter', 1, ingressConnectionEstablishmentRatePerSec=30)
    await client.admin.command('setParameter', 1, ingressConnectionEstablishmentBurstCapacitySecs=1)
    await client.admin.command('setParameter', 1, ingressConnectionEstablishmentMaxQueueDepth=1)

    # Warm the pool so there are existing connections.
    tasks = []
    for i in range(10):
        tasks.append(asyncio.create_task(target()))
    await asyncio.wait(tasks)

    tasks = []
    for i in range(200):
        tasks.append(asyncio.create_task(target()))

    await asyncio.wait(tasks)


asyncio.run(main())

and this one for sync:

from pymongo import MongoClient
from concurrent.futures import ThreadPoolExecutor, wait

client = MongoClient(maxConnecting=100)

def target():
    client.admin.command("ping")

def main():
    client.admin.command('setParameter', 1, ingressConnectionEstablishmentRateLimiterEnabled=True)
    client.admin.command('setParameter', 1, ingressConnectionEstablishmentRatePerSec=30)
    client.admin.command('setParameter', 1, ingressConnectionEstablishmentBurstCapacitySecs=1)
    client.admin.command('setParameter', 1, ingressConnectionEstablishmentMaxQueueDepth=1)

    # Warm the pool so there are existing connections.
    print("1")
    tasks = []
    pool = ThreadPoolExecutor(200)
    for i in range(10):
        tasks.append(pool.submit(target))
    wait(tasks)
    print("2")
    tasks = []
    for i in range(200):
        tasks.append(pool.submit(target))
    wait(tasks)

main()

[ShaneHarvey]

2 comments:

1. I believe this logic needs to move to connection checkout. Here in connection check in we already know the connection is useable because we're checking it back in after a successful command.
2. Instead of a 1ms read can we reuse the existing _perished() + conn_closed() methods?

[blink1073]

Done

[ShaneHarvey]

Nice work!

[ShaneHarvey]

We need to move this logic so that it covers the TCP+TLS handshake which happen up above.

[blink1073]

I set a breakpoint in the TCP+TLS handshake error handler and confirmed that handshakes are succeeding. The error only occurs on hello/auth.

[ShaneHarvey]

Okay I'm actually surprised by this since the design SPM-4319 indicates the rate limiter rejection happens before the TLS handshake.

[blink1073]

Ideally we'd like to detect connection reset by peer errors and only backoff for those, but in my testing, during backoff, we can sometimes get a regular EOF instead of a closing error, which triggers the raise OSError("connection closed") error.

[NoahStapp]

Is calling is_closing here better? It'll catch more edge cases in theory.

[blink1073]

Hmm let me try that.

[blink1073]

No, it is ambiguous as to whether connection_lost as been called yet. Since connection_lost is synchronous, checking for self._closed.done() assures that we have actually lost the connection.

[ShaneHarvey]

Could you merge backpressure? Originally I added the incorrect labels here. It should be "SystemOverloadedError" and "RetryableError"

[ShaneHarvey]

Instead of inspecting the error message after the fact, is it possible we can record some state to determine if the error happened during DNS+TCP or after? Like:

# Assume all non dns/tcp/timeout errors mean the server rejected the connection due to overload.
if not errorDuringDnsTcp and not timeoutError:
    error._add_error_label("SystemOverloadedError")