Pool: prevent trimming the last idle connection under load

asyncpg/pool.py

@@ -287,12 +287,70 @@ def _maybe_cancel_inactive_callback(self) -> None:
             self._inactive_callback.cancel()
             self._inactive_callback = None
 
+    def _can_deactivate_inactive_connection(self) -> bool:
+        """Return True if an idle connection may be deactivated (trimmed).
+
+        Constraints:
+        - Do not trim if there are waiters in the pool queue (including acquiring).
+        - Do not trim below pool min size (leave at least `minsize` open connections).
+        - Keep at least one idle connection available (i.e., at least 2 idle holders so
+          trimming one still leaves one idle).
+        """
+        pool = getattr(self, "_pool", None)
+        if pool is None:
+            # No pool state; allow default trimming behavior.
+            return True
+
+        # Follow original logic: if pool is closing, handle as default (allow trim).
+        if getattr(pool, "_closing", False):
+            return True
+
+        minsize = int(getattr(pool, "_minsize", 0) or 0)
+
+        # Compute the number of tasks waiting to acquire a connection.
+        q = getattr(pool, "_queue", None)
+        if q is not None:
+            getters = getattr(q, "_getters", None)
+            waiters = len(getters) if getters is not None else 0
+        else:
+            waiters = 0
+
+        # Include tasks currently in the process of acquiring.
+        waiters += int(getattr(pool, "_acquiring", 0) or 0)
+
+        # Count open (live) connections and how many of them are idle.
+        open_conns = 0
+        idle = 0
+        holders = list(getattr(pool, "_holders", []) or [])
+        for h in holders:
+            if getattr(h, "_con", None) is not None:
+                open_conns += 1
+                if not getattr(h, "_in_use", None):
+                    idle += 1
+
+        # Conditions to allow trimming one idle connection:
+        # - No waiters.
+        # - Trimming one won't drop below minsize (so open_conns - 1 >= minsize).
+        # - After trimming one idle, at least one idle remains (so idle >= 2).
+        return (
+                waiters == 0 and
+                (open_conns - 1) >= minsize and
+                idle >= 2
+        )
+
     def _deactivate_inactive_connection(self) -> None:
         if self._in_use is not None:
             raise exceptions.InternalClientError(
                 'attempting to deactivate an acquired connection')
 
         if self._con is not None:
+            # Only deactivate if doing so respects pool size and demand constraints.
+            if not self._can_deactivate_inactive_connection():
+                # Still mark this holder as available and keep the connection.
+                # Re-arm the inactivity timer so we can reevaluate later.
+                self._setup_inactive_callback()
+                return
+
             # The connection is idle and not in use, so it's fine to
             # use terminate() instead of close().
             self._con.terminate()