fix a bug on windows

Author: egorsmkv

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "rsloop"
-version = "0.1.25"
+version = "0.1.26"
 edition = "2024"
 description = "An event loop for asyncio written in Rust"
 license = "Apache-2.0"

Cargo.toml

@@ -4,7 +4,7 @@ build-backend = "maturin"
 
 [project]
 name = "rsloop"
-version = "0.1.25"
+version = "0.1.26"
 description = "An event loop for asyncio written in Rust"
 readme = "README.md"
 license = { file = "LICENSE" }

pyproject.toml

@@ -1607,7 +1607,17 @@ impl StreamTransportCore {
         };
         let mut writer = writer.lock().expect("poisoned direct tasked writer");
         match writer.deref_mut() {
-            TaskedDirectWriter::Tcp(stream) => stream.write(data),
+            TaskedDirectWriter::Tcp(stream) => {
+                #[cfg(windows)]
+                // A paused reader does not consume Winsock's asynchronous reset
+                // notification. Surface it before another small write can appear
+                // to succeed from the local send buffer.
+                if let Some(err) = stream.take_error()? {
+                    return Err(err);
+                }
+
+                stream.write(data)
+            }
             #[cfg(unix)]
             TaskedDirectWriter::Unix(stream) => stream.write(data),
         }

src/stream_transport.rs

@@ -7,6 +7,7 @@
 import os
 import signal
 import socket
+import struct
 import sys
 import tempfile
 import time
@@ -222,6 +223,48 @@ def connection_lost(self, exc):
         self.assertEqual(received, b"response-before-eof")
         self.assertEqual(events, ["data", "eof", "lost"])
 
+    @unittest.skipUnless(sys.platform == "win32", "requires Winsock")
+    def test_write_reports_reset_while_reading_is_paused(self) -> None:
+        async def main() -> BaseException | None:
+            loop = asyncio.get_running_loop()
+            connection_lost = loop.create_future()
+
+            class ClientProtocol(asyncio.Protocol):
+                def connection_made(self, transport):
+                    transport.set_write_buffer_limits(0)
+                    transport.pause_reading()
+
+                def connection_lost(self, exc):
+                    if not connection_lost.done():
+                        connection_lost.set_result(exc)
+
+            server_sock = socket.create_server(("127.0.0.1", 0))
+            try:
+                transport, _ = await loop.create_connection(
+                    ClientProtocol, *server_sock.getsockname()[:2]
+                )
+                peer_sock, _ = server_sock.accept()
+                peer_sock.setsockopt(
+                    socket.SOL_SOCKET,
+                    socket.SO_LINGER,
+                    struct.pack("ii", 1, 0),
+                )
+                peer_sock.close()
+                try:
+                    for _ in range(1000):
+                        await asyncio.sleep(0)
+                        transport.write(b"foo")
+                        if connection_lost.done():
+                            break
+
+                    return await asyncio.wait_for(connection_lost, 3.0)
+                finally:
+                    transport.close()
+            finally:
+                server_sock.close()
+
+        self.assertIsNotNone(rsloop.run(main()))
+
     def test_readexactly_larger_than_flow_control_window(self) -> None:
         # Regression test: a readexactly() waiting for more than 2 * limit
         # bytes (default limit 64 KiB) used to deadlock because the reader