Skip to content

Normalize asyncio writer close races #4217

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
lh0156 wants to merge 5 commits into
base: master
Choose a base branch
from
+97 −5
Open

Normalize asyncio writer close races #4217

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
2132282
Normalize asyncio writer close races
lh0156 Jul 26, 2026
d7dd61c
Preserve packed command TypeErrors
lh0156 Jul 31, 2026
fa2279f
Normalize closed async readers
lh0156 Aug 2, 2026
4588fa1
Handle closed transport write TypeErrors
lh0156 Aug 2, 2026
7c3a840
Preserve write input attribute errors
lh0156 Aug 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 9 additions & 4 deletions redis/asyncio/connection.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1143,8 +1143,14 @@ async def check_health(self):
)

async def _send_packed_command(self, command: Iterable[bytes]) -> None:
self._writer.writelines(command)
await self._writer.drain()
writer = self._writer
if writer is None:
raise ConnectionError("Connection closed while writing")
try:
writer.writelines(command)
await writer.drain()
except AttributeError as error:
Comment thread
petyaslavova marked this conversation as resolved.
Comment thread
petyaslavova marked this conversation as resolved.
raise ConnectionError("Connection closed while writing") from error
Comment thread
cursor[bot] marked this conversation as resolved.
Comment on lines +1152 to +1155

@chatgpt-codex-connector chatgpt-codex-connector Bot Jul 31, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Handle the actual closed-transport TypeError

For the reported Python 3.12 close race, the traceback in #3546 fails inside StreamWriter.writelines() as TypeError: 'NoneType' object is not callable, but this new handler only translates AttributeError; the actual failure still escapes as a raw TypeError and bypasses redis-py's ConnectionError retry path. Please classify that specific closed-transport TypeError without converting arbitrary caller/data TypeErrors.

Useful? React with 👍 / 👎.

Comment thread
petyaslavova marked this conversation as resolved.
Comment on lines +1157 to +1159

@chatgpt-codex-connector chatgpt-codex-connector Bot Aug 2, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Avoid retrying caller TypeErrors as close races

In the final code, this branch still identifies the closed-transport race solely by str(error). If a public send_packed_command() caller passes a malformed iterable that raises the same CPython text while writelines() consumes it (for example by calling a None callback), that caller/data bug is rewritten as ConnectionError, the connection is disconnected, and configured retry logic can replay a non-transient bad command instead of surfacing the original TypeError; please classify the transport race by its source/state rather than the exception message alone.

AGENTS.md reference: AGENTS.md:L120-L123

Useful? React with 👍 / 👎.


async def send_packed_command(
self, command: Union[bytes, str, Iterable[bytes]], check_health: bool = True
Expand All @@ -1164,8 +1170,7 @@ async def send_packed_command(
self._send_packed_command(command), self.socket_timeout
)
else:
self._writer.writelines(command)
await self._writer.drain()
await self._send_packed_command(command)
except asyncio.TimeoutError:
await self.disconnect(nowait=True)
raise TimeoutError("Timeout writing to socket") from None
Expand Down
33 changes: 33 additions & 0 deletions tests/test_asyncio/test_connection.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -83,6 +83,39 @@ def test_connection_default_parser_matches_default_protocol():
assert conn.protocol == 3


@pytest.mark.parametrize("socket_timeout", [None, 1])
async def test_closed_writer_during_write_is_connection_error(socket_timeout):
conn = Connection(socket_timeout=socket_timeout)
conn._reader = mock.Mock()
conn._writer = mock.Mock()
conn._writer.writelines.side_effect = AttributeError(
"'NoneType' object has no attribute 'writelines'"
)
conn.disconnect = mock.AsyncMock()

with pytest.raises(ConnectionError, match="Connection closed while writing"):
await conn.send_packed_command([b"PING\r\n"], check_health=False)

conn.disconnect.assert_awaited_once_with(nowait=True)
conn._reader = None
conn._writer = None


async def test_invalid_packed_command_type_error_is_not_connection_error():
conn = Connection()
conn._reader = mock.Mock()
conn._writer = mock.Mock()
conn._writer.writelines.side_effect = TypeError("invalid packed command")
conn.disconnect = mock.AsyncMock()

with pytest.raises(TypeError, match="invalid packed command"):
await conn.send_packed_command([b"PING\r\n"], check_health=False)

conn.disconnect.assert_awaited_once_with(nowait=True)
conn._reader = None
conn._writer = None


@pytest.mark.parametrize(
("buffer", "eof", "expected"),
[
Expand Down