Fix a problem with connection waiters that are never awaited

CHANGES/4562.bugfix

@@ -0,0 +1 @@
+Fix a problem with connection waiters that are never awaited.

CONTRIBUTORS.txt

@@ -130,6 +130,7 @@ Igor Alexandrov
 Igor Davydenko
 Igor Mozharovsky
 Igor Pavlov
+Illia Volochii
 Ilya Chichak
 Ilya Gruzinov
 Ingmar Steen

aiohttp/connector.py

@@ -453,8 +453,7 @@ async def connect(self, req: 'ClientRequest',
             fut = self._loop.create_future()
 
             # This connection will now count towards the limit.
-            waiters = self._waiters[key]
-            waiters.append(fut)
+            self._waiters[key].append(fut)
 
             if traces:
                 for trace in traces:
@@ -463,21 +462,18 @@ async def connect(self, req: 'ClientRequest',
             try:
                 await fut
             except BaseException as e:
-                # remove a waiter even if it was cancelled, normally it's
-                #  removed when it's notified
-                try:
-                    waiters.remove(fut)
-                except ValueError:  # fut may no longer be in list
-                    pass
+                if key in self._waiters:
+                    # remove a waiter even if it was cancelled, normally it's
+                    #  removed when it's notified
+                    try:
+                        self._waiters[key].remove(fut)
+                    except ValueError:  # fut may no longer be in list
+                        pass
 
                 raise e
             finally:
-                if not waiters:
-                    try:
-                        del self._waiters[key]
-                    except KeyError:
-                        # the key was evicted before.
-                        pass
+                if key in self._waiters and not self._waiters[key]:
+                    del self._waiters[key]
 
             if traces:
                 for trace in traces:

tests/test_connector.py

@@ -2367,3 +2367,42 @@ async def send_dns_cache_hit(self, *args, **kwargs):
     connector._throttle_dns_events[key] = EventResultOrError(loop)
     traces = [DummyTracer()]
     assert await connector._resolve_host("", 0, traces) == [token]
+
+
+async def test_connector_does_not_remove_needed_waiters(loop, key) -> None:
+    proto = create_mocked_conn(loop)
+    proto.is_connected.return_value = True
+
+    req = ClientRequest('GET', URL('https://localhost:80'), loop=loop)
+    connection_key = req.connection_key
+
+    connector = aiohttp.BaseConnector()
+    connector._available_connections = mock.Mock(return_value=0)
+    connector._conns[key] = [(proto, loop.time())]
+    connector._create_connection = create_mocked_conn(loop)
+    connector._create_connection.return_value = loop.create_future()
+    connector._create_connection.return_value.set_result(proto)
+
+    dummy_waiter = loop.create_future()
+
+    async def await_connection_and_check_waiters():
+        connection = await connector.connect(req, [], ClientTimeout())
+        try:
+            assert connection_key in connector._waiters
+            assert dummy_waiter in connector._waiters[connection_key]
+        finally:
+            connection.close()
+
+    async def allow_connection_and_add_dummy_waiter():
+        # `asyncio.gather` may execute coroutines not in order.
+        # Skip one event loop run cycle in such a case.
+        if connection_key not in connector._waiters:
+            await asyncio.sleep(0)
+        connector._waiters[connection_key].popleft().set_result(None)
+        del connector._waiters[connection_key]
+        connector._waiters[connection_key].append(dummy_waiter)
+
+    await asyncio.gather(
+        await_connection_and_check_waiters(),
+        allow_connection_and_add_dummy_waiter(),
+    )