client: fix upload timeouts with sock_read set (#7150)

## What do these changes do?

Prevent the `sock_read` timeout callback from firing by only scheduling
it afterthe payload (if any) has been fully written.

## Are there changes in behavior for the user?

No

## Related issue number

Fixes #7149

## Checklist

- [x] I think the code is well written
- [x] Unit tests for the changes exist
- [x] Documentation reflects the changes
- [x] If you provide code modification, please add yourself to
`CONTRIBUTORS.txt`
  * The format is <Name> <Surname>.
  * Please keep alphabetical order, the file is sorted by names.
- [x] Add a new news fragment into the `CHANGES` folder
  * name it `<issue_id>.<type>` for example (588.bugfix)
* if you don't have an `issue_id` change it to the pr id after creating
the pr
  * ensure type is one of the following:
    * `.feature`: Signifying a new feature.
    * `.bugfix`: Signifying a bug fix.
    * `.doc`: Signifying a documentation improvement.
    * `.removal`: Signifying a deprecation or removal of public API.
* `.misc`: A ticket has been closed, but it is not of interest to users.
* Make sure to use full sentences with correct case and punctuation, for
example: "Fix issue with non-ascii contents in doctest text files."

---------

Co-authored-by: Sam Bull <[email protected]>

CHANGES/7149.bugfix

@@ -0,0 +1 @@
+changed ``sock_read`` timeout to start after writing has finished, to avoid read timeouts caused by an unfinished write. -- by :user:`dtrifiro`

aiohttp/client_proto.py

@@ -157,7 +157,6 @@ def set_response_params(
         self._skip_payload = skip_payload
 
         self._read_timeout = read_timeout
-        self._reschedule_timeout()
 
         self._timeout_ceil_threshold = timeout_ceil_threshold
 
@@ -193,6 +192,9 @@ def _reschedule_timeout(self) -> None:
         else:
             self._read_timeout_handle = None
 
+    def start_timeout(self) -> None:
+        self._reschedule_timeout()
+
     def _on_read_timeout(self) -> None:
         exc = ServerTimeoutError("Timeout on reading data from socket")
         self.set_exception(exc)

aiohttp/client_reqrep.py

@@ -551,6 +551,8 @@ async def write_bytes(
                 protocol.set_exception(exc)
         except Exception as exc:
             protocol.set_exception(exc)
+        else:
+            protocol.start_timeout()
         finally:
             self._writer = None
 

tests/test_client_functional.py

@@ -9,7 +9,7 @@
 import pathlib
 import socket
 import ssl
-from typing import Any
+from typing import Any, AsyncIterator
 from unittest import mock
 
 import pytest
@@ -673,6 +673,25 @@ async def handler(request):
             await resp.content.read()
 
 
+async def test_read_timeout_on_write(aiohttp_client: Any) -> None:
+    async def gen_payload() -> AsyncIterator[str]:
+        # Delay writing to ensure read timeout isn't triggered before writing completes.
+        await asyncio.sleep(0.5)
+        yield b"foo"
+
+    async def handler(request: web.Request) -> web.Response:
+        return web.Response(body=await request.read())
+
+    app = web.Application()
+    app.router.add_put("/", handler)
+
+    timeout = aiohttp.ClientTimeout(total=None, sock_read=0.1)
+    client = await aiohttp_client(app)
+    async with client.put("/", data=gen_payload(), timeout=timeout) as resp:
+        result = await resp.read()  # Should not trigger a read timeout.
+    assert result == b"foo"
+
+
 async def test_timeout_on_reading_data(aiohttp_client: Any, mocker: Any) -> None:
     loop = asyncio.get_event_loop()
 

tests/test_client_proto.py

@@ -109,12 +109,15 @@ async def test_empty_data(loop: Any) -> None:
 async def test_schedule_timeout(loop: Any) -> None:
     proto = ResponseHandler(loop=loop)
     proto.set_response_params(read_timeout=1)
+    assert proto._read_timeout_handle is None
+    proto.start_timeout()
     assert proto._read_timeout_handle is not None
 
 
 async def test_drop_timeout(loop: Any) -> None:
     proto = ResponseHandler(loop=loop)
     proto.set_response_params(read_timeout=1)
+    proto.start_timeout()
     assert proto._read_timeout_handle is not None
     proto._drop_timeout()
     assert proto._read_timeout_handle is None
@@ -123,6 +126,7 @@ async def test_drop_timeout(loop: Any) -> None:
 async def test_reschedule_timeout(loop: Any) -> None:
     proto = ResponseHandler(loop=loop)
     proto.set_response_params(read_timeout=1)
+    proto.start_timeout()
     assert proto._read_timeout_handle is not None
     h = proto._read_timeout_handle
     proto._reschedule_timeout()
@@ -133,6 +137,7 @@ async def test_reschedule_timeout(loop: Any) -> None:
 async def test_eof_received(loop: Any) -> None:
     proto = ResponseHandler(loop=loop)
     proto.set_response_params(read_timeout=1)
+    proto.start_timeout()
     assert proto._read_timeout_handle is not None
     proto.eof_received()
     assert proto._read_timeout_handle is None