From d16d9b2b83cd1cb9504a1fc90a790ea640b310d3 Mon Sep 17 00:00:00 2001
From: Steven Silvester <steve.silvester@mongodb.com>
Date: Thu, 1 Oct 2026 16:05:49 -0500
Subject: [PATCH 1/6] PYTHON-6136 Fix pool checkout accounting rollback on
 BaseException

A BaseException (KeyboardInterrupt, CancelledError, GreenletExit)
landing inside Pool.checkout could leak checkout accounting: the size
gate (self.requests) was only rolled back after a socket was acquired,
and the maxConnecting _pending counter was only rolled back on the
happy path. Roll back both counters, with notify(), when a
BaseException interrupts the gate or the connection attempt. Rework
the gevent killall race test to amplify the checkout unwind windows
and assert the counters fully drain after the pool settles.
---
 pymongo/asynchronous/pool.py     | 66 ++++++++++++++++++++++++--------
 pymongo/synchronous/pool.py      | 66 ++++++++++++++++++++++++--------
 test/asynchronous/test_client.py | 57 +++++++++++++++++++--------
 test/test_client.py              | 57 +++++++++++++++++++--------
 4 files changed, 182 insertions(+), 64 deletions(-)

diff --git a/pymongo/asynchronous/pool.py b/pymongo/asynchronous/pool.py
index 69ac14e7e4..7d34193d73 100644
--- a/pymongo/asynchronous/pool.py
+++ b/pymongo/asynchronous/pool.py
@@ -980,24 +980,44 @@ async def _get_conn(
         else:
             deadline = None
 
-        async with self.size_cond:
-            self._raise_if_not_ready(checkout_started_time, emit_event=True)
-            while not (self.requests < self.max_pool_size):
-                timeout = deadline - time.monotonic() if deadline else None
-                if not await _async_cond_wait(self.size_cond, timeout):
-                    # Timed out, notify the next thread to ensure a
-                    # timeout doesn't consume the condition.
-                    if self.requests < self.max_pool_size:
-                        self.size_cond.notify()
-                    self._raise_wait_queue_timeout(checkout_started_time)
+        # Roll back the size gate if a BaseException lands here (PYTHON-6136).
+        requests_incremented = False
+        try:
+            async with self.size_cond:
                 self._raise_if_not_ready(checkout_started_time, emit_event=True)
-            self.requests += 1
+                while not (self.requests < self.max_pool_size):
+                    timeout = deadline - time.monotonic() if deadline else None
+                    if not await _async_cond_wait(self.size_cond, timeout):
+                        # Timed out, notify the next thread to ensure a
+                        # timeout doesn't consume the condition.
+                        if self.requests < self.max_pool_size:
+                            self.size_cond.notify()
+                        self._raise_wait_queue_timeout(checkout_started_time)
+                    self._raise_if_not_ready(checkout_started_time, emit_event=True)
+                self.requests += 1
+                requests_incremented = True
+        except BaseException:
+            if requests_incremented:
+                # Gevent grants the re-acquire during unwind (PYTHON-6074).
+                accounted = False
+                try:
+                    async with self.size_cond:
+                        self.requests -= 1
+                        accounted = True
+                        self.size_cond.notify()
+                finally:
+                    if not accounted:
+                        async with self.size_cond:
+                            self.requests -= 1
+                            self.size_cond.notify()
+            raise
 
         # We've now acquired the semaphore and must release it on error.
         conn = None
         incremented = False
         emitted_event = False
         is_new_conn = False
+        pending_incremented = False
         try:
             async with self.lock:
                 self.active_sockets += 1
@@ -1022,6 +1042,7 @@ async def _get_conn(
                         conn = self.conns.popleft()
                     except IndexError:
                         self._pending += 1
+                        pending_incremented = True
                 if conn:  # We got a socket from the pool
                     if await self._perished(conn):
                         conn = None
@@ -1033,6 +1054,7 @@ async def _get_conn(
                     finally:
                         async with self._max_connecting_cond:
                             self._pending -= 1
+                            pending_incremented = False
                             self._max_connecting_cond.notify()
 
             conn.active = True
@@ -1043,12 +1065,24 @@ async def _get_conn(
                     self.active_contexts.add(conn.cancel_context)
         # Catch KeyboardInterrupt, CancelledError, etc. and cleanup.
         except BaseException:
+            if pending_incremented:
+                # Gevent grants the re-acquire during unwind (PYTHON-6074).
+                pending_accounted = False
+                try:
+                    async with self._max_connecting_cond:
+                        self._pending -= 1
+                        pending_accounted = True
+                        self._max_connecting_cond.notify()
+                finally:
+                    if not pending_accounted:
+                        async with self._max_connecting_cond:
+                            self._pending -= 1
+                            self._max_connecting_cond.notify()
+
             if conn:
                 # We checked out a socket but authentication failed.
                 await conn.close_conn(ConnectionClosedReason.ERROR)
-            # Re-apply the accounting if a GreenletExit interrupts
-            # during the size_cond acquisition; during unwind gevent
-            # lets the re-acquire complete (PYTHON-6074).
+            # Re-apply the accounting if a BaseException interrupts here (PYTHON-6074).
             accounted = False
             try:
                 async with self.size_cond:
@@ -1121,9 +1155,7 @@ async def checkin(self, conn: AsyncConnection) -> None:
         conn.pinned_cursor = False
         self._pinned_sockets.discard(conn)
         forked = self.pid != os.getpid()
-        # Re-apply the accounting if a gevent GreenletExit interrupts during
-        # the size_cond acquisition; gevent lets the re-acquire complete while
-        # unwinding (PYTHON-6074).
+        # Re-apply the accounting if a BaseException interrupts here (PYTHON-6074).
         close_conn_reason: Optional[str] = None
         emit_closed = False
         accounted = False
diff --git a/pymongo/synchronous/pool.py b/pymongo/synchronous/pool.py
index c977729ba8..56713aeae5 100644
--- a/pymongo/synchronous/pool.py
+++ b/pymongo/synchronous/pool.py
@@ -976,24 +976,44 @@ def _get_conn(
         else:
             deadline = None
 
-        with self.size_cond:
-            self._raise_if_not_ready(checkout_started_time, emit_event=True)
-            while not (self.requests < self.max_pool_size):
-                timeout = deadline - time.monotonic() if deadline else None
-                if not _cond_wait(self.size_cond, timeout):
-                    # Timed out, notify the next thread to ensure a
-                    # timeout doesn't consume the condition.
-                    if self.requests < self.max_pool_size:
-                        self.size_cond.notify()
-                    self._raise_wait_queue_timeout(checkout_started_time)
+        # Roll back the size gate if a BaseException lands here (PYTHON-6136).
+        requests_incremented = False
+        try:
+            with self.size_cond:
                 self._raise_if_not_ready(checkout_started_time, emit_event=True)
-            self.requests += 1
+                while not (self.requests < self.max_pool_size):
+                    timeout = deadline - time.monotonic() if deadline else None
+                    if not _cond_wait(self.size_cond, timeout):
+                        # Timed out, notify the next thread to ensure a
+                        # timeout doesn't consume the condition.
+                        if self.requests < self.max_pool_size:
+                            self.size_cond.notify()
+                        self._raise_wait_queue_timeout(checkout_started_time)
+                    self._raise_if_not_ready(checkout_started_time, emit_event=True)
+                self.requests += 1
+                requests_incremented = True
+        except BaseException:
+            if requests_incremented:
+                # Gevent grants the re-acquire during unwind (PYTHON-6074).
+                accounted = False
+                try:
+                    with self.size_cond:
+                        self.requests -= 1
+                        accounted = True
+                        self.size_cond.notify()
+                finally:
+                    if not accounted:
+                        with self.size_cond:
+                            self.requests -= 1
+                            self.size_cond.notify()
+            raise
 
         # We've now acquired the semaphore and must release it on error.
         conn = None
         incremented = False
         emitted_event = False
         is_new_conn = False
+        pending_incremented = False
         try:
             with self.lock:
                 self.active_sockets += 1
@@ -1018,6 +1038,7 @@ def _get_conn(
                         conn = self.conns.popleft()
                     except IndexError:
                         self._pending += 1
+                        pending_incremented = True
                 if conn:  # We got a socket from the pool
                     if self._perished(conn):
                         conn = None
@@ -1029,6 +1050,7 @@ def _get_conn(
                     finally:
                         with self._max_connecting_cond:
                             self._pending -= 1
+                            pending_incremented = False
                             self._max_connecting_cond.notify()
 
             conn.active = True
@@ -1039,12 +1061,24 @@ def _get_conn(
                     self.active_contexts.add(conn.cancel_context)
         # Catch KeyboardInterrupt, CancelledError, etc. and cleanup.
         except BaseException:
+            if pending_incremented:
+                # Gevent grants the re-acquire during unwind (PYTHON-6074).
+                pending_accounted = False
+                try:
+                    with self._max_connecting_cond:
+                        self._pending -= 1
+                        pending_accounted = True
+                        self._max_connecting_cond.notify()
+                finally:
+                    if not pending_accounted:
+                        with self._max_connecting_cond:
+                            self._pending -= 1
+                            self._max_connecting_cond.notify()
+
             if conn:
                 # We checked out a socket but authentication failed.
                 conn.close_conn(ConnectionClosedReason.ERROR)
-            # Re-apply the accounting if a GreenletExit interrupts
-            # during the size_cond acquisition; during unwind gevent
-            # lets the re-acquire complete (PYTHON-6074).
+            # Re-apply the accounting if a BaseException interrupts here (PYTHON-6074).
             accounted = False
             try:
                 with self.size_cond:
@@ -1117,9 +1151,7 @@ def checkin(self, conn: Connection) -> None:
         conn.pinned_cursor = False
         self._pinned_sockets.discard(conn)
         forked = self.pid != os.getpid()
-        # Re-apply the accounting if a gevent GreenletExit interrupts during
-        # the size_cond acquisition; gevent lets the re-acquire complete while
-        # unwinding (PYTHON-6074).
+        # Re-apply the accounting if a BaseException interrupts here (PYTHON-6074).
         close_conn_reason: Optional[str] = None
         emit_closed = False
         accounted = False
diff --git a/test/asynchronous/test_client.py b/test/asynchronous/test_client.py
index 56810b1433..b8ec8479d4 100644
--- a/test/asynchronous/test_client.py
+++ b/test/asynchronous/test_client.py
@@ -2744,20 +2744,18 @@ def test_gevent_kill_churn_deadlock(self):
         import gevent.thread as _gthread
         from gevent import Timeout, spawn
 
-        # AMPLIFY_RACE=1 widens gevent's brief sleep on a contended lock
-        # so a kill lands there reliably. Only the bare sleep() is
-        # widened; timed sleeps pass through. The unfixed test then
-        # deadlocks within seconds.
+        # AMPLIFY_RACE=1 widens the kill windows below; keep AMPLIFY_SECONDS
+        # small or ops starve.
+        amplify_seconds = 0.0
         if os.environ.get("AMPLIFY_RACE", "0") == "1":
-            _AMPLIFY_SECONDS = float(os.environ.get("AMPLIFY_SECONDS", "0.02"))
+            amplify_seconds = float(os.environ.get("AMPLIFY_SECONDS", "0.002"))
 
             _orig_thread_sleep = _gthread.sleep
 
             def _amplified_sleep(*args):
-                if not args:  # bare sleep(): the courtesy yield on a failed
-                    # non-blocking acquire (Condition.notify -> _is_owned -> acquire(False))
-                    _orig_thread_sleep(_AMPLIFY_SECONDS)
-                else:  # sleep(0.001), sleep(2), etc.: passthrough
+                if not args:  # bare sleep(): checkin courtesy yield
+                    _orig_thread_sleep(amplify_seconds)
+                else:  # timed sleeps pass through
                     _orig_thread_sleep(*args)
 
             _gthread.sleep = _amplified_sleep
@@ -2767,6 +2765,31 @@ def _amplified_sleep(*args):
         coll = client.pymongo_test.coll
         coll.insert_one({})
 
+        # Widen the post-gate checkout windows (PYTHON-6136).
+        if amplify_seconds:
+            pool = async_get_pool(client)  # type:ignore
+
+            class _AmplifiedCondition:
+                def __init__(self, cond, seconds):
+                    self._cond = cond
+                    self._seconds = seconds
+
+                def __enter__(self):
+                    self._cond.__enter__()
+                    return self
+
+                def __exit__(self, *args):
+                    self._cond.__exit__(*args)
+                    time.sleep(self._seconds)
+
+                def __getattr__(self, name):
+                    return getattr(self._cond, name)
+
+            pool.size_cond = _AmplifiedCondition(pool.size_cond, amplify_seconds)
+            pool._max_connecting_cond = _AmplifiedCondition(
+                pool._max_connecting_cond, amplify_seconds
+            )
+
         op_count = [0]
         running = [True]
         workers: list = []
@@ -2780,9 +2803,12 @@ def worker():
                 except Exception:
                     return
 
+        # Scale the kill cadence or ops starve.
+        reaper_interval = max(0.003, amplify_seconds * 8)
+
         def reaper():
             while running[0]:
-                time.sleep(0.003)
+                time.sleep(reaper_interval)
                 if not workers:
                     continue
                 idx = random.randrange(len(workers))
@@ -2823,12 +2849,8 @@ def reaper():
                     coll.find_one({})
             except Timeout:
                 self.fail("Pool gate saturated (PYTHON-6074)")
-            # Deterministic check: a saturated size gate pins the pool's
-            # checkout counters at maxPoolSize (PYTHON-6074).
-            pool = async_get_pool(client)  # type:ignore
-            self.assertLess(pool.requests, pool.max_pool_size)
-            self.assertLess(pool.active_sockets, pool.max_pool_size)
             self.assertGreater(op_count[0], 0)
+            pool = async_get_pool(client)  # type:ignore
         finally:
             running[0] = False
             gevent.killall(workers, block=False)
@@ -2845,6 +2867,11 @@ def reaper():
                     client.close()
             except Timeout:
                 pass
+        # Post-settle: the counters must have fully drained.
+        time.sleep(1.0)
+        self.assertEqual(pool.requests, 0)
+        self.assertEqual(pool._pending, 0)
+        self.assertEqual(pool.active_sockets, 0)
 
 
 class TestClientLazyConnect(AsyncIntegrationTest):
diff --git a/test/test_client.py b/test/test_client.py
index ebd1e670ee..9e7a41d506 100644
--- a/test/test_client.py
+++ b/test/test_client.py
@@ -2695,20 +2695,18 @@ def test_gevent_kill_churn_deadlock(self):
         import gevent.thread as _gthread
         from gevent import Timeout, spawn
 
-        # AMPLIFY_RACE=1 widens gevent's brief sleep on a contended lock
-        # so a kill lands there reliably. Only the bare sleep() is
-        # widened; timed sleeps pass through. The unfixed test then
-        # deadlocks within seconds.
+        # AMPLIFY_RACE=1 widens the kill windows below; keep AMPLIFY_SECONDS
+        # small or ops starve.
+        amplify_seconds = 0.0
         if os.environ.get("AMPLIFY_RACE", "0") == "1":
-            _AMPLIFY_SECONDS = float(os.environ.get("AMPLIFY_SECONDS", "0.02"))
+            amplify_seconds = float(os.environ.get("AMPLIFY_SECONDS", "0.002"))
 
             _orig_thread_sleep = _gthread.sleep
 
             def _amplified_sleep(*args):
-                if not args:  # bare sleep(): the courtesy yield on a failed
-                    # non-blocking acquire (Condition.notify -> _is_owned -> acquire(False))
-                    _orig_thread_sleep(_AMPLIFY_SECONDS)
-                else:  # sleep(0.001), sleep(2), etc.: passthrough
+                if not args:  # bare sleep(): checkin courtesy yield
+                    _orig_thread_sleep(amplify_seconds)
+                else:  # timed sleeps pass through
                     _orig_thread_sleep(*args)
 
             _gthread.sleep = _amplified_sleep
@@ -2718,6 +2716,31 @@ def _amplified_sleep(*args):
         coll = client.pymongo_test.coll
         coll.insert_one({})
 
+        # Widen the post-gate checkout windows (PYTHON-6136).
+        if amplify_seconds:
+            pool = get_pool(client)  # type:ignore
+
+            class _AmplifiedCondition:
+                def __init__(self, cond, seconds):
+                    self._cond = cond
+                    self._seconds = seconds
+
+                def __enter__(self):
+                    self._cond.__enter__()
+                    return self
+
+                def __exit__(self, *args):
+                    self._cond.__exit__(*args)
+                    time.sleep(self._seconds)
+
+                def __getattr__(self, name):
+                    return getattr(self._cond, name)
+
+            pool.size_cond = _AmplifiedCondition(pool.size_cond, amplify_seconds)
+            pool._max_connecting_cond = _AmplifiedCondition(
+                pool._max_connecting_cond, amplify_seconds
+            )
+
         op_count = [0]
         running = [True]
         workers: list = []
@@ -2731,9 +2754,12 @@ def worker():
                 except Exception:
                     return
 
+        # Scale the kill cadence or ops starve.
+        reaper_interval = max(0.003, amplify_seconds * 8)
+
         def reaper():
             while running[0]:
-                time.sleep(0.003)
+                time.sleep(reaper_interval)
                 if not workers:
                     continue
                 idx = random.randrange(len(workers))
@@ -2774,12 +2800,8 @@ def reaper():
                     coll.find_one({})
             except Timeout:
                 self.fail("Pool gate saturated (PYTHON-6074)")
-            # Deterministic check: a saturated size gate pins the pool's
-            # checkout counters at maxPoolSize (PYTHON-6074).
-            pool = get_pool(client)  # type:ignore
-            self.assertLess(pool.requests, pool.max_pool_size)
-            self.assertLess(pool.active_sockets, pool.max_pool_size)
             self.assertGreater(op_count[0], 0)
+            pool = get_pool(client)  # type:ignore
         finally:
             running[0] = False
             gevent.killall(workers, block=False)
@@ -2796,6 +2818,11 @@ def reaper():
                     client.close()
             except Timeout:
                 pass
+        # Post-settle: the counters must have fully drained.
+        time.sleep(1.0)
+        self.assertEqual(pool.requests, 0)
+        self.assertEqual(pool._pending, 0)
+        self.assertEqual(pool.active_sockets, 0)
 
 
 class TestClientLazyConnect(IntegrationTest):

From 04dccc5f7fda56c0b1a0afc027a0c59f7bcc2f13 Mon Sep 17 00:00:00 2001
From: Steven Silvester <steve.silvester@mongodb.com>
Date: Thu, 1 Oct 2026 17:27:24 -0500
Subject: [PATCH 2/6] PYTHON-6136 Always roll back the size gate on checkout
 unwind

---
 pymongo/asynchronous/pool.py | 49 +++++++++++++++++++-----------------
 pymongo/synchronous/pool.py  | 49 +++++++++++++++++++-----------------
 2 files changed, 52 insertions(+), 46 deletions(-)

diff --git a/pymongo/asynchronous/pool.py b/pymongo/asynchronous/pool.py
index 7d34193d73..ec1c88a7e7 100644
--- a/pymongo/asynchronous/pool.py
+++ b/pymongo/asynchronous/pool.py
@@ -1065,39 +1065,42 @@ async def _get_conn(
                     self.active_contexts.add(conn.cancel_context)
         # Catch KeyboardInterrupt, CancelledError, etc. and cleanup.
         except BaseException:
-            if pending_incremented:
-                # Gevent grants the re-acquire during unwind (PYTHON-6074).
-                pending_accounted = False
-                try:
-                    async with self._max_connecting_cond:
-                        self._pending -= 1
-                        pending_accounted = True
-                        self._max_connecting_cond.notify()
-                finally:
-                    if not pending_accounted:
+            try:
+                if pending_incremented:
+                    # Gevent grants the re-acquire during unwind (PYTHON-6074).
+                    pending_accounted = False
+                    try:
                         async with self._max_connecting_cond:
                             self._pending -= 1
+                            pending_accounted = True
                             self._max_connecting_cond.notify()
+                    finally:
+                        if not pending_accounted:
+                            async with self._max_connecting_cond:
+                                self._pending -= 1
+                                self._max_connecting_cond.notify()
 
-            if conn:
-                # We checked out a socket but authentication failed.
-                await conn.close_conn(ConnectionClosedReason.ERROR)
-            # Re-apply the accounting if a BaseException interrupts here (PYTHON-6074).
-            accounted = False
-            try:
-                async with self.size_cond:
-                    self.requests -= 1
-                    if incremented:
-                        self.active_sockets -= 1
-                    accounted = True
-                    self.size_cond.notify()
+                if conn:
+                    # We checked out a socket but authentication failed.
+                    await conn.close_conn(ConnectionClosedReason.ERROR)
             finally:
-                if not accounted:
+                # Always roll back the size gate, even if the cleanups above
+                # were interrupted (PYTHON-6136).
+                accounted = False
+                try:
                     async with self.size_cond:
                         self.requests -= 1
                         if incremented:
                             self.active_sockets -= 1
+                        accounted = True
                         self.size_cond.notify()
+                finally:
+                    if not accounted:
+                        async with self.size_cond:
+                            self.requests -= 1
+                            if incremented:
+                                self.active_sockets -= 1
+                            self.size_cond.notify()
 
             if not emitted_event:
                 self._telemetry.checkout_failed(
diff --git a/pymongo/synchronous/pool.py b/pymongo/synchronous/pool.py
index 56713aeae5..a851f6a9b7 100644
--- a/pymongo/synchronous/pool.py
+++ b/pymongo/synchronous/pool.py
@@ -1061,39 +1061,42 @@ def _get_conn(
                     self.active_contexts.add(conn.cancel_context)
         # Catch KeyboardInterrupt, CancelledError, etc. and cleanup.
         except BaseException:
-            if pending_incremented:
-                # Gevent grants the re-acquire during unwind (PYTHON-6074).
-                pending_accounted = False
-                try:
-                    with self._max_connecting_cond:
-                        self._pending -= 1
-                        pending_accounted = True
-                        self._max_connecting_cond.notify()
-                finally:
-                    if not pending_accounted:
+            try:
+                if pending_incremented:
+                    # Gevent grants the re-acquire during unwind (PYTHON-6074).
+                    pending_accounted = False
+                    try:
                         with self._max_connecting_cond:
                             self._pending -= 1
+                            pending_accounted = True
                             self._max_connecting_cond.notify()
+                    finally:
+                        if not pending_accounted:
+                            with self._max_connecting_cond:
+                                self._pending -= 1
+                                self._max_connecting_cond.notify()
 
-            if conn:
-                # We checked out a socket but authentication failed.
-                conn.close_conn(ConnectionClosedReason.ERROR)
-            # Re-apply the accounting if a BaseException interrupts here (PYTHON-6074).
-            accounted = False
-            try:
-                with self.size_cond:
-                    self.requests -= 1
-                    if incremented:
-                        self.active_sockets -= 1
-                    accounted = True
-                    self.size_cond.notify()
+                if conn:
+                    # We checked out a socket but authentication failed.
+                    conn.close_conn(ConnectionClosedReason.ERROR)
             finally:
-                if not accounted:
+                # Always roll back the size gate, even if the cleanups above
+                # were interrupted (PYTHON-6136).
+                accounted = False
+                try:
                     with self.size_cond:
                         self.requests -= 1
                         if incremented:
                             self.active_sockets -= 1
+                        accounted = True
                         self.size_cond.notify()
+                finally:
+                    if not accounted:
+                        with self.size_cond:
+                            self.requests -= 1
+                            if incremented:
+                                self.active_sockets -= 1
+                            self.size_cond.notify()
 
             if not emitted_event:
                 self._telemetry.checkout_failed(

From 2191a2fe0ff69ddd4cd1bbb6efe7e0d4d8f9029f Mon Sep 17 00:00:00 2001
From: Steven Silvester <steve.silvester@mongodb.com>
Date: Thu, 1 Oct 2026 19:51:13 -0500
Subject: [PATCH 3/6] PYTHON-6136 Add regression tests for interrupted checkout
 cleanup

---
 test/asynchronous/test_pooling.py | 50 +++++++++++++++++++++++++++++++
 test/test_pooling.py              | 50 +++++++++++++++++++++++++++++++
 2 files changed, 100 insertions(+)

diff --git a/test/asynchronous/test_pooling.py b/test/asynchronous/test_pooling.py
index 661bd4e3d2..8452d18b61 100644
--- a/test/asynchronous/test_pooling.py
+++ b/test/asynchronous/test_pooling.py
@@ -342,6 +342,56 @@ async def __aexit__(self, *args):
         self.assertEqual(0, cx_pool.requests)
         self.assertEqual(0, cx_pool.active_sockets)
 
+    async def test_checkout_error_accounting_on_connect_keyboard_interrupt(self):
+        # PYTHON-6136: a KeyboardInterrupt from connect() must roll back the
+        # size gate, the maxConnecting gate, and active_sockets.
+        cx_pool = await self.create_pool(max_pool_size=1)
+
+        with patch.object(cx_pool, "connect", side_effect=KeyboardInterrupt()):
+            with self.assertRaises(KeyboardInterrupt):
+                async with cx_pool.checkout():
+                    pass
+
+        self.assertEqual(0, cx_pool.requests)
+        self.assertEqual(0, cx_pool.active_sockets)
+        self.assertEqual(0, cx_pool._pending)
+
+    async def test_checkout_error_accounting_on_kill_during_pending_cleanup(self):
+        # PYTHON-6136: an interruption during the pending-gate cleanup must
+        # not skip the size-gate rollback below it.
+        cx_pool = await self.create_pool(max_pool_size=1)
+
+        class _InterruptOnSecondEnter(type(cx_pool._max_connecting_cond)):
+            def __init__(self, lock):
+                super().__init__(lock)
+                self.enters = 0
+
+            async def __aenter__(self):
+                self.enters += 1
+                if self.enters == 2:
+                    # First enter is the checkout wait, second is connect()'s
+                    # cleanup. Simulate a kill delivered while waiting there.
+                    raise KeyboardInterrupt()
+                return await super().__aenter__()
+
+            async def __aexit__(self, *args):
+                return await super().__aexit__(*args)
+
+            def notify(self, n=1):
+                # The handler's pending-gate rollback is itself killed.
+                raise KeyboardInterrupt()
+
+        cx_pool._max_connecting_cond = _InterruptOnSecondEnter(cx_pool._max_connecting_cond._lock)
+
+        with patch.object(cx_pool, "connect", side_effect=asyncio.CancelledError()):
+            with self.assertRaises(KeyboardInterrupt):
+                async with cx_pool.checkout():
+                    pass
+
+        self.assertEqual(0, cx_pool.requests)
+        self.assertEqual(0, cx_pool.active_sockets)
+        self.assertEqual(0, cx_pool._pending)
+
     async def test_pool_removes_closed_socket(self):
         # Test that Pool removes explicitly closed socket.
         cx_pool = await self.create_pool()
diff --git a/test/test_pooling.py b/test/test_pooling.py
index a3f0eaf589..2e8e243cbf 100644
--- a/test/test_pooling.py
+++ b/test/test_pooling.py
@@ -342,6 +342,56 @@ def __exit__(self, *args):
         self.assertEqual(0, cx_pool.requests)
         self.assertEqual(0, cx_pool.active_sockets)
 
+    def test_checkout_error_accounting_on_connect_keyboard_interrupt(self):
+        # PYTHON-6136: a KeyboardInterrupt from connect() must roll back the
+        # size gate, the maxConnecting gate, and active_sockets.
+        cx_pool = self.create_pool(max_pool_size=1)
+
+        with patch.object(cx_pool, "connect", side_effect=KeyboardInterrupt()):
+            with self.assertRaises(KeyboardInterrupt):
+                with cx_pool.checkout():
+                    pass
+
+        self.assertEqual(0, cx_pool.requests)
+        self.assertEqual(0, cx_pool.active_sockets)
+        self.assertEqual(0, cx_pool._pending)
+
+    def test_checkout_error_accounting_on_kill_during_pending_cleanup(self):
+        # PYTHON-6136: an interruption during the pending-gate cleanup must
+        # not skip the size-gate rollback below it.
+        cx_pool = self.create_pool(max_pool_size=1)
+
+        class _InterruptOnSecondEnter(type(cx_pool._max_connecting_cond)):
+            def __init__(self, lock):
+                super().__init__(lock)
+                self.enters = 0
+
+            def __enter__(self):
+                self.enters += 1
+                if self.enters == 2:
+                    # First enter is the checkout wait, second is connect()'s
+                    # cleanup. Simulate a kill delivered while waiting there.
+                    raise KeyboardInterrupt()
+                return super().__enter__()
+
+            def __exit__(self, *args):
+                return super().__exit__(*args)
+
+            def notify(self, n=1):
+                # The handler's pending-gate rollback is itself killed.
+                raise KeyboardInterrupt()
+
+        cx_pool._max_connecting_cond = _InterruptOnSecondEnter(cx_pool._max_connecting_cond._lock)
+
+        with patch.object(cx_pool, "connect", side_effect=asyncio.CancelledError()):
+            with self.assertRaises(KeyboardInterrupt):
+                with cx_pool.checkout():
+                    pass
+
+        self.assertEqual(0, cx_pool.requests)
+        self.assertEqual(0, cx_pool.active_sockets)
+        self.assertEqual(0, cx_pool._pending)
+
     def test_pool_removes_closed_socket(self):
         # Test that Pool removes explicitly closed socket.
         cx_pool = self.create_pool()

From a909572aad3199c59e30d4bf7025fc6e9dcd5fcd Mon Sep 17 00:00:00 2001
From: Steven Silvester <steve.silvester@mongodb.com>
Date: Thu, 1 Oct 2026 20:05:04 -0500
Subject: [PATCH 4/6] PYTHON-6136 Bind pool before the race test's try block

---
 test/asynchronous/test_client.py | 3 +--
 test/test_client.py              | 3 +--
 2 files changed, 2 insertions(+), 4 deletions(-)

diff --git a/test/asynchronous/test_client.py b/test/asynchronous/test_client.py
index b8ec8479d4..62133b814f 100644
--- a/test/asynchronous/test_client.py
+++ b/test/asynchronous/test_client.py
@@ -2765,9 +2765,9 @@ def _amplified_sleep(*args):
         coll = client.pymongo_test.coll
         coll.insert_one({})
 
+        pool = async_get_pool(client)  # type:ignore
         # Widen the post-gate checkout windows (PYTHON-6136).
         if amplify_seconds:
-            pool = async_get_pool(client)  # type:ignore
 
             class _AmplifiedCondition:
                 def __init__(self, cond, seconds):
@@ -2850,7 +2850,6 @@ def reaper():
             except Timeout:
                 self.fail("Pool gate saturated (PYTHON-6074)")
             self.assertGreater(op_count[0], 0)
-            pool = async_get_pool(client)  # type:ignore
         finally:
             running[0] = False
             gevent.killall(workers, block=False)
diff --git a/test/test_client.py b/test/test_client.py
index 9e7a41d506..d6e004fa75 100644
--- a/test/test_client.py
+++ b/test/test_client.py
@@ -2716,9 +2716,9 @@ def _amplified_sleep(*args):
         coll = client.pymongo_test.coll
         coll.insert_one({})
 
+        pool = get_pool(client)  # type:ignore
         # Widen the post-gate checkout windows (PYTHON-6136).
         if amplify_seconds:
-            pool = get_pool(client)  # type:ignore
 
             class _AmplifiedCondition:
                 def __init__(self, cond, seconds):
@@ -2801,7 +2801,6 @@ def reaper():
             except Timeout:
                 self.fail("Pool gate saturated (PYTHON-6074)")
             self.assertGreater(op_count[0], 0)
-            pool = get_pool(client)  # type:ignore
         finally:
             running[0] = False
             gevent.killall(workers, block=False)

From dbbc6150ebbde413be862b19040ea5fb804c17eb Mon Sep 17 00:00:00 2001
From: Steven Silvester <steve.silvester@mongodb.com>
Date: Thu, 1 Oct 2026 20:29:05 -0500
Subject: [PATCH 5/6] PYTHON-6136 Roll back operation_count on every failed
 checkout

---
 pymongo/asynchronous/pool.py      | 7 +++++++
 pymongo/synchronous/pool.py       | 7 +++++++
 test/asynchronous/test_pooling.py | 6 ++++++
 test/test_pooling.py              | 6 ++++++
 4 files changed, 26 insertions(+)

diff --git a/pymongo/asynchronous/pool.py b/pymongo/asynchronous/pool.py
index ec1c88a7e7..3e54c78b95 100644
--- a/pymongo/asynchronous/pool.py
+++ b/pymongo/asynchronous/pool.py
@@ -1003,13 +1003,18 @@ async def _get_conn(
                 try:
                     async with self.size_cond:
                         self.requests -= 1
+                        self.operation_count -= 1
                         accounted = True
                         self.size_cond.notify()
                 finally:
                     if not accounted:
                         async with self.size_cond:
                             self.requests -= 1
+                            self.operation_count -= 1
                             self.size_cond.notify()
+            else:
+                # The gate never admitted; still undo the load increment above.
+                self.operation_count -= 1
             raise
 
         # We've now acquired the semaphore and must release it on error.
@@ -1090,6 +1095,7 @@ async def _get_conn(
                 try:
                     async with self.size_cond:
                         self.requests -= 1
+                        self.operation_count -= 1
                         if incremented:
                             self.active_sockets -= 1
                         accounted = True
@@ -1098,6 +1104,7 @@ async def _get_conn(
                     if not accounted:
                         async with self.size_cond:
                             self.requests -= 1
+                            self.operation_count -= 1
                             if incremented:
                                 self.active_sockets -= 1
                             self.size_cond.notify()
diff --git a/pymongo/synchronous/pool.py b/pymongo/synchronous/pool.py
index a851f6a9b7..9ae94b8365 100644
--- a/pymongo/synchronous/pool.py
+++ b/pymongo/synchronous/pool.py
@@ -999,13 +999,18 @@ def _get_conn(
                 try:
                     with self.size_cond:
                         self.requests -= 1
+                        self.operation_count -= 1
                         accounted = True
                         self.size_cond.notify()
                 finally:
                     if not accounted:
                         with self.size_cond:
                             self.requests -= 1
+                            self.operation_count -= 1
                             self.size_cond.notify()
+            else:
+                # The gate never admitted; still undo the load increment above.
+                self.operation_count -= 1
             raise
 
         # We've now acquired the semaphore and must release it on error.
@@ -1086,6 +1091,7 @@ def _get_conn(
                 try:
                     with self.size_cond:
                         self.requests -= 1
+                        self.operation_count -= 1
                         if incremented:
                             self.active_sockets -= 1
                         accounted = True
@@ -1094,6 +1100,7 @@ def _get_conn(
                     if not accounted:
                         with self.size_cond:
                             self.requests -= 1
+                            self.operation_count -= 1
                             if incremented:
                                 self.active_sockets -= 1
                             self.size_cond.notify()
diff --git a/test/asynchronous/test_pooling.py b/test/asynchronous/test_pooling.py
index 8452d18b61..948f0648f4 100644
--- a/test/asynchronous/test_pooling.py
+++ b/test/asynchronous/test_pooling.py
@@ -306,6 +306,7 @@ def notify():
         # Accounting was applied exactly once.
         self.assertEqual(0, cx_pool.requests)
         self.assertEqual(0, cx_pool.active_sockets)
+        self.assertEqual(0, cx_pool.operation_count)
 
     async def test_checkout_error_accounting_on_kill_during_acquire(self):
         # PYTHON-6074: an exception delivered while the checkout error
@@ -341,6 +342,7 @@ async def __aexit__(self, *args):
         # The fallback applied the accounting exactly once.
         self.assertEqual(0, cx_pool.requests)
         self.assertEqual(0, cx_pool.active_sockets)
+        self.assertEqual(0, cx_pool.operation_count)
 
     async def test_checkout_error_accounting_on_connect_keyboard_interrupt(self):
         # PYTHON-6136: a KeyboardInterrupt from connect() must roll back the
@@ -355,6 +357,7 @@ async def test_checkout_error_accounting_on_connect_keyboard_interrupt(self):
         self.assertEqual(0, cx_pool.requests)
         self.assertEqual(0, cx_pool.active_sockets)
         self.assertEqual(0, cx_pool._pending)
+        self.assertEqual(0, cx_pool.operation_count)
 
     async def test_checkout_error_accounting_on_kill_during_pending_cleanup(self):
         # PYTHON-6136: an interruption during the pending-gate cleanup must
@@ -391,6 +394,7 @@ def notify(self, n=1):
         self.assertEqual(0, cx_pool.requests)
         self.assertEqual(0, cx_pool.active_sockets)
         self.assertEqual(0, cx_pool._pending)
+        self.assertEqual(0, cx_pool.operation_count)
 
     async def test_pool_removes_closed_socket(self):
         # Test that Pool removes explicitly closed socket.
@@ -513,6 +517,8 @@ async def test_wait_queue_timeout(self):
             1,
             f"Waited {duration:.2f} seconds for a socket, expected {wait_queue_timeout:f}",
         )
+        # The load metric must not be inflated by the failed checkout.
+        self.assertEqual(0, pool.operation_count)
 
     async def test_no_wait_queue_timeout(self):
         # Verify get_socket() with no wait_queue_timeout blocks forever.
diff --git a/test/test_pooling.py b/test/test_pooling.py
index 2e8e243cbf..fead75b6f8 100644
--- a/test/test_pooling.py
+++ b/test/test_pooling.py
@@ -306,6 +306,7 @@ def notify():
         # Accounting was applied exactly once.
         self.assertEqual(0, cx_pool.requests)
         self.assertEqual(0, cx_pool.active_sockets)
+        self.assertEqual(0, cx_pool.operation_count)
 
     def test_checkout_error_accounting_on_kill_during_acquire(self):
         # PYTHON-6074: an exception delivered while the checkout error
@@ -341,6 +342,7 @@ def __exit__(self, *args):
         # The fallback applied the accounting exactly once.
         self.assertEqual(0, cx_pool.requests)
         self.assertEqual(0, cx_pool.active_sockets)
+        self.assertEqual(0, cx_pool.operation_count)
 
     def test_checkout_error_accounting_on_connect_keyboard_interrupt(self):
         # PYTHON-6136: a KeyboardInterrupt from connect() must roll back the
@@ -355,6 +357,7 @@ def test_checkout_error_accounting_on_connect_keyboard_interrupt(self):
         self.assertEqual(0, cx_pool.requests)
         self.assertEqual(0, cx_pool.active_sockets)
         self.assertEqual(0, cx_pool._pending)
+        self.assertEqual(0, cx_pool.operation_count)
 
     def test_checkout_error_accounting_on_kill_during_pending_cleanup(self):
         # PYTHON-6136: an interruption during the pending-gate cleanup must
@@ -391,6 +394,7 @@ def notify(self, n=1):
         self.assertEqual(0, cx_pool.requests)
         self.assertEqual(0, cx_pool.active_sockets)
         self.assertEqual(0, cx_pool._pending)
+        self.assertEqual(0, cx_pool.operation_count)
 
     def test_pool_removes_closed_socket(self):
         # Test that Pool removes explicitly closed socket.
@@ -513,6 +517,8 @@ def test_wait_queue_timeout(self):
             1,
             f"Waited {duration:.2f} seconds for a socket, expected {wait_queue_timeout:f}",
         )
+        # The load metric must not be inflated by the failed checkout.
+        self.assertEqual(0, pool.operation_count)
 
     def test_no_wait_queue_timeout(self):
         # Verify get_socket() with no wait_queue_timeout blocks forever.

From 40762fa1f3beffef02139ef4c7f3a2ec742bb168 Mon Sep 17 00:00:00 2001
From: Steven Silvester <steve.silvester@mongodb.com>
Date: Thu, 1 Oct 2026 20:42:36 -0500
Subject: [PATCH 6/6] PYTHON-6136 Lock-protect the operation_count rollback on
 the gate path

---
 pymongo/asynchronous/pool.py | 10 +++++++++-
 pymongo/synchronous/pool.py  | 10 +++++++++-
 2 files changed, 18 insertions(+), 2 deletions(-)

diff --git a/pymongo/asynchronous/pool.py b/pymongo/asynchronous/pool.py
index 3e54c78b95..3c58cb924f 100644
--- a/pymongo/asynchronous/pool.py
+++ b/pymongo/asynchronous/pool.py
@@ -1014,7 +1014,15 @@ async def _get_conn(
                             self.size_cond.notify()
             else:
                 # The gate never admitted; still undo the load increment above.
-                self.operation_count -= 1
+                accounted = False
+                try:
+                    async with self.size_cond:
+                        self.operation_count -= 1
+                        accounted = True
+                finally:
+                    if not accounted:
+                        async with self.size_cond:
+                            self.operation_count -= 1
             raise
 
         # We've now acquired the semaphore and must release it on error.
diff --git a/pymongo/synchronous/pool.py b/pymongo/synchronous/pool.py
index 9ae94b8365..7dd5722df8 100644
--- a/pymongo/synchronous/pool.py
+++ b/pymongo/synchronous/pool.py
@@ -1010,7 +1010,15 @@ def _get_conn(
                             self.size_cond.notify()
             else:
                 # The gate never admitted; still undo the load increment above.
-                self.operation_count -= 1
+                accounted = False
+                try:
+                    with self.size_cond:
+                        self.operation_count -= 1
+                        accounted = True
+                finally:
+                    if not accounted:
+                        with self.size_cond:
+                            self.operation_count -= 1
             raise
 
         # We've now acquired the semaphore and must release it on error.