diff --git a/CHANGES/10464.bugfix.rst b/CHANGES/10464.bugfix.rst
new file mode 100644
index 00000000000..4e21000a317
--- /dev/null
+++ b/CHANGES/10464.bugfix.rst
@@ -0,0 +1 @@
+Changed connection creation to explicitly close sockets if an exception is raised in the event loop's ``create_connection`` method -- by :user:`top-oai`.
diff --git a/CONTRIBUTORS.txt b/CONTRIBUTORS.txt
index 03e35ce789f..19cca5b8c99 100644
--- a/CONTRIBUTORS.txt
+++ b/CONTRIBUTORS.txt
@@ -43,6 +43,7 @@ Andrej Antonov
 Andrew Leech
 Andrew Lytvyn
 Andrew Svetlov
+Andrew Top
 Andrew Zhou
 Andrii Soldatenko
 Anes Abismail
diff --git a/aiohttp/connector.py b/aiohttp/connector.py
index 263a4823334..99b281b24a5 100644
--- a/aiohttp/connector.py
+++ b/aiohttp/connector.py
@@ -1101,6 +1101,7 @@ async def _wrap_create_connection(
         client_error: Type[Exception] = ClientConnectorError,
         **kwargs: Any,
     ) -> Tuple[asyncio.Transport, ResponseHandler]:
+        sock: Union[socket.socket, None] = None
         try:
             async with ceil_timeout(
                 timeout.sock_connect, ceil_threshold=timeout.ceil_threshold
@@ -1112,7 +1113,11 @@ async def _wrap_create_connection(
                     interleave=self._interleave,
                     loop=self._loop,
                 )
-                return await self._loop.create_connection(*args, **kwargs, sock=sock)
+                connection = await self._loop.create_connection(
+                    *args, **kwargs, sock=sock
+                )
+                sock = None
+                return connection
         except cert_errors as exc:
             raise ClientConnectorCertificateError(req.connection_key, exc) from exc
         except ssl_errors as exc:
@@ -1121,6 +1126,12 @@ async def _wrap_create_connection(
             if exc.errno is None and isinstance(exc, asyncio.TimeoutError):
                 raise
             raise client_error(req.connection_key, exc) from exc
+        finally:
+            if sock is not None:
+                # Will be hit if an exception is thrown before the event loop takes the socket.
+                # In that case, proactively close the socket to guard against event loop leaks.
+                # For example, see https://github.com/MagicStack/uvloop/issues/653.
+                sock.close()
 
     def _warn_about_tls_in_tls(
         self,
diff --git a/tests/conftest.py b/tests/conftest.py
index e24d05b31ff..6e4a93d2937 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -222,6 +222,7 @@ def start_connection() -> Iterator[mock.Mock]:
         "aiohttp.connector.aiohappyeyeballs.start_connection",
         autospec=True,
         spec_set=True,
+        return_value=mock.create_autospec(socket.socket, spec_set=True, instance=True),
     ) as start_connection_mock:
         yield start_connection_mock
 
diff --git a/tests/test_connector.py b/tests/test_connector.py
index 61b452eee81..ebdb1a6b513 100644
--- a/tests/test_connector.py
+++ b/tests/test_connector.py
@@ -646,6 +646,29 @@ async def test_tcp_connector_certificate_error(
     await conn.close()
 
 
+async def test_tcp_connector_closes_socket_on_error(
+    loop: asyncio.AbstractEventLoop, start_connection: mock.AsyncMock
+) -> None:
+    req = ClientRequest("GET", URL("https://127.0.0.1:443"), loop=loop)
+
+    conn = aiohttp.TCPConnector()
+    with (
+        mock.patch.object(
+            conn._loop,
+            "create_connection",
+            autospec=True,
+            spec_set=True,
+            side_effect=ValueError,
+        ),
+        pytest.raises(ValueError),
+    ):
+        await conn.connect(req, [], ClientTimeout())
+
+    assert start_connection.return_value.close.called
+
+    await conn.close()
+
+
 async def test_tcp_connector_server_hostname_default(
     loop: asyncio.AbstractEventLoop, start_connection: mock.AsyncMock
 ) -> None:
diff --git a/tests/test_proxy.py b/tests/test_proxy.py
index 2aaafedeb5c..2359f5bf17b 100644
--- a/tests/test_proxy.py
+++ b/tests/test_proxy.py
@@ -216,6 +216,7 @@ async def make_conn() -> aiohttp.TCPConnector:
         "aiohttp.connector.aiohappyeyeballs.start_connection",
         autospec=True,
         spec_set=True,
+        return_value=mock.create_autospec(socket.socket, spec_set=True, instance=True),
     )
     def test_proxy_connection_error(self, start_connection: mock.Mock) -> None:
         async def make_conn() -> aiohttp.TCPConnector: