diff --git a/CHANGES/10941.bugfix.rst b/CHANGES/10941.bugfix.rst
new file mode 120000
index 00000000000..aa085cc590d
--- /dev/null
+++ b/CHANGES/10941.bugfix.rst
@@ -0,0 +1 @@
+10915.bugfix.rst
\ No newline at end of file
diff --git a/CHANGES/10943.bugfix.rst b/CHANGES/10943.bugfix.rst
new file mode 120000
index 00000000000..aa085cc590d
--- /dev/null
+++ b/CHANGES/10943.bugfix.rst
@@ -0,0 +1 @@
+10915.bugfix.rst
\ No newline at end of file
diff --git a/CHANGES/10946.feature.rst b/CHANGES/10946.feature.rst
new file mode 120000
index 00000000000..879a4227358
--- /dev/null
+++ b/CHANGES/10946.feature.rst
@@ -0,0 +1 @@
+10847.feature.rst
\ No newline at end of file
diff --git a/CHANGES/10952.feature.rst b/CHANGES/10952.feature.rst
new file mode 120000
index 00000000000..b565aa68ee0
--- /dev/null
+++ b/CHANGES/10952.feature.rst
@@ -0,0 +1 @@
+9732.feature.rst
\ No newline at end of file
diff --git a/aiohttp/client_middleware_digest_auth.py b/aiohttp/client_middleware_digest_auth.py
index e9eb3ba82e2..b63efaf0142 100644
--- a/aiohttp/client_middleware_digest_auth.py
+++ b/aiohttp/client_middleware_digest_auth.py
@@ -408,8 +408,6 @@ async def __call__(
             # Check if we need to authenticate
             if not self._authenticate(response):
                 break
-            elif retry_count < 1:
-                response.release()  # Release the response to enable connection reuse on retry
 
         # At this point, response is guaranteed to be defined
         assert response is not None
diff --git a/aiohttp/payload.py b/aiohttp/payload.py
index 7339e720fc9..8b16d16aa87 100644
--- a/aiohttp/payload.py
+++ b/aiohttp/payload.py
@@ -810,7 +810,10 @@ async def write_with_length(
 
         try:
             while True:
-                chunk = await self._iter.__anext__()
+                if sys.version_info >= (3, 10):
+                    chunk = await anext(self._iter)
+                else:
+                    chunk = await self._iter.__anext__()
                 if remaining_bytes is None:
                     await writer.write(chunk)
                 # If we have a content length limit
diff --git a/aiohttp/resolver.py b/aiohttp/resolver.py
index 8e30b05d47d..9cdcdb0864b 100644
--- a/aiohttp/resolver.py
+++ b/aiohttp/resolver.py
@@ -160,7 +160,8 @@ async def close(self) -> None:
             self._resolver = None  # type: ignore[assignment] # Clear reference to resolver
             return
         # Otherwise cancel our dedicated resolver
-        self._resolver.cancel()
+        if self._resolver is not None:
+            self._resolver.cancel()
         self._resolver = None  # type: ignore[assignment] # Clear reference
 
 
diff --git a/tests/conftest.py b/tests/conftest.py
index 7e3f85fdd95..17b51266cd7 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -8,7 +8,7 @@
 from hashlib import md5, sha1, sha256
 from pathlib import Path
 from tempfile import TemporaryDirectory
-from typing import Any, Callable, Generator, Iterator
+from typing import Any, AsyncIterator, Callable, Generator, Iterator
 from unittest import mock
 from uuid import uuid4
 
@@ -338,15 +338,13 @@ def parametrize_zlib_backend(
 
 
 @pytest.fixture()
-def cleanup_payload_pending_file_closes(
+async def cleanup_payload_pending_file_closes(
     loop: asyncio.AbstractEventLoop,
-) -> Generator[None, None, None]:
+) -> AsyncIterator[None]:
     """Ensure all pending file close operations complete during test teardown."""
     yield
     if payload._CLOSE_FUTURES:
         # Only wait for futures from the current loop
         loop_futures = [f for f in payload._CLOSE_FUTURES if f.get_loop() is loop]
         if loop_futures:
-            loop.run_until_complete(
-                asyncio.gather(*loop_futures, return_exceptions=True)
-            )
+            await asyncio.gather(*loop_futures, return_exceptions=True)
diff --git a/tests/test_client_functional.py b/tests/test_client_functional.py
index 2123b48ae14..70c9efb98dc 100644
--- a/tests/test_client_functional.py
+++ b/tests/test_client_functional.py
@@ -4382,3 +4382,128 @@ async def handler(request: web.Request) -> web.Response:
             response.raise_for_status()
 
         assert len(client._session.connector._conns) == 1
+
+
+async def test_post_content_exception_connection_kept(
+    aiohttp_client: AiohttpClient,
+) -> None:
+    """Test that connections are kept after content.set_exception() with POST."""
+
+    async def handler(request: web.Request) -> web.Response:
+        await request.read()
+        return web.Response(
+            body=b"x" * 1000
+        )  # Larger response to ensure it's not pre-buffered
+
+    app = web.Application()
+    app.router.add_post("/", handler)
+    client = await aiohttp_client(app)
+
+    # POST request with body - connection should be closed after content exception
+    resp = await client.post("/", data=b"request body")
+
+    with pytest.raises(RuntimeError):
+        async with resp:
+            assert resp.status == 200
+            resp.content.set_exception(RuntimeError("Simulated error"))
+            await resp.read()
+
+    assert resp.closed
+
+    # Wait for any pending operations to complete
+    await resp.wait_for_close()
+
+    assert client._session.connector is not None
+    # Connection is kept because content.set_exception() is a client-side operation
+    # that doesn't affect the underlying connection state
+    assert len(client._session.connector._conns) == 1
+
+
+async def test_network_error_connection_closed(
+    aiohttp_client: AiohttpClient,
+) -> None:
+    """Test that connections are closed after network errors."""
+
+    async def handler(request: web.Request) -> NoReturn:
+        # Read the request body
+        await request.read()
+
+        # Start sending response but close connection before completing
+        response = web.StreamResponse()
+        response.content_length = 1000  # Promise 1000 bytes
+        await response.prepare(request)
+
+        # Send partial data then force close the connection
+        await response.write(b"x" * 100)  # Only send 100 bytes
+        # Force close the transport to simulate network error
+        assert request.transport is not None
+        request.transport.close()
+        assert False, "Will not return"
+
+    app = web.Application()
+    app.router.add_post("/", handler)
+    client = await aiohttp_client(app)
+
+    # POST request that will fail due to network error
+    with pytest.raises(aiohttp.ClientPayloadError):
+        resp = await client.post("/", data=b"request body")
+        async with resp:
+            await resp.read()  # This should fail
+
+    # Give event loop a chance to process connection cleanup
+    await asyncio.sleep(0)
+
+    assert client._session.connector is not None
+    # Connection should be closed due to network error
+    assert len(client._session.connector._conns) == 0
+
+
+async def test_client_side_network_error_connection_closed(
+    aiohttp_client: AiohttpClient,
+) -> None:
+    """Test that connections are closed after client-side network errors."""
+    handler_done = asyncio.Event()
+
+    async def handler(request: web.Request) -> NoReturn:
+        # Read the request body
+        await request.read()
+
+        # Start sending a large response
+        response = web.StreamResponse()
+        response.content_length = 10000  # Promise 10KB
+        await response.prepare(request)
+
+        # Send some data
+        await response.write(b"x" * 1000)
+
+        # Keep the response open - we'll interrupt from client side
+        await asyncio.wait_for(handler_done.wait(), timeout=5.0)
+        assert False, "Will not return"
+
+    app = web.Application()
+    app.router.add_post("/", handler)
+    client = await aiohttp_client(app)
+
+    # POST request that will fail due to client-side network error
+    with pytest.raises(aiohttp.ClientPayloadError):
+        resp = await client.post("/", data=b"request body")
+        async with resp:
+            # Simulate client-side network error by closing the transport
+            # This simulates connection reset, network failure, etc.
+            assert resp.connection is not None
+            assert resp.connection.protocol is not None
+            assert resp.connection.protocol.transport is not None
+            resp.connection.protocol.transport.close()
+
+            # This should fail with connection error
+            await resp.read()
+
+    # Signal handler to finish
+    handler_done.set()
+
+    # Give event loop a chance to process connection cleanup
+    await asyncio.sleep(0)
+
+    assert client._session.connector is not None
+    # Connection should be closed due to client-side network error
+    assert len(client._session.connector._conns) == 0
diff --git a/tests/test_client_middleware.py b/tests/test_client_middleware.py
index 5894795dc21..883d853d2e8 100644
--- a/tests/test_client_middleware.py
+++ b/tests/test_client_middleware.py
@@ -891,7 +891,6 @@ async def __call__(
                 response = await handler(request)
                 if retry_count == 0:
                     retry_count += 1
-                    response.release()  # Release the response to enable connection reuse
                     continue
                 return response
 
diff --git a/tests/test_resolver.py b/tests/test_resolver.py
index 7950f3b0f39..3a9d1a70a23 100644
--- a/tests/test_resolver.py
+++ b/tests/test_resolver.py
@@ -660,3 +660,40 @@ async def test_dns_resolver_manager_missing_loop_data() -> None:
 
         # Verify no exception was raised
         assert loop not in manager._loop_data
+
+
+@pytest.mark.skipif(not getaddrinfo, reason="aiodns >=3.2.0 required")
+@pytest.mark.usefixtures("check_no_lingering_resolvers")
+async def test_async_resolver_close_multiple_times() -> None:
+    """Test that AsyncResolver.close() can be called multiple times without error."""
+    with patch("aiodns.DNSResolver") as mock_dns_resolver:
+        mock_resolver = Mock()
+        mock_resolver.cancel = Mock()
+        mock_dns_resolver.return_value = mock_resolver
+
+        # Create a resolver with custom args (dedicated resolver)
+        resolver = AsyncResolver(nameservers=["8.8.8.8"])
+
+        # Close it once
+        await resolver.close()
+        mock_resolver.cancel.assert_called_once()
+
+        # Close it again - should not raise AttributeError
+        await resolver.close()
+        # cancel should still only be called once
+        mock_resolver.cancel.assert_called_once()
+
+
+@pytest.mark.skipif(not getaddrinfo, reason="aiodns >=3.2.0 required")
+@pytest.mark.usefixtures("check_no_lingering_resolvers")
+async def test_async_resolver_close_with_none_resolver() -> None:
+    """Test that AsyncResolver.close() handles None resolver gracefully."""
+    with patch("aiodns.DNSResolver"):
+        # Create a resolver with custom args (dedicated resolver)
+        resolver = AsyncResolver(nameservers=["8.8.8.8"])
+
+        # Manually set resolver to None to simulate edge case
+        resolver._resolver = None  # type: ignore[assignment]
+
+        # This should not raise AttributeError
+        await resolver.close()
diff --git a/tests/test_web_functional.py b/tests/test_web_functional.py
index 40fddc3aaf0..42c3edf20d1 100644
--- a/tests/test_web_functional.py
+++ b/tests/test_web_functional.py
@@ -1945,6 +1945,9 @@ async def handler(request: web.Request) -> web.Response:
             await resp.read()
     assert resp.closed
 
+    # Wait for any pending operations to complete
+    await resp.wait_for_close()
+
     assert session._connector is not None
     assert len(session._connector._conns) == 1