fix(windows): preserve data-path backpressure

Sources/NIOPosix/BSDSocketAPIWindows.swift

@@ -189,6 +189,50 @@ extension Shutdown {
 
 // MARK: _BSDSocketProtocol implementation
 extension NIOBSDSocket {
+    /// Runs a Winsock data-path call and converts its error reporting into an
+    /// `IOResult`, mirroring what `Posix.syscall` does for errno on POSIX
+    /// platforms. A `WSAEWOULDBLOCK` failure becomes `.wouldBlock(0)` and any
+    /// other failure throws an `IOError`.
+    ///
+    /// Unlike the POSIX variant there is no `EINTR` retry loop. `WSAEINTR` only
+    /// arises from the legacy `WSACancelBlockingCall`, which never applies to
+    /// the nonblocking sockets used here.
+    @inline(never)
+    static func winsockSyscall<Value: FixedWidthInteger>(
+        where function: String = #function,
+        _ body: () -> Value
+    ) throws -> IOResult<Value> {
+        let result = body()
+        if result == -1 {  // SOCKET_ERROR
+            let error = WSAGetLastError()
+            if error == WSAEWOULDBLOCK {
+                return .wouldBlock(0)
+            }
+            throw IOError(winsock: error, reason: function)
+        }
+        return .processed(result)
+    }
+
+    /// Variant for Winsock calls that report the transferred count through a
+    /// `DWORD` out-parameter. The wrapper owns that slot, hands it to `body`,
+    /// and reads it only after the call succeeds. A failing call may leave any
+    /// value in the slot, and it is never read.
+    @inline(never)
+    static func winsockSyscall<Value: FixedWidthInteger>(
+        where function: String = #function,
+        _ body: (_ transferred: inout DWORD) -> CInt
+    ) throws -> IOResult<Value> {
+        var transferred: DWORD = 0
+        if body(&transferred) == SOCKET_ERROR {
+            let error = WSAGetLastError()
+            if error == WSAEWOULDBLOCK {
+                return .wouldBlock(0)
+            }
+            throw IOError(winsock: error, reason: function)
+        }
+        return .processed(Value(transferred))
+    }
+
     @inline(never)
     static func accept(
         socket s: NIOBSDSocket.Handle,
@@ -292,11 +336,9 @@ extension NIOBSDSocket {
         buffer buf: UnsafeMutableRawPointer,
         length len: size_t
     ) throws -> IOResult<size_t> {
-        let iResult: CInt = CNIOWindows_recv(s, buf, CInt(len), 0)
-        if iResult == SOCKET_ERROR {
-            throw IOError(winsock: WSAGetLastError(), reason: "recv")
-        }
-        return .processed(size_t(iResult))
+        try winsockSyscall {
+            CNIOWindows_recv(s, buf, CInt(len), 0)
+        }.map(size_t.init)
     }
 
     @inline(never)
@@ -332,12 +374,10 @@ extension NIOBSDSocket {
             )
         }
 
-        var dwNumberOfBytesRecvd: DWORD = 0
         // FIXME(compnerd) is the socket guaranteed to not be overlapped?
-        if WSARecvMsg(s, lpMsg, &dwNumberOfBytesRecvd, nil, nil) == SOCKET_ERROR {
-            throw IOError(winsock: WSAGetLastError(), reason: "recvmsg")
+        return try winsockSyscall { transferred in
+            WSARecvMsg(s, lpMsg, &transferred, nil, nil)
         }
-        return .processed(size_t(dwNumberOfBytesRecvd))
     }
 
     @inline(never)
@@ -372,19 +412,17 @@ extension NIOBSDSocket {
         }
 
         let lpMsg: LPWSAMSG = UnsafeMutablePointer<WSAMSG>(mutating: lpMsg)
-        var NumberOfBytesSent: DWORD = 0
         // FIXME(compnerd) is the socket guaranteed to not be overlapped?
-        if WSASendMsg(
-            Handle,
-            lpMsg,
-            DWORD(dwFlags),
-            &NumberOfBytesSent,
-            nil,
-            nil
-        ) == SOCKET_ERROR {
-            throw IOError(winsock: WSAGetLastError(), reason: "sendmsg")
+        return try winsockSyscall { transferred in
+            WSASendMsg(
+                Handle,
+                lpMsg,
+                DWORD(dwFlags),
+                &transferred,
+                nil,
+                nil
+            )
         }
-        return .processed(size_t(NumberOfBytesSent))
     }
 
     @inline(never)
@@ -393,25 +431,20 @@ extension NIOBSDSocket {
         buffer buf: UnsafeRawPointer,
         length len: size_t
     ) throws -> IOResult<size_t> {
-        let iResult: CInt = CNIOWindows_send(s, buf, CInt(len), 0)
-        if iResult == SOCKET_ERROR {
-            throw IOError(winsock: WSAGetLastError(), reason: "send")
-        }
-        return .processed(size_t(iResult))
+        try winsockSyscall {
+            CNIOWindows_send(s, buf, CInt(len), 0)
+        }.map(size_t.init)
     }
 
     @inline(never)
     static func writev(
         socket s: NIOBSDSocket.Handle,
         iovecs: UnsafeBufferPointer<IOVector>
     ) throws -> IOResult<Int> {
-        var bytesSent: DWORD = 0
         let ptr = UnsafeMutablePointer(mutating: iovecs.baseAddress)
-        let result = WSASend(s, ptr, UInt32(iovecs.count), &bytesSent, 0, nil, nil)
-        if result == SOCKET_ERROR {
-            throw IOError(winsock: WSAGetLastError(), reason: "WSASend")
+        return try winsockSyscall { transferred in
+            WSASend(s, ptr, UInt32(iovecs.count), &transferred, 0, nil, nil)
         }
-        return .processed(Int(bytesSent))
     }
 
     @inline(never)

Tests/NIOPosixTests/BSDSocketAPIWindowsTests.swift

@@ -0,0 +1,92 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the SwiftNIO open source project
+//
+// Copyright (c) 2026 Apple Inc. and the SwiftNIO project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.txt for the list of SwiftNIO project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+#if os(Windows)
+import NIOCore
+import WinSDK
+import XCTest
+
+@testable import NIOPosix
+
+final class BSDSocketAPIWindowsTests: XCTestCase {
+    func testWinsockSyscallReturnsWouldBlock() throws {
+        let result: IOResult<CInt> = try NIOBSDSocket.winsockSyscall {
+            WSASetLastError(WSAEWOULDBLOCK)
+            return SOCKET_ERROR
+        }
+
+        XCTAssertEqual(.wouldBlock(0), result)
+    }
+
+    func testWinsockSyscallThrowsOtherErrors() {
+        let errorCode = WSAECONNRESET
+        let call: () throws -> IOResult<CInt> = {
+            try NIOBSDSocket.winsockSyscall {
+                WSASetLastError(errorCode)
+                return SOCKET_ERROR
+            }
+        }
+
+        XCTAssertThrowsError(try call()) { error in
+            guard let ioError = error as? IOError else {
+                return XCTFail("Expected IOError, got \(error)")
+            }
+            guard case .winsock(let actualErrorCode) = ioError.error else {
+                return XCTFail("Expected a Winsock error, got \(ioError)")
+            }
+            XCTAssertEqual(errorCode, actualErrorCode)
+        }
+    }
+
+    func testWinsockSyscallWithTransferredCountReturnsProcessed() throws {
+        let result: IOResult<Int> = try NIOBSDSocket.winsockSyscall { transferred in
+            transferred = 42
+            return 0
+        }
+
+        XCTAssertEqual(.processed(42), result)
+    }
+
+    func testWinsockSyscallWithTransferredCountIgnoresGarbageWhenWouldBlock() throws {
+        let result: IOResult<Int> = try NIOBSDSocket.winsockSyscall { transferred in
+            transferred = .max
+            WSASetLastError(WSAEWOULDBLOCK)
+            return SOCKET_ERROR
+        }
+
+        XCTAssertEqual(.wouldBlock(0), result)
+    }
+
+    func testWinsockSyscallWithTransferredCountThrowsOtherErrors() {
+        let errorCode = WSAECONNRESET
+        let call: () throws -> IOResult<Int> = {
+            try NIOBSDSocket.winsockSyscall { transferred in
+                transferred = 42
+                WSASetLastError(errorCode)
+                return SOCKET_ERROR
+            }
+        }
+
+        XCTAssertThrowsError(try call()) { error in
+            guard let ioError = error as? IOError else {
+                return XCTFail("Expected IOError, got \(error)")
+            }
+            guard case .winsock(let actualErrorCode) = ioError.error else {
+                return XCTFail("Expected a Winsock error, got \(ioError)")
+            }
+            XCTAssertEqual(errorCode, actualErrorCode)
+        }
+    }
+}
+#endif

Tests/NIOPosixTests/SocketChannelTest.swift

@@ -74,6 +74,85 @@ final class SocketChannelTest: XCTestCase {
         try futureB.wait()
     }
 
+    func testStreamWritesSurviveSocketBackpressure() throws {
+        /// The channel event loop confines all mutable state.
+        final class AccumulateReadsHandler: ChannelInboundHandler, @unchecked Sendable {
+            typealias InboundIn = ByteBuffer
+
+            private let expectedByteCount: Int
+            private let completionPromise: EventLoopPromise<ByteBuffer>
+            private var received: ByteBuffer!
+            private var completed = false
+
+            init(expectedByteCount: Int, completionPromise: EventLoopPromise<ByteBuffer>) {
+                self.expectedByteCount = expectedByteCount
+                self.completionPromise = completionPromise
+            }
+
+            func handlerAdded(context: ChannelHandlerContext) {
+                self.received = context.channel.allocator.buffer(capacity: self.expectedByteCount)
+            }
+
+            func channelRead(context: ChannelHandlerContext, data: NIOAny) {
+                var buffer = Self.unwrapInboundIn(data)
+                self.received.writeBuffer(&buffer)
+                if !self.completed && self.received.readableBytes >= self.expectedByteCount {
+                    self.completed = true
+                    self.completionPromise.succeed(self.received)
+                }
+            }
+
+            func channelInactive(context: ChannelHandlerContext) {
+                if !self.completed {
+                    self.completed = true
+                    self.completionPromise.fail(ChannelError.eof)
+                }
+                context.fireChannelInactive()
+            }
+        }
+
+        func runTest(receiver: Channel, sender: Channel) throws {
+            let totalByteCount = 4 * 1024 * 1024
+            let chunkSize = 4 * 1024
+            let receivedPromise = receiver.eventLoop.makePromise(of: ByteBuffer.self)
+
+            try receiver.setOption(.autoRead, value: false).wait()
+            try receiver.setOption(.socketOption(.so_rcvbuf), value: 1_024).wait()
+            try sender.setOption(.socketOption(.so_sndbuf), value: 1_024).wait()
+            try receiver.pipeline.addHandler(
+                AccumulateReadsHandler(
+                    expectedByteCount: totalByteCount,
+                    completionPromise: receivedPromise
+                )
+            ).wait()
+
+            var expected = sender.allocator.buffer(capacity: totalByteCount)
+            expected.writeBytes((0..<totalByteCount).map { UInt8(truncatingIfNeeded: $0) })
+
+            // Keeping reads disabled while several megabytes are flushed forces the sender through
+            // kernel backpressure before the peer can drain the socket.
+            for offset in stride(from: 0, to: totalByteCount - chunkSize, by: chunkSize) {
+                sender.write(
+                    expected.getSlice(at: offset, length: chunkSize)!,
+                    promise: nil
+                )
+            }
+            let writeFuture = sender.writeAndFlush(
+                expected.getSlice(at: totalByteCount - chunkSize, length: chunkSize)!
+            )
+            let writeRemainedPending = try sender.eventLoop.submit {
+                !writeFuture.isFulfilled
+            }.wait()
+            XCTAssertTrue(writeRemainedPending, "The write completed before the receiver began draining the socket.")
+
+            try receiver.setOption(.autoRead, value: true).wait()
+            try writeFuture.wait()
+            XCTAssertEqual(expected, try receivedPromise.futureResult.wait())
+        }
+
+        try withCrossConnectedTCPChannels(forceSeparateEventLoops: true, runTest)
+    }
+
     public func testDelayedConnectSetsUpRemotePeerAddress() throws {
         let group = MultiThreadedEventLoopGroup(numberOfThreads: 1)
         defer { XCTAssertNoThrow(try group.syncShutdownGracefully()) }