Fix Sentinel pool capacity loss after failover

redis/connection.py

@@ -3301,10 +3301,14 @@ def release(self, connection: "Connection") -> None:
             else:
                 # Pool doesn't own this connection, do not add it back
                 # to the pool.
-                # The created connections count should not be changed,
-                # because the connection was not created by the pool.
                 # Still need to decrement USED since it was counted in get_connection()
                 connection.disconnect()
+                # Subclasses such as SentinelConnectionPool can override
+                # owns_connection() with a comparison different from local PID
+                # ownership. When such a subclass rejects a connection, also require
+                # connection.pid == self.pid before reclaiming its slot.
+                if connection.pid == self.pid:
+                    self._created_connections -= 1
                 record_connection_count(
                     pool_name="unknown_pool",
                     connection_state=ConnectionState.USED,

tests/test_sentinel.py

@@ -1,3 +1,4 @@
+import os
 import socket
 from unittest import mock
 
@@ -259,6 +260,47 @@ def test_slave_round_robin(cluster, sentinel, master_ip):
         next(rotator)
 
 
+@pytest.mark.fixed_client
+@pytest.mark.onlynoncluster
+def test_master_failover_reclaims_discarded_connection_slot():
+    master_a = ("master-a", 6379)
+    master_b = ("master-b", 6379)
+
+    class FakeConnection:
+        def __init__(self, **kwargs):
+            self.host, self.port = master_a
+            self.pid = os.getpid()
+
+        def connect(self):
+            pass
+
+        def disconnect(self):
+            pass
+
+        def can_read(self, timeout=0):
+            return False
+
+        def should_reconnect(self):
+            return False
+
+    pool = SentinelConnectionPool(
+        "mymaster",
+        mock.MagicMock(),
+        connection_class=FakeConnection,
+        max_connections=2,
+    )
+    pool.proxy.master_address = master_a
+
+    for _ in range(pool.max_connections):
+        connection = pool.get_connection()
+        pool.proxy.master_address = master_b
+        pool.release(connection)
+        pool.proxy.master_address = master_a
+
+    assert pool._created_connections == 0
+    pool.get_connection()
+
+
 @pytest.mark.onlynoncluster
 def test_ckquorum(sentinel):
     resp = sentinel.sentinel_ckquorum("mymaster")