Read subprocess stderr with a blocking loop, not readabilityHandler

`readabilityHandler` installs a libdispatch read source on the stderr
pipe's file descriptor. Tearing that source down races epoll event
delivery on Linux: `_dispatch_event_loop_drain` can dereference a
`dispatch_muxnote` that a concurrent source teardown has already freed,
crashing the process. This was observed intermittently in Linux CI while
tearing down clangd connections and confirmed by core-dump analysis.

Forward stderr with a blocking read loop on a dedicated queue instead. A
blocking read never registers the file descriptor with libdispatch's
event loop, so no read source is created for the pipe.

Read the raw file descriptor rather than `FileHandle.read(upToCount:)`:
on non-Darwin platforms the latter blocks until it accumulates the full
requested count or reaches EOF, which would withhold small,
intermittently emitted messages until the subprocess exits. A raw
`read` returns as soon as any bytes are available.

Extract the loop into `readInBackground` and add tests covering
forwarding through end-of-file and prompt delivery of data before EOF.

Author: rintaro

Sources/LanguageServerProtocolTransport/JSONRPCConnection.swift

@@ -16,8 +16,16 @@ public import LanguageServerProtocol
 import Synchronization
 @_spi(SourceKitLSP) import ToolsProtocolsSwiftExtensions
 
-#if canImport(Android)
+#if canImport(Darwin)
+import Darwin
+#elseif canImport(Glibc)
+import Glibc
+#elseif canImport(Musl)
+import Musl
+#elseif canImport(Android)
 import Android
+#elseif canImport(WinSDK)
+import WinSDK
 #endif
 
 #if canImport(CDispatch)
@@ -187,13 +195,11 @@ public final class JSONRPCConnection: Connection {
       Logger(subsystem: LoggingScope.subsystem, category: stderrLoggingCategory).info("\($0)")
     }
     let stderrHandler = Pipe()
-    stderrHandler.fileHandleForReading.readabilityHandler = { fileHandle in
-      let newData = fileHandle.availableData
-      if newData.count == 0 {
-        stderrHandler.fileHandleForReading.readabilityHandler = nil
-      } else {
-        logForwarder.handleDataFromPipe(newData)
-      }
+    Self.readInBackground(
+      fileHandle: stderrHandler.fileHandleForReading,
+      queueLabel: "\(name)-stderr-read-queue"
+    ) { data in
+      logForwarder.handleDataFromPipe(data)
     }
     process.standardError = stderrHandler
     process.terminationHandler = { process in
@@ -222,6 +228,54 @@ public final class JSONRPCConnection: Connection {
   }
   #endif  // os(macOS) || !canImport(Darwin)
 
+  /// Read from `fileHandle` on a dedicated background queue with a blocking loop, forwarding each chunk to
+  /// `receiveData` until end-of-file, then invoke `reachedEndOfFile`.
+  ///
+  /// A blocking read is used rather than a `readabilityHandler`. `readabilityHandler` installs a libdispatch
+  /// read source on the file descriptor; tearing that source down races epoll event delivery on Linux and
+  /// can crash the process in `_dispatch_event_loop_drain`. A blocking read never registers the descriptor
+  /// with libdispatch.
+  ///
+  /// The raw file descriptor is read instead of `FileHandle.read(upToCount:)`: on non-Darwin platforms the
+  /// latter blocks until it accumulates the full requested count or reaches EOF, which would withhold small,
+  /// intermittently emitted output until the writer closes the pipe. A raw `read` returns as soon as any
+  /// bytes are available.
+  @_spi(Testing)
+  public static func readInBackground(
+    fileHandle: FileHandle,
+    queueLabel: String,
+    receiveData: @Sendable @escaping (Data) -> Void,
+    reachedEndOfFile: (@Sendable () -> Void)? = nil
+  ) {
+    let fileDescriptor = fileHandle.fileDescriptor
+    let queue = DispatchQueue(label: queueLabel)
+    queue.async {
+      var buffer = [UInt8](repeating: 0, count: 8 * 1024)
+      while true {
+        let bytesRead = buffer.withUnsafeMutableBytes { ptr -> Int in
+          #if os(Windows)
+          return Int(_read(fileDescriptor, ptr.baseAddress, UInt32(ptr.count)))
+          #else
+          return read(fileDescriptor, ptr.baseAddress, ptr.count)
+          #endif
+        }
+        if bytesRead > 0 {
+          receiveData(Data(buffer[0..<bytesRead]))
+        } else if bytesRead < 0 && errno == EINTR {
+          // Interrupted by a signal; retry the read.
+          continue
+        } else {
+          // Reached end-of-file (0) or hit an unrecoverable error; stop reading.
+          break
+        }
+      }
+      // Keep the handle alive for the duration of the loop so its file descriptor is not closed while
+      // `read` is blocked on it.
+      withExtendedLifetime(fileHandle) {}
+      reachedEndOfFile?()
+    }
+  }
+
   deinit {
     assert(state == .closed)
   }

Tests/LanguageServerProtocolTransportTests/PipeForwardingTests.swift

@@ -0,0 +1,72 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the Swift.org open source project
+//
+// Copyright (c) 2014 - 2025 Apple Inc. and the Swift project authors
+// Licensed under Apache License v2.0 with Runtime Library Exception
+//
+// See https://swift.org/LICENSE.txt for license information
+// See https://swift.org/CONTRIBUTORS.txt for the list of Swift project authors
+//
+//===----------------------------------------------------------------------===//
+
+@_spi(Testing) import LanguageServerProtocolTransport
+import Synchronization
+import ToolsProtocolsTestSupport
+import XCTest
+
+import class Foundation.Pipe
+import struct Foundation.Data
+
+final class PipeForwardingTests: XCTestCase {
+  /// `readInBackground` forwards everything written to the pipe and signals end-of-file once the write end
+  /// is closed.
+  func testReadInBackgroundForwardsAllDataAndSignalsEndOfFile() async throws {
+    let pipe = Pipe()
+    let received = Mutex<Data>(Data())
+    let reachedEndOfFile = expectation(description: "reached end of file")
+
+    JSONRPCConnection.readInBackground(
+      fileHandle: pipe.fileHandleForReading,
+      queueLabel: "test-read-in-background"
+    ) { data in
+      received.withLock { $0.append(data) }
+    } reachedEndOfFile: {
+      reachedEndOfFile.fulfill()
+    }
+
+    try pipe.fileHandleForWriting.write(contentsOf: Data("hello".utf8))
+    try pipe.fileHandleForWriting.write(contentsOf: Data(" world".utf8))
+    try pipe.fileHandleForWriting.close()
+
+    try await fulfillmentOfOrThrow(reachedEndOfFile)
+    XCTAssertEqual(received.withLock { $0 }, Data("hello world".utf8))
+  }
+
+  /// `readInBackground` delivers data as soon as it is available, before the pipe reaches end-of-file. A
+  /// read that accumulated until its buffer filled or EOF (as `FileHandle.read(upToCount:)` does on
+  /// non-Darwin platforms) would withhold this small chunk until the write end is closed.
+  func testReadInBackgroundDeliversDataBeforeEndOfFile() async throws {
+    let pipe = Pipe()
+    let received = Mutex<Data>(Data())
+    let receivedChunk = expectation(description: "received chunk before EOF")
+    receivedChunk.assertForOverFulfill = false
+
+    JSONRPCConnection.readInBackground(
+      fileHandle: pipe.fileHandleForReading,
+      queueLabel: "test-read-in-background-prompt"
+    ) { data in
+      let total = received.withLock { $0.append(data); return $0 }
+      if total == Data("ping".utf8) {
+        receivedChunk.fulfill()
+      }
+    }
+
+    // Do not close the write end: the chunk must be delivered without relying on EOF.
+    try pipe.fileHandleForWriting.write(contentsOf: Data("ping".utf8))
+    try await fulfillmentOfOrThrow(receivedChunk)
+
+    // Let the read loop terminate so its background thread does not outlive the test.
+    try pipe.fileHandleForWriting.close()
+  }
+}