diff --git a/doc/changelog.rst b/doc/changelog.rst
index d2f49b7fd8..00a9b83c34 100644
--- a/doc/changelog.rst
+++ b/doc/changelog.rst
@@ -1,6 +1,18 @@
 Changelog
 =========
 
+Changes in Version 4.19.0 (2026/XX/XX)
+--------------------------------------
+
+Bug fixes
+.........
+
+- Fixed a bug where the synchronous client could permanently deadlock under
+  gevent when a greenlet was killed while checking a connection back into
+  the pool (`PYTHON-6074`_).
+
+.. _PYTHON-6074: https://jira.mongodb.org/browse/PYTHON-6074
+
 Changes in Version 4.18.0 (2026/09/03)
 --------------------------------------
 
diff --git a/pymongo/asynchronous/pool.py b/pymongo/asynchronous/pool.py
index b8dd042dc6..434af0ef5a 100644
--- a/pymongo/asynchronous/pool.py
+++ b/pymongo/asynchronous/pool.py
@@ -1044,11 +1044,24 @@ async def _get_conn(
             if conn:
                 # We checked out a socket but authentication failed.
                 await conn.close_conn(ConnectionClosedReason.ERROR)
-            async with self.size_cond:
-                self.requests -= 1
-                if incremented:
-                    self.active_sockets -= 1
-                self.size_cond.notify()
+            # Re-apply the accounting if a GreenletExit interrupts
+            # during the size_cond acquisition; during unwind gevent
+            # lets the re-acquire complete (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()
+            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(
@@ -1060,6 +1073,40 @@ async def _get_conn(
 
         return conn
 
+    def _checkin_apply(
+        self, conn: AsyncConnection, txn: bool, cursor: bool, forked: bool
+    ) -> tuple[Optional[str], bool, bool]:
+        """Apply checkin accounting; caller holds ``size_cond``.
+
+        No cooperative I/O, so safe while a gevent greenlet unwinds. Returns
+        ``(close_conn_reason, emit_closed, appended)`` for outside the lock.
+        """
+        self.active_contexts.discard(conn.cancel_context)
+        if txn:
+            self.ntxns -= 1
+        elif cursor:
+            self.ncursors -= 1
+        self.requests -= 1
+        self.active_sockets -= 1
+        self.operation_count -= 1
+        close_conn_reason: Optional[str] = None
+        emit_closed = False
+        appended = False
+        if not forked:
+            if self.closed:
+                close_conn_reason = ConnectionClosedReason.POOL_CLOSED
+            elif conn.closed:
+                # CMAP requires the closed event be emitted after the check in.
+                emit_closed = True
+            elif self.stale_generation(conn.generation, conn.service_id):
+                close_conn_reason = ConnectionClosedReason.STALE
+            else:
+                conn.update_last_checkin_time()
+                conn.update_is_writable(bool(self.is_writable))
+                self.conns.appendleft(conn)
+                appended = True
+        return close_conn_reason, emit_closed, appended
+
     async def checkin(self, conn: AsyncConnection) -> None:
         """Return the connection to the pool, or if it's closed discard it.
 
@@ -1071,44 +1118,41 @@ async def checkin(self, conn: AsyncConnection) -> None:
         conn.pinned_txn = False
         conn.pinned_cursor = False
         self._pinned_sockets.discard(conn)
-        async with self.lock:
-            self.active_contexts.discard(conn.cancel_context)
+        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).
+        close_conn_reason: Optional[str] = None
+        emit_closed = False
+        accounted = False
+        try:
+            async with self.size_cond:
+                close_conn_reason, emit_closed, appended = self._checkin_apply(
+                    conn, txn, cursor, forked
+                )
+                accounted = True
+                if appended:
+                    # Notify any threads waiting to create a connection.
+                    self._max_connecting_cond.notify()
+                self.size_cond.notify()
+        finally:
+            if not accounted:
+                async with self.size_cond:
+                    close_conn_reason, emit_closed, appended = self._checkin_apply(
+                        conn, txn, cursor, forked
+                    )
+                    if appended:
+                        self._max_connecting_cond.notify()
+                    self.size_cond.notify()
         telemetry = self._telemetry
         if telemetry._should_publish or (telemetry._log and _is_debug_enabled(_CONNECTION_LOGGER)):
             telemetry.checked_in(conn.id)
-        if self.pid != os.getpid():
+        if emit_closed:
+            telemetry.connection_closed(conn.id, ConnectionClosedReason.ERROR)
+        if forked:
             await self.reset_without_pause()
-        else:
-            if self.closed:
-                await conn.close_conn(ConnectionClosedReason.POOL_CLOSED)
-            elif conn.closed:
-                # CMAP requires the closed event be emitted after the check in.
-                self._telemetry.connection_closed(conn.id, ConnectionClosedReason.ERROR)
-            else:
-                close_conn = False
-                async with self.lock:
-                    # Hold the lock to ensure this section does not race with
-                    # Pool.reset().
-                    if self.stale_generation(conn.generation, conn.service_id):
-                        close_conn = True
-                    else:
-                        conn.update_last_checkin_time()
-                        conn.update_is_writable(bool(self.is_writable))
-                        self.conns.appendleft(conn)
-                        # Notify any threads waiting to create a connection.
-                        self._max_connecting_cond.notify()
-                if close_conn:
-                    await conn.close_conn(ConnectionClosedReason.STALE)
-
-        async with self.size_cond:
-            if txn:
-                self.ntxns -= 1
-            elif cursor:
-                self.ncursors -= 1
-            self.requests -= 1
-            self.active_sockets -= 1
-            self.operation_count -= 1
-            self.size_cond.notify()
+        elif close_conn_reason is not None:
+            await conn.close_conn(close_conn_reason)
 
     async def _perished(self, conn: AsyncConnection) -> bool:
         """Return True and close the connection if it is "perished".
diff --git a/pymongo/synchronous/pool.py b/pymongo/synchronous/pool.py
index e2a708e18f..255dd09bf0 100644
--- a/pymongo/synchronous/pool.py
+++ b/pymongo/synchronous/pool.py
@@ -1040,11 +1040,24 @@ def _get_conn(
             if conn:
                 # We checked out a socket but authentication failed.
                 conn.close_conn(ConnectionClosedReason.ERROR)
-            with self.size_cond:
-                self.requests -= 1
-                if incremented:
-                    self.active_sockets -= 1
-                self.size_cond.notify()
+            # Re-apply the accounting if a GreenletExit interrupts
+            # during the size_cond acquisition; during unwind gevent
+            # lets the re-acquire complete (PYTHON-6074).
+            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(
@@ -1056,6 +1069,40 @@ def _get_conn(
 
         return conn
 
+    def _checkin_apply(
+        self, conn: Connection, txn: bool, cursor: bool, forked: bool
+    ) -> tuple[Optional[str], bool, bool]:
+        """Apply checkin accounting; caller holds ``size_cond``.
+
+        No cooperative I/O, so safe while a gevent greenlet unwinds. Returns
+        ``(close_conn_reason, emit_closed, appended)`` for outside the lock.
+        """
+        self.active_contexts.discard(conn.cancel_context)
+        if txn:
+            self.ntxns -= 1
+        elif cursor:
+            self.ncursors -= 1
+        self.requests -= 1
+        self.active_sockets -= 1
+        self.operation_count -= 1
+        close_conn_reason: Optional[str] = None
+        emit_closed = False
+        appended = False
+        if not forked:
+            if self.closed:
+                close_conn_reason = ConnectionClosedReason.POOL_CLOSED
+            elif conn.closed:
+                # CMAP requires the closed event be emitted after the check in.
+                emit_closed = True
+            elif self.stale_generation(conn.generation, conn.service_id):
+                close_conn_reason = ConnectionClosedReason.STALE
+            else:
+                conn.update_last_checkin_time()
+                conn.update_is_writable(bool(self.is_writable))
+                self.conns.appendleft(conn)
+                appended = True
+        return close_conn_reason, emit_closed, appended
+
     def checkin(self, conn: Connection) -> None:
         """Return the connection to the pool, or if it's closed discard it.
 
@@ -1067,44 +1114,41 @@ def checkin(self, conn: Connection) -> None:
         conn.pinned_txn = False
         conn.pinned_cursor = False
         self._pinned_sockets.discard(conn)
-        with self.lock:
-            self.active_contexts.discard(conn.cancel_context)
+        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).
+        close_conn_reason: Optional[str] = None
+        emit_closed = False
+        accounted = False
+        try:
+            with self.size_cond:
+                close_conn_reason, emit_closed, appended = self._checkin_apply(
+                    conn, txn, cursor, forked
+                )
+                accounted = True
+                if appended:
+                    # Notify any threads waiting to create a connection.
+                    self._max_connecting_cond.notify()
+                self.size_cond.notify()
+        finally:
+            if not accounted:
+                with self.size_cond:
+                    close_conn_reason, emit_closed, appended = self._checkin_apply(
+                        conn, txn, cursor, forked
+                    )
+                    if appended:
+                        self._max_connecting_cond.notify()
+                    self.size_cond.notify()
         telemetry = self._telemetry
         if telemetry._should_publish or (telemetry._log and _is_debug_enabled(_CONNECTION_LOGGER)):
             telemetry.checked_in(conn.id)
-        if self.pid != os.getpid():
+        if emit_closed:
+            telemetry.connection_closed(conn.id, ConnectionClosedReason.ERROR)
+        if forked:
             self.reset_without_pause()
-        else:
-            if self.closed:
-                conn.close_conn(ConnectionClosedReason.POOL_CLOSED)
-            elif conn.closed:
-                # CMAP requires the closed event be emitted after the check in.
-                self._telemetry.connection_closed(conn.id, ConnectionClosedReason.ERROR)
-            else:
-                close_conn = False
-                with self.lock:
-                    # Hold the lock to ensure this section does not race with
-                    # Pool.reset().
-                    if self.stale_generation(conn.generation, conn.service_id):
-                        close_conn = True
-                    else:
-                        conn.update_last_checkin_time()
-                        conn.update_is_writable(bool(self.is_writable))
-                        self.conns.appendleft(conn)
-                        # Notify any threads waiting to create a connection.
-                        self._max_connecting_cond.notify()
-                if close_conn:
-                    conn.close_conn(ConnectionClosedReason.STALE)
-
-        with self.size_cond:
-            if txn:
-                self.ntxns -= 1
-            elif cursor:
-                self.ncursors -= 1
-            self.requests -= 1
-            self.active_sockets -= 1
-            self.operation_count -= 1
-            self.size_cond.notify()
+        elif close_conn_reason is not None:
+            conn.close_conn(close_conn_reason)
 
     def _perished(self, conn: Connection) -> bool:
         """Return True and close the connection if it is "perished".
diff --git a/test/asynchronous/test_client.py b/test/asynchronous/test_client.py
index c7ddd8b559..b1dbd461be 100644
--- a/test/asynchronous/test_client.py
+++ b/test/asynchronous/test_client.py
@@ -2624,6 +2624,123 @@ def timeout_task():
         # Unpatch the instance.
         del pool.connect
 
+    @async_client_context.require_sync
+    def test_gevent_kill_churn_deadlock(self):
+        """A greenlet killed mid-operation must not leave the pool deadlocked (PYTHON-6074)."""
+        if not gevent_monkey_patched():
+            raise SkipTest("Must be running monkey patched by gevent")
+        import random
+
+        import gevent
+        import gevent.thread as _gthread
+        from gevent import Timeout, spawn
+
+        # PYTHON-6074: widen gevent's courtesy-yield window (the bare
+        # sleep() on a failed non-blocking lock acquire, which
+        # Condition.notify()'s _is_owned() hits on every checkin) so a
+        # GreenletExit lands in that window deterministically. Synthetic:
+        # only the bare sleep() is widened; sleep(t) with args passes
+        # through, so the reaper and watchdog timings are unchanged. Set
+        # AMPLIFY_RACE=1 to enable; the unfixed test then deadlocks
+        # within seconds every run.
+        if os.environ.get("AMPLIFY_RACE", "0") == "1":
+            _AMPLIFY_SECONDS = float(os.environ.get("AMPLIFY_SECONDS", "0.02"))
+
+            _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
+                    _orig_thread_sleep(*args)
+
+            _gthread.sleep = _amplified_sleep
+            self.addCleanup(setattr, _gthread, "sleep", _orig_thread_sleep)
+
+        client = self.async_rs_or_single_client(maxPoolSize=2)
+        coll = client.pymongo_test.coll
+        coll.insert_one({})
+
+        op_count = [0]
+        running = [True]
+        workers: list = []
+
+        def worker():
+            while running[0]:
+                try:
+                    coll.find_one({})
+                    op_count[0] += 1
+                    time.sleep(0.001)
+                except Exception:
+                    return
+
+        def reaper():
+            while running[0]:
+                time.sleep(0.003)
+                if not workers:
+                    continue
+                idx = random.randrange(len(workers))
+                try:
+                    workers[idx].kill(block=False)
+                except Exception:
+                    pass
+                workers[idx] = spawn(worker)
+
+        workers[:] = [spawn(worker) for _ in range(8)]
+        reaper_gr = spawn(reaper)
+        try:
+            # Watchdog: fail (never hang) if no op completes for 8s while
+            # workers are alive. Without the fix the op counter freezes
+            # permanently once the size gate saturates; with the fix, ops keep
+            # flowing and the loop simply runs out the deadline.
+            last = op_count[0]
+            stale = 0.0
+            deadline = time.monotonic() + 12
+            while time.monotonic() < deadline:
+                time.sleep(2)
+                if op_count[0] == last:
+                    stale += 2
+                    if stale >= 8:
+                        self.fail(
+                            "Deadlock detected (PYTHON-6074): no ops completed "
+                            f"for {stale:.0f}s, op_count={op_count[0]}"
+                        )
+                else:
+                    stale = 0.0
+                    last = op_count[0]
+            # Direct liveness probe that does not depend on *when* the leak
+            # landed: on the fixed driver ops keep flowing, so this completes
+            # well within the timeout; on the unfixed driver a saturated size
+            # gate makes it block, the timeout fires and the test fails.
+            try:
+                with Timeout(3):
+                    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)
+        finally:
+            running[0] = False
+            gevent.killall(workers, block=False)
+            try:
+                reaper_gr.kill(block=False)
+            except Exception:
+                pass
+            # Close the client but never hang on a wedged pool: without the
+            # PYTHON-6074 fix the pool's size gate is saturated and close() can
+            # block forever, so bound it and let the watchdog's self.fail()
+            # propagate.
+            try:
+                with Timeout(5):
+                    client.close()
+            except Timeout:
+                pass
+
 
 class TestClientLazyConnect(AsyncIntegrationTest):
     """Test concurrent operations on a lazily-connecting MongoClient."""
diff --git a/test/asynchronous/test_pooling.py b/test/asynchronous/test_pooling.py
index ef670774c8..661bd4e3d2 100644
--- a/test/asynchronous/test_pooling.py
+++ b/test/asynchronous/test_pooling.py
@@ -277,6 +277,71 @@ def add(self, item):
         self.assertEqual(0, cx_pool.active_sockets)
         self.assertEqual(0, cx_pool.requests)
 
+    async def test_checkout_error_accounting_no_double_decrement(self):
+        # PYTHON-6074: an exception delivered while the checkout error handler
+        # is inside size_cond.notify() (a yield point under gevent, where a
+        # greenlet can be killed) must not cause the accounting to be applied
+        # a second time by the handler's fallback.
+        cx_pool = await self.create_pool(max_pool_size=1)
+
+        real_notify = cx_pool.size_cond.notify
+        notify_calls = []
+
+        def notify():
+            notify_calls.append(1)
+            if len(notify_calls) == 1:
+                # Simulate a kill delivered at the notify() yield point.
+                raise KeyboardInterrupt()
+            real_notify()
+
+        cx_pool.size_cond.notify = notify
+        try:
+            with patch.object(cx_pool, "connect", side_effect=asyncio.CancelledError()):
+                with self.assertRaises(KeyboardInterrupt):
+                    async with cx_pool.checkout():
+                        pass
+        finally:
+            cx_pool.size_cond.notify = real_notify
+
+        # Accounting was applied exactly once.
+        self.assertEqual(0, cx_pool.requests)
+        self.assertEqual(0, cx_pool.active_sockets)
+
+    async def test_checkout_error_accounting_on_kill_during_acquire(self):
+        # PYTHON-6074: an exception delivered while the checkout error
+        # handler is waiting to acquire size_cond (a yield point under
+        # gevent) must not leak the checkout accounting; the handler's
+        # fallback re-applies it.
+        cx_pool = await self.create_pool(max_pool_size=1)
+
+        class _InterruptOnSecondEnter(type(cx_pool.size_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 semaphore, second is the
+                    # error handler. Simulate a kill delivered while blocked
+                    # on the second.
+                    raise KeyboardInterrupt()
+                return await super().__aenter__()
+
+            async def __aexit__(self, *args):
+                return await super().__aexit__(*args)
+
+        cx_pool.size_cond = _InterruptOnSecondEnter(cx_pool.size_cond._lock)
+
+        with patch.object(cx_pool, "connect", side_effect=asyncio.CancelledError()):
+            with self.assertRaises(KeyboardInterrupt):
+                async with cx_pool.checkout():
+                    pass
+
+        # The fallback applied the accounting exactly once.
+        self.assertEqual(0, cx_pool.requests)
+        self.assertEqual(0, cx_pool.active_sockets)
+
     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_client.py b/test/test_client.py
index 20cceeb037..39fec44f82 100644
--- a/test/test_client.py
+++ b/test/test_client.py
@@ -2575,6 +2575,123 @@ def timeout_task():
         # Unpatch the instance.
         del pool.connect
 
+    @client_context.require_sync
+    def test_gevent_kill_churn_deadlock(self):
+        """A greenlet killed mid-operation must not leave the pool deadlocked (PYTHON-6074)."""
+        if not gevent_monkey_patched():
+            raise SkipTest("Must be running monkey patched by gevent")
+        import random
+
+        import gevent
+        import gevent.thread as _gthread
+        from gevent import Timeout, spawn
+
+        # PYTHON-6074: widen gevent's courtesy-yield window (the bare
+        # sleep() on a failed non-blocking lock acquire, which
+        # Condition.notify()'s _is_owned() hits on every checkin) so a
+        # GreenletExit lands in that window deterministically. Synthetic:
+        # only the bare sleep() is widened; sleep(t) with args passes
+        # through, so the reaper and watchdog timings are unchanged. Set
+        # AMPLIFY_RACE=1 to enable; the unfixed test then deadlocks
+        # within seconds every run.
+        if os.environ.get("AMPLIFY_RACE", "0") == "1":
+            _AMPLIFY_SECONDS = float(os.environ.get("AMPLIFY_SECONDS", "0.02"))
+
+            _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
+                    _orig_thread_sleep(*args)
+
+            _gthread.sleep = _amplified_sleep
+            self.addCleanup(setattr, _gthread, "sleep", _orig_thread_sleep)
+
+        client = self.rs_or_single_client(maxPoolSize=2)
+        coll = client.pymongo_test.coll
+        coll.insert_one({})
+
+        op_count = [0]
+        running = [True]
+        workers: list = []
+
+        def worker():
+            while running[0]:
+                try:
+                    coll.find_one({})
+                    op_count[0] += 1
+                    time.sleep(0.001)
+                except Exception:
+                    return
+
+        def reaper():
+            while running[0]:
+                time.sleep(0.003)
+                if not workers:
+                    continue
+                idx = random.randrange(len(workers))
+                try:
+                    workers[idx].kill(block=False)
+                except Exception:
+                    pass
+                workers[idx] = spawn(worker)
+
+        workers[:] = [spawn(worker) for _ in range(8)]
+        reaper_gr = spawn(reaper)
+        try:
+            # Watchdog: fail (never hang) if no op completes for 8s while
+            # workers are alive. Without the fix the op counter freezes
+            # permanently once the size gate saturates; with the fix, ops keep
+            # flowing and the loop simply runs out the deadline.
+            last = op_count[0]
+            stale = 0.0
+            deadline = time.monotonic() + 12
+            while time.monotonic() < deadline:
+                time.sleep(2)
+                if op_count[0] == last:
+                    stale += 2
+                    if stale >= 8:
+                        self.fail(
+                            "Deadlock detected (PYTHON-6074): no ops completed "
+                            f"for {stale:.0f}s, op_count={op_count[0]}"
+                        )
+                else:
+                    stale = 0.0
+                    last = op_count[0]
+            # Direct liveness probe that does not depend on *when* the leak
+            # landed: on the fixed driver ops keep flowing, so this completes
+            # well within the timeout; on the unfixed driver a saturated size
+            # gate makes it block, the timeout fires and the test fails.
+            try:
+                with Timeout(3):
+                    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)
+        finally:
+            running[0] = False
+            gevent.killall(workers, block=False)
+            try:
+                reaper_gr.kill(block=False)
+            except Exception:
+                pass
+            # Close the client but never hang on a wedged pool: without the
+            # PYTHON-6074 fix the pool's size gate is saturated and close() can
+            # block forever, so bound it and let the watchdog's self.fail()
+            # propagate.
+            try:
+                with Timeout(5):
+                    client.close()
+            except Timeout:
+                pass
+
 
 class TestClientLazyConnect(IntegrationTest):
     """Test concurrent operations on a lazily-connecting MongoClient."""
diff --git a/test/test_pooling.py b/test/test_pooling.py
index 64e6aa931f..a3f0eaf589 100644
--- a/test/test_pooling.py
+++ b/test/test_pooling.py
@@ -277,6 +277,71 @@ def add(self, item):
         self.assertEqual(0, cx_pool.active_sockets)
         self.assertEqual(0, cx_pool.requests)
 
+    def test_checkout_error_accounting_no_double_decrement(self):
+        # PYTHON-6074: an exception delivered while the checkout error handler
+        # is inside size_cond.notify() (a yield point under gevent, where a
+        # greenlet can be killed) must not cause the accounting to be applied
+        # a second time by the handler's fallback.
+        cx_pool = self.create_pool(max_pool_size=1)
+
+        real_notify = cx_pool.size_cond.notify
+        notify_calls = []
+
+        def notify():
+            notify_calls.append(1)
+            if len(notify_calls) == 1:
+                # Simulate a kill delivered at the notify() yield point.
+                raise KeyboardInterrupt()
+            real_notify()
+
+        cx_pool.size_cond.notify = notify
+        try:
+            with patch.object(cx_pool, "connect", side_effect=asyncio.CancelledError()):
+                with self.assertRaises(KeyboardInterrupt):
+                    with cx_pool.checkout():
+                        pass
+        finally:
+            cx_pool.size_cond.notify = real_notify
+
+        # Accounting was applied exactly once.
+        self.assertEqual(0, cx_pool.requests)
+        self.assertEqual(0, cx_pool.active_sockets)
+
+    def test_checkout_error_accounting_on_kill_during_acquire(self):
+        # PYTHON-6074: an exception delivered while the checkout error
+        # handler is waiting to acquire size_cond (a yield point under
+        # gevent) must not leak the checkout accounting; the handler's
+        # fallback re-applies it.
+        cx_pool = self.create_pool(max_pool_size=1)
+
+        class _InterruptOnSecondEnter(type(cx_pool.size_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 semaphore, second is the
+                    # error handler. Simulate a kill delivered while blocked
+                    # on the second.
+                    raise KeyboardInterrupt()
+                return super().__enter__()
+
+            def __exit__(self, *args):
+                return super().__exit__(*args)
+
+        cx_pool.size_cond = _InterruptOnSecondEnter(cx_pool.size_cond._lock)
+
+        with patch.object(cx_pool, "connect", side_effect=asyncio.CancelledError()):
+            with self.assertRaises(KeyboardInterrupt):
+                with cx_pool.checkout():
+                    pass
+
+        # The fallback applied the accounting exactly once.
+        self.assertEqual(0, cx_pool.requests)
+        self.assertEqual(0, cx_pool.active_sockets)
+
     def test_pool_removes_closed_socket(self):
         # Test that Pool removes explicitly closed socket.
         cx_pool = self.create_pool()