fix(quic): satisfy mypy 2.2 comparison-overlap in stream read loop (#1374)

Use is_reset() in the QUICStream.read() wait loop so concurrent peer
resets remain detectable without tripping mypy 2.2 comparison-overlap
after the entry state guard.

Fixes #1373

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: acul71

libp2p/transport/quic/stream.py

@@ -251,7 +251,7 @@ async def read(self, n: int | None = None) -> bytes:
                             return data
 
                         # Check if stream was reset (unblocking read loop)
-                        if self._state == StreamState.RESET:
+                        if self.is_reset():
                             raise QUICStreamResetError(
                                 f"Stream {self.stream_id} was reset"
                             )

newsfragments/1373.internal.rst

@@ -0,0 +1 @@
+Fixed CI lint failure on ``main`` caused by mypy 2.2 ``comparison-overlap`` in the QUIC stream read loop. Reads blocked on ``_receive_event`` still raise ``QUICStreamResetError`` when the stream is reset concurrently.

tests/core/transport/quic/test_stream.py

@@ -0,0 +1,75 @@
+"""Unit tests for QUICStream behavior."""
+
+from unittest.mock import Mock
+
+import pytest
+from multiaddr.multiaddr import Multiaddr
+import trio
+from trio.testing import wait_all_tasks_blocked
+
+from libp2p.crypto.ed25519 import create_new_key_pair
+from libp2p.peer.id import ID
+from libp2p.transport.quic.config import QUICTransportConfig
+from libp2p.transport.quic.connection import QUICConnection
+from libp2p.transport.quic.exceptions import QUICStreamResetError
+from libp2p.transport.quic.stream import QUICStream, StreamDirection
+
+
+@pytest.fixture
+def quic_connection() -> QUICConnection:
+    mock_quic = Mock()
+    mock_quic.next_event.return_value = None
+    mock_quic.datagrams_to_send.return_value = []
+    mock_quic.get_timer.return_value = None
+    mock_quic.connect = Mock()
+    mock_quic.close = Mock()
+    mock_quic.send_stream_data = Mock()
+    mock_quic.reset_stream = Mock()
+
+    mock_transport = Mock()
+    mock_transport._config = QUICTransportConfig()
+
+    private_key = create_new_key_pair().private_key
+    peer_id = ID.from_pubkey(private_key.get_public_key())
+
+    return QUICConnection(
+        quic_connection=mock_quic,
+        remote_addr=("127.0.0.1", 4001),
+        remote_peer_id=None,
+        local_peer_id=peer_id,
+        is_initiator=True,
+        maddr=Multiaddr("/ip4/127.0.0.1/udp/4001/quic"),
+        transport=mock_transport,
+        resource_scope=None,
+        security_manager=Mock(),
+    )
+
+
+@pytest.mark.trio
+async def test_read_raises_reset_error_when_reset_during_wait(
+    quic_connection: QUICConnection,
+) -> None:
+    """read() must detect peer reset while blocked on _receive_event."""
+    stream = QUICStream(
+        connection=quic_connection,
+        stream_id=1,
+        direction=StreamDirection.INBOUND,
+        remote_addr=("127.0.0.1", 4001),
+    )
+
+    reset_error: QUICStreamResetError | None = None
+
+    async def read_and_capture() -> None:
+        nonlocal reset_error
+        try:
+            await stream.read()
+        except QUICStreamResetError as exc:
+            reset_error = exc
+
+    async with trio.open_nursery() as nursery:
+        nursery.start_soon(read_and_capture)
+        await wait_all_tasks_blocked()
+        await stream.handle_reset(error_code=1)
+
+    assert reset_error is not None
+    assert "reset" in str(reset_error).lower()