diff --git a/redis/connection.py b/redis/connection.py
index 7ae5d82e1c..bc6904bc43 100644
--- a/redis/connection.py
+++ b/redis/connection.py
@@ -1576,6 +1576,14 @@ def _connect(self):
         # we want to mimic what socket.create_connection does to support
         # ipv4/ipv6, but we want to set options prior to calling
         # socket.connect()
+
+        # Last caught connection error.
+        # Re-thrown if we are unable to connect to any of the options returned
+        # by getaddrinfo.
+        # Note that we must clear this variable before returning - otherwise,
+        # a caught err's traceback points to this frame, which points to err.
+        # Clearing this lets refcounting reclaim the exception immediately
+        # without deferring to the python garbage collector.
         err = None
 
         for res in socket.getaddrinfo(
@@ -1602,6 +1610,10 @@ def _connect(self):
 
                 # set the socket_timeout now that we're connected
                 sock.settimeout(self.socket_timeout)
+
+                # If a previous connection attempt failed, clear the error
+                err = None
+
                 return sock
 
             except OSError as _:
@@ -1614,7 +1626,11 @@ def _connect(self):
                     sock.close()
 
         if err is not None:
-            raise err
+            try:
+                raise err
+            finally:
+                # Ensure we clear local references to caught exceptions
+                err = None
         raise OSError("socket.getaddrinfo returned an empty list")
 
     def _host_error(self):
diff --git a/tests/test_connection.py b/tests/test_connection.py
index f232615b6c..bbed141006 100644
--- a/tests/test_connection.py
+++ b/tests/test_connection.py
@@ -5,6 +5,7 @@
 import selectors
 import socket
 import ssl
+import sys
 import threading
 import time
 import types
@@ -574,6 +575,69 @@ def test_disconnect__close_OSError(self):
         mock_sock.close.assert_called_once()
         assert conn._sock is None
 
+    def test_connect_breaks_exception_reference_cycle(self):
+        """
+        On connection failure, _connect must not leave its frame retaining the
+        raised exception.
+        The exception's traceback references the frame, so a retained local
+        would form a cycle only reclaimable by the GC.
+        The finally clause in _connect breaks it by clearing the local.
+        """
+        conn = Connection(host="localhost", port=6379)
+        addr_info = (socket.AF_INET, socket.SOCK_STREAM, 0, "", ("127.0.0.1", 6379))
+        with (
+            patch.object(socket, "getaddrinfo", return_value=[addr_info]),
+            patch.object(socket, "socket") as socket_factory,
+        ):
+            socket_factory.return_value.connect.side_effect = OSError("refused")
+            with pytest.raises(OSError) as exc_info:
+                conn._connect()
+
+        # Locate the _connect frame in the propagated traceback and confirm its
+        # err local was cleared, proving no exception<->frame cycle survives.
+        connect_frame = None
+        tb = exc_info.value.__traceback__
+        while tb is not None:
+            if tb.tb_frame.f_code.co_name == "_connect":
+                connect_frame = tb.tb_frame
+            tb = tb.tb_next
+        assert connect_frame is not None
+        assert connect_frame.f_locals.get("err") is None
+
+    def test_connect_breaks_reference_cycle_when_a_later_address_succeeds(self):
+        """
+        When an address fails but a later one connects, _connect must not
+        return while still holding the caught exception.
+        The exception's traceback references the frame, so a retained local
+        would form a cycle only reclaimable by the GC.
+        """
+        conn = Connection(host="localhost", port=6379)
+        addr_infos = [
+            (socket.AF_INET, socket.SOCK_STREAM, 0, "", ("127.0.0.1", 6379)),
+            (socket.AF_INET, socket.SOCK_STREAM, 0, "", ("127.0.0.2", 6379)),
+        ]
+
+        # _connect calls getaddrinfo, so hook that as a way to peek the caller
+        connect_frames = []
+
+        def capturing_getaddrinfo(*args, **kwargs):
+            connect_frames.append(sys._getframe(1))
+            return addr_infos
+
+        failing_sock, working_sock = MagicMock(), MagicMock()
+        failing_sock.connect.side_effect = OSError("refused")
+
+        with (
+            patch.object(socket, "getaddrinfo", capturing_getaddrinfo),
+            patch.object(socket, "socket", side_effect=[failing_sock, working_sock]),
+        ):
+            assert conn._connect() is working_sock
+
+        # Error should be cleared in the connect frame
+        assert len(connect_frames) == 1
+        (connect_frame,) = connect_frames
+        assert connect_frame.f_locals.get("err") is None
+
     @pytest.mark.parametrize(
         "connection_kwargs",
         [